带 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 上更新