帶 Agent 的非同步應用程式

本指南演示瞭如何使用 Starlette 將 smolagents 庫中的同步 Agent 整合到非同步 Python Web 應用程式中。該示例旨在幫助剛接觸非同步 Python 和 Agent 整合的使用者瞭解將同步 Agent 邏輯與非同步 Web 伺服器相結合的最佳實踐。

概述

Starlette：一個輕量級的 ASGI 框架，用於在 Python 中構建非同步 Web 應用程式。
anyio.to_thread.run_sync：用於在後臺執行緒中執行阻塞（同步）程式碼的工具，以防止其阻塞非同步事件迴圈。
CodeAgent：來自 smolagents 庫的一個 Agent，能夠以程式設計方式解決任務。

為何要使用後臺執行緒？

CodeAgent.run() 同步執行 Python 程式碼。如果直接在非同步端點中呼叫它，將會阻塞 Starlette 的事件迴圈，從而降低效能和可伸縮性。透過使用 anyio.to_thread.run_sync 將此操作轉移到後臺執行緒，即使在高併發下，你也能保持應用的響應性和高效性。

示例工作流

Starlette 應用公開了一個 /run-agent 端點，該端點接受一個包含 task 字串的 JSON 負載。
當收到請求時，Agent 將使用 anyio.to_thread.run_sync 在後臺執行緒中執行。
結果將以 JSON 響應的形式返回。

使用 CodeAgent 構建 Starlette 應用

1. 安裝依賴

pip install smolagents starlette anyio uvicorn

2. 應用程式程式碼 ( main.py )

import anyio.to_thread
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from smolagents import CodeAgent, InferenceClientModel

agent = CodeAgent(
    model=InferenceClientModel(model_id="Qwen/Qwen2.5-Coder-32B-Instruct"),
    tools=[],
)

async def run_agent(request: Request):
    data = await request.json()
    task = data.get("task", "")
    # Run the agent synchronously in a background thread
    result = await anyio.to_thread.run_sync(agent.run, task)
    return JSONResponse({"result": result})

app = Starlette(routes=[
    Route("/run-agent", run_agent, methods=["POST"]),
])

3. 執行應用

uvicorn async_agent.main:app --reload

4. 測試端點

curl -X POST https://:8000/run-agent -H 'Content-Type: application/json' -d '{"task": "What is 2+2?"}'

預期響應

{"result": "4"}

延伸閱讀

完整程式碼請參見 examples/async_agent。

< > 在 GitHub 上更新