在 Starlette 异步 Web 应用中集成 smolagents CodeAgent用 anyio.to_thread 让同步 Agent 不再阻塞事件循环【免费下载链接】smolagents smolagents: a barebones library for agents that think in code.项目地址: https://gitcode.com/gh_mirrors/smo/smolagents本指南以仓库中的 async_agent 官方示例对应文档 docs/source/ko/examples/async_agent.md为核心讲解如何把 smolagents 的CodeAgent集成进 Starlette 异步 Web 应用借助anyio.to_thread.run_sync把同步的 Agent 执行任务放到后台线程避免阻塞 ASGI 事件循环。读完你将掌握一套可直接复制的异步 Agent 服务搭建方案包括依赖安装、应用代码、运行与测试方法以及CodeAgent.run()与InferenceClientModel的源码级原理。核心概念一览在动手写代码前先明确三个关键角色StarlettePython 中构建异步 Web 应用的轻量级 ASGI 框架。它运行在单一事件循环上所有协程共享同一个线程因此任何耗时的同步操作都会拖住整个服务。anyio.to_thread.run_syncanyio 提供的实用工具可以把一段阻塞同步代码放到独立的后台线程中执行而事件循环可以继续处理其他请求不会被卡住。CodeAgentsmolagents 库中的核心 Agent 类型位于 src/smolagents/agents.py。它的特点是用代码思考——LLM 以 Python 代码的形式生成工具调用然后由解析器执行。关于 Agent 的基础用法可以参考 guided_tour 与 agents 参考文档。为什么必须用后台线程——源码给出的答案CodeAgent.run()是一个同步阻塞方法。查看 src/smolagents/agents.py 中run()的签名def run( self, task: str, stream: bool False, reset: bool True, images: list[PIL.Image.Image] | None None, additional_args: dict | None None, max_steps: int | None None, return_full_result: bool | None None, ) - Any | RunResult:在非流式模式下它会调用self._run_stream(...)并list(...)一口气执行完所有步骤见 agents.pyLLM 生成代码 → 解析 → Python 执行器运行 → 再生成下一轮直到产出FinalAnswerStep为止。这个过程完全发生在调用它的线程里。如果在 Starlette 的异步端点中直接调用agent.run(task)整个事件循环都会被这个同步调用占据期间所有其他请求都无法被调度应用在高并发下会迅速失去响应。正确做法是把这一同步调用委托给anyio.to_thread.run_sync让 Agent 在后台线程中运行事件循环则保持空闲、持续处理新连接。补充smolagents 生态中类似的线程与事件循环配合模式并不少见。例如 src/smolagents/tools.py 中的ToolCollection.from_mcp就注明会spawn a separate thread to run an asyncio event loop来对接 MCP 服务器——可见把同步/异步边界划清楚是整个框架的基本设计思路。示例工作流整体流程非常简洁Starlette 应用暴露一个/run-agent端点接收包含task字符串的 JSON 载荷请求到达后通过anyio.to_thread.run_sync在后台线程中运行 AgentAgent 执行完毕结果以 JSON 响应返回给客户端。完整实战构建一个异步 Agent 服务第 1 步安装依赖pip install smolagents starlette anyio uvicorn这四个依赖缺一不可对应 examples/async_agent/requirements.txtsmolagents提供CodeAgent与InferenceClientModelstarlette提供Starlette应用、Route路由与JSONResponseanyio提供to_thread.run_sync后台线程工具Starlette 本身也依赖 anyio这里显式引入以便直接使用uvicornASGI 服务器负责托管并运行 Starlette 应用。第 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( modelInferenceClientModel(model_idQwen/Qwen3-Next-80B-A3B-Thinking), 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]), ])仓库中的 examples/async_agent/main.py 在此基础上做了增强把 Agent 的创建、线程执行、错误处理都拆成了独立函数更贴近生产使用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 # Create a simple agent instance (customize as needed) def get_agent(): # You can set custom model, or tools as needed return CodeAgent( modelInferenceClientModel(model_idQwen/Qwen3-Next-80B-A3B-Thinking), tools[], ) async def run_agent_in_thread(task: str): agent get_agent() # The agents run method is synchronous result await anyio.to_thread.run_sync(agent.run, task) return result async def run_agent_endpoint(request: Request): data await request.json() task data.get(task) if not task: return JSONResponse({error: Missing task in request body.}, status_code400) try: result await run_agent_in_thread(task) return JSONResponse({result: result}) except Exception as e: return JSONResponse({error: str(e)}, status_code500) routes [ Route(/run-agent, run_agent_endpoint, methods[POST]), ] app Starlette(debugTrue, routesroutes)增强版多出的两点值得借鉴参数校验task缺失时返回400 Bad Request而不是静默地用空字符串执行异常兜底Agent 执行失败模型超时、代码执行报错等时返回500并携带错误信息避免未捕获异常导致连接被直接断开。第 3 步运行应用uvicorn async_agent.main:app --reload说明async_agent.main表示模块async_agent/main.py中的app对象--reload开启热重载开发调试时修改代码会自动重启服务。服务默认监听http://localhost:8000。第 4 步测试端点curl -X POST http://localhost:8000/run-agent -H Content-Type: application/json -d {task: What is 22?}预期响应{result: 4}Agent 接收到任务后在后台线程中完成推理 → 写代码 → 执行 → 输出的完整闭环最终把答案包装成 JSON 返回。源码级深入模型与 Agent 的底层细节InferenceClientModel走 Hugging Face Inference Providers 的模型封装示例中InferenceClientModel(model_idQwen/Qwen3-Next-80B-A3B-Thinking)的完整定义位于 src/smolagents/models.py。它通过 Hugging Face Inference Providers 与模型交互支持 Cerebras、Cohere、Fal、Fireworks、HF-Inference、Hyperbolic、Nebius、Novita、Replicate、SambaNova、Together 等多个服务商。常用参数如下见 models.py参数默认值说明model_idQwen/Qwen3-Next-80B-A3B-ThinkingHugging Face 模型 ID或已部署 Inference Endpoint 的 URLproviderNone即 auto推理服务商名称为 auto 时按用户配置的服务商顺序自动选择传入base_url时该参数不生效tokenNone认证令牌。未提供时依次尝试环境变量HF_TOKEN与 HF CLI 配置中保存的 tokengated 模型还需相应的读权限api_keyNonetoken的别名用于对齐 OpenAI 客户端风格两者不能同时设置timeout120单次 API 请求超时时间秒base_urlNone自定义推理 URL与model二选一bill_toNone企业版 Hub 组织计费账户client_kwargsNone透传给底层InferenceClient的额外关键字参数engine InferenceClientModel( model_idQwen/Qwen3-Next-80B-A3B-Thinking, providerhyperbolic, tokenyour_hf_token_here, max_tokens5000, )值得注意的是token未显式传入时会读取HF_TOKEN环境变量这意味着你在本地先执行export HF_TOKENhf_xxx即可免去在代码里硬编码凭据的麻烦。更多模型用法见 models 参考文档。CodeAgent.run()同步执行的完整语义run()除了我们关心的同步阻塞这一属性外还支持若干实用参数见 agents.pystream设为True时返回一个生成器逐步骤产出执行过程ActionStep、PlanningStep、FinalAnswerStep等适合做流式输出或进度展示默认False全部执行完后只返回最终答案reset是否在本次运行前清空会话记忆默认True若设为False则可基于上一次对话继续max_steps限制 Agent 最多执行的步数防止无限循环additional_args向 Agent 注入额外变量如 DataFrame、图片Agent 可以直接以变量名访问return_full_result为True时返回包含output、token_usage、steps、timing、state的RunResult对象见 agents.py便于统计 token 消耗与排查问题。在异步场景下这些参数同样可以通过run_sync透传例如result await anyio.to_thread.run_sync(agent.run, task, True, True, None, None, 10, True)生产环境进阶建议把示例跑通之后以下几点能让服务更稳健线程池边界anyio.to_thread.run_sync使用 anyio 的默认线程池上限约为 40 个线程。如果 Agent 任务耗时很长一次run()可能几十秒并发请求数接近线程池上限时新的调用会排队等待必要时可为不同的业务配置独立的上限或对端点做并发限流。超时控制InferenceClientModel的timeout只约束单次模型 API 调用而一次agent.run()包含多轮 LLM 调用加代码执行。若需对整体请求设置超时可在端点层用asyncio.wait_for包裹或借助max_steps限制推理轮数。Agent 实例复用策略示例中的get_agent()每次请求都新建 AgentCodeAgent内部有 Python 执行器与记忆状态新建实例天然隔离了请求间状态。若追求更低延迟可以复用实例但要注意run(resetTrue)的默认行为避免任务间的状态串扰。换用更快的模型默认的Qwen/Qwen3-Next-80B-A3B-Thinking是推理型大模型回答质量高但延迟较大。对延迟敏感的场景可换用更小的模型或搭配provider参数选择延迟更低的推理服务商。错误分类处理参考增强版示例把缺少参数400与Agent 执行失败500分开处理前端才能给出准确的用户提示。继续深入完整的可运行示例代码见 examples/async_agent/main.py配套说明见 examples/async_agent/README.md想了解 Agent 的流式执行、记忆管理与更多参数可阅读 agents 参考文档 与 models 参考文档该主题的英文版官方文档位于 docs/source/en/examples/async_agent.md韩文版即本文所依据的 docs/source/ko/examples/async_agent.md。【免费下载链接】smolagents smolagents: a barebones library for agents that think in code.项目地址: https://gitcode.com/gh_mirrors/smo/smolagents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考