AI Agent落地实操:手写调度器+HTTP工具链+SQLite记忆
1. 项目概述这不是写代码是给AI装上“手脚”和“脑子”“AI Agent 搭建实操指南”——这八个字最近在技术圈刷屏但很多人点进去发现全是概念图、架构图、PPT截图或者直接跳转到某个闭源平台的注册页。我去年带三个团队落地了从客服智能体、研发辅助Agent到合规审查助手的七个项目踩过所有能踩的坑模型调用超时卡死在“思考中”、工具链权限错配导致API反复403、记忆模块把用户昨天问的“报销流程”和前天问的“年假天数”混成一团、甚至出现Agent自己调用删除生产数据库的插件……这些根本不是理论问题是螺丝没拧紧、线没接对、保险丝没装上的实操问题。所谓Agent本质就是让大模型不再只当个“嘴强王者”而是能主动查文档、调接口、读文件、写代码、发邮件、点按钮——它得有手、有脚、有记性、有判断力还得知道什么时候该停手。这篇指南不讲LLM原理不画四层抽象架构图就拆解你明天上午就能在自己笔记本上跑起来的完整链路从零选型、环境初始化、工具注册、记忆配置、执行编排到真正在终端里输入“帮我查下Q3销售Top3客户生成对比表格发邮件”然后看着它一步步打开CRM API、拉数据、调用pandas处理、用matplotlib画图、调Outlook SMTP发附件。适合两类人一类是刚学完LangChain文档但连第一个Tool都注册失败的开发者另一类是业务方负责人想搞清“我们采购的Agent平台到底在后台干了什么”避免被厂商话术绕晕。全文所有命令、配置、参数值均来自我本地实测环境Mac M2 Pro Ubuntu 22.04 Python 3.11拒绝“理论上可行”。2. 核心设计思路为什么放弃LangGraph、AutoGen坚持手写Executor市面上90%的Agent教程一上来就推LangGraph或AutoGen理由很光鲜“支持复杂状态机”、“内置循环控制”、“社区生态成熟”。但我在真实项目里发现这些框架的“成熟”恰恰是落地的最大陷阱。举个最典型的例子某金融客户要求Agent必须严格遵循“先查监管条例→再比对合同条款→最后生成风险提示”的三步顺序且每步失败必须原路返回上一步重试。LangGraph的StateGraph看似完美可一旦在第二步调用外部法规API时网络抖动超时整个state会卡在“pending”状态而它的retry机制默认只重试当前节点不会回滚到第一步重新加载最新版条例——结果Agent拿着过期的监管条文继续往下走输出全错。AutoGen更麻烦它的GroupChatManager底层依赖WebSocket长连接在企业内网防火墙策略下频繁断连重连后上下文全丢。我最终选择放弃所有高级框架用Python原生concurrent.futures.ThreadPoolExecutor自定义状态机实现核心调度器原因很实在可控性每个步骤的输入/输出、超时阈值、重试次数、错误降级策略全部显式声明。比如查CRM数据这步我硬编码timeout8.5秒因为监控显示CRM平均响应7.2秒留1.3秒缓冲失败后自动切到缓存快照绝不让错误蔓延。可观测性Executor每执行一个Task立刻写入结构化日志JSON格式包含task_id、start_time、end_time、input_hash、output_truncate_200、error_type。运维同事用Grafana看一眼error_typeAPI_TIMEOUT的曲线飙升就知道是CRM那边出问题而不是怀疑Agent逻辑。轻量性整个调度核心代码仅387行无任何第三方依赖。客户审计时要求提供全部源码我直接打包一个.py文件他们用python -m py_compile agent_executor.py就能验证没藏后门。有人问“不用框架那记忆、工具、规划这些能力怎么实现”答案是只封装最必要、最稳定的原子能力。记忆模块用SQLite全文索引不是向量库因为客户明确要求“所有对话记录必须100%可检索、可导出、不依赖GPU”工具调用统一走OpenAPI 3.0规范的HTTP Client所有插件必须提供/openapi.jsonAgent启动时自动加载并校验schema规划能力干脆砍掉用预置的YAML流程模板替代——业务规则变化时运营人员改个YAML比调Prompt工程快十倍。这种“反潮流”的设计换来的是上线后连续217天零P0故障。记住Agent的价值不在多酷炫而在多可靠。当你需要它每天自动处理3000份合同审核时一个能稳定运行的while循环远胜十个花里胡哨的状态机。3. 环境与依赖Python虚拟环境里的“最小生存包”别急着pip install langchain先解决一个致命问题Python版本和包冲突。我见过太多团队在conda环境里装了PyTorch 2.3结果LangChain 0.1.0依赖的llama-cpp-python又强制要PyTorch 2.1最后整个环境变成“包坟场”。我的方案是彻底隔离——不用conda不用poetry就用Python原生venv且只装四个包httpx、jinja2、pydantic、sqlite3后者是标准库。其他所有能力都通过HTTP服务暴露Agent本身只是个轻量调度器。具体操作分三步3.1 创建纯净虚拟环境# Mac/Linux终端执行Windows请用PowerShell python3.11 -m venv ./agent_env source ./agent_env/bin/activate # Linux/Mac # Windows: .\agent_env\Scripts\Activate.ps1 # 验证环境纯净度 pip list --outdated # 应该返回空 pip install --upgrade pip pip install httpx jinja2 pydantic提示绝对不要在venv里装langchain、llamaindex、transformers。这些包体积大、依赖深、更新频繁是线上事故的温床。它们应该部署在独立服务中Agent只通过HTTP调用。3.2 工具服务化部署以CRM查询为例CRM插件不能写成Python函数塞进Agent里必须拆成独立HTTP服务。我用FastAPI写了个极简服务# crm_service.py from fastapi import FastAPI, HTTPException import httpx app FastAPI() app.post(/query_sales) async def query_sales(q: dict): # 硬编码超时8.5秒与Agent调度器一致 async with httpx.AsyncClient(timeout8.5) as client: try: resp await client.post(https://crm-api.example.com/v2/sales, jsonq, headers{X-API-Key: prod-key-xxxx}) resp.raise_for_status() return resp.json() except httpx.TimeoutException: raise HTTPException(504, CRM API timeout) except httpx.HTTPStatusError as e: raise HTTPException(e.response.status_code, str(e))启动命令uvicorn crm_service:app --host 0.0.0.0 --port 8001 --workers 4注意这个服务监听8001端口Agent调度器后续通过http://localhost:8001/query_sales调用。所有工具服务都按此模式部署端口从8001开始递增8002法规库、8003邮件服务……形成清晰的服务网格。33 记忆模块SQLite不是妥协是精准选择向量数据库别闹。客户法务部明确要求“所有用户对话必须能按时间、关键词、用户ID三字段100%精确检索且导出为Excel时格式零失真”。ChromaDB的模糊匹配、Pinecone的向量近似全不符合。我用SQLiteFTS5全文搜索扩展实现-- memory.db 初始化SQL CREATE VIRTUAL TABLE IF NOT EXISTS chat_history USING fts5( user_id, session_id, timestamp, input_text, output_text, content_type UNINDEXED -- 此字段不参与全文索引只用于精确过滤 ); -- 创建时间索引加速范围查询 CREATE INDEX IF NOT EXISTS idx_time ON chat_history(timestamp); CREATE INDEX IF NOT EXISTS idx_user ON chat_history(user_id);Python读写封装极简import sqlite3 from datetime import datetime def save_message(user_id: str, session_id: str, input_txt: str, output_txt: str): conn sqlite3.connect(memory.db) c conn.cursor() c.execute(INSERT INTO chat_history VALUES (?, ?, ?, ?, ?), (user_id, session_id, datetime.now().isoformat(), input_txt, output_txt)) conn.commit() conn.close() def search_messages(user_id: str, keyword: str, since: str None) - list: conn sqlite3.connect(memory.db) c conn.cursor() sql SELECT * FROM chat_history WHERE user_id ? AND input_text MATCH ? params [user_id, keyword] if since: sql AND timestamp ? params.append(since) c.execute(sql, params) return c.fetchall()实测10万条对话记录下search_messages(U123, 报销)平均耗时23ms比任何向量库的“相似度top3”还快还准。这才是业务需要的“记忆”。4. 核心模块实现从Prompt到Production的七道工序Agent不是写个Prompt就能跑它是一条精密装配线。我把整个构建过程拆成七个不可跳过的工序每道工序都有明确输入、输出、验收标准。少一道上线必崩。4.1 工具注册用OpenAPI 3.0代替手工写Function Call别再手写{name: query_crm, description: 查询CRM销售数据}这种脆弱结构。所有工具必须提供标准OpenAPI 3.0文档Agent启动时自动解析。我写了个tool_loader.pyimport httpx import json from pydantic import BaseModel class ToolSpec(BaseModel): name: str description: str method: str url: str request_schema: dict response_schema: dict def load_tools_from_openapi(openapi_url: str) - list[ToolSpec]: resp httpx.get(openapi_url) spec resp.json() tools [] for path, methods in spec[paths].items(): for method, op in methods.items(): if x-agent-tool not in op.get(extensions, {}): continue # 跳过非Agent工具 tools.append(ToolSpec( nameop[operationId], descriptionop[summary], methodmethod.upper(), urlfhttp://localhost:{get_port_by_service(op[servers][0][url])}{path}, request_schemaop[requestBody][content][application/json][schema], response_schemaop[responses][200][content][application/json][schema] )) return tools关键点x-agent-tool是自定义扩展字段只有打上这个标记的API才被Agent加载。这样CRM团队更新API时只要同步更新OpenAPI文档并保持x-agent-tool标记Agent重启后自动适配无需修改一行Agent代码。4.2 输入解析用Jinja2模板做Prompt工程的“钢筋”Prompt不是写散文是写结构化程序。我禁用所有f-string拼接全部用Jinja2模板{# prompt_templates/crm_query.j2 #} 你是一个专业的销售数据分析助手。请严格按以下步骤执行 1. 从用户输入中提取时间范围格式YYYY-MM-DD至YYYY-MM-DD、产品线如云服务、硬件、地区如华东、华北 2. 调用query_sales工具参数必须包含 - time_range: [{{ time_start }}, {{ time_end }}] - product_line: {{ product_line }} - region: {{ region }} 3. 收到结果后用中文生成简洁报告包含Top3客户名称、销售额、环比变化 用户输入{{ user_input }}Python渲染from jinja2 import Environment, FileSystemLoader env Environment(loaderFileSystemLoader(prompt_templates)) template env.get_template(crm_query.j2) prompt template.render( user_input查下2024年Q3华东地区云服务销售额Top3客户, time_start2024-07-01, time_end2024-09-30, product_line云服务, region华东 )实操心得模板里所有变量必须有默认值如{{ region|default(全国) }}否则用户输入缺字段时整个Prompt崩溃。我强制要求每个模板文件配套schema.json用JSON Schema校验传入参数不合法直接抛异常绝不让错误进入大模型。4.3 执行编排手写状态机的七种状态Agent调度器核心是有限状态机FSM我定义七种状态每种状态有唯一入口、唯一出口、超时保护状态触发条件执行动作超时下一状态INIT启动加载工具列表、初始化内存连接2sPARSE_INPUTPARSE_INPUT收到用户输入调用NLP模块提取实体时间/产品/地区1.5sVALIDATE_ENTITIESVALIDATE_ENTITIES实体提取完成校验时间格式、产品线是否存在0.5sGENERATE_TOOL_CALLGENERATE_TOOL_CALL实体校验通过渲染Prompt模板调用大模型API12sEXECUTE_TOOLEXECUTE_TOOL大模型返回tool_call解析JSON调用对应HTTP工具工具自身timeoutHANDLE_TOOL_RESULTHANDLE_TOOL_RESULT工具返回判断是否需重试/降级/终止1sGENERATE_RESPONSEGENERATE_RESPONSE工具结果就绪渲染回复模板保存记忆3sIDLE状态流转代码简化class AgentFSM: def __init__(self): self.state INIT self.context {} def transition(self, event: str, data: dict None): if self.state INIT and event START: self._load_tools() self.state PARSE_INPUT elif self.state PARSE_INPUT and event INPUT_RECEIVED: self.context.update(self._parse_entities(data[input])) self.state VALIDATE_ENTITIES # ... 其他状态省略 else: raise RuntimeError(fInvalid transition: {self.state} {event})注意每个状态的超时值不是拍脑袋而是基于压测数据。比如GENERATE_TOOL_CALL设12秒是因为我们测试了100次Qwen2-7B API调用P95延迟是11.2秒加0.8秒缓冲。所有超时值写死在代码里不从配置文件读——配置中心挂了Agent至少还能靠超时熔断保命。4.4 错误熔断比Retry更关键的“断电保护”90%的Agent教程只讲Retry却忽略更致命的场景当CRM工具连续5次超时Agent不该傻等而该立即切换到“降级模式”。我的熔断器设计如下class CircuitBreaker: def __init__(self, failure_threshold5, timeout60): self.failure_count 0 self.failure_threshold failure_threshold self.timeout timeout self.last_failure_time 0 def call(self, func, *args, **kwargs): if self._is_open(): return self._fallback(*args, **kwargs) # 返回缓存数据或静态文案 try: result func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise e def _is_open(self): if self.failure_count self.failure_threshold: if time.time() - self.last_failure_time self.timeout: return True return False def _on_failure(self): self.failure_count 1 self.last_failure_time time.time() def _on_success(self): self.failure_count 0实测效果CRM服务宕机时Agent在第5次失败后自动启用本地SQLite缓存的上周数据回复“根据最新可用数据Q3 Top3客户为...”而非卡死报错。业务部门反馈“比原来一直转圈好一万倍”。4.5 输出渲染用Markdown生成可执行的“活文档”Agent的输出不能只是文字得是能直接点击、复制、运行的活文档。我强制所有回复用Markdown且包含可交互元素## Q3销售Top3客户数据截至2024-09-30 | 排名 | 客户名称 | 销售额万元 | 环比变化 | |------|----------|----------------|----------| | 1 | 某科技有限公司 | 1,280 | 12.3% | | 2 | 某集团控股 | 956 | 5.7% | | 3 | 某信息股份 | 732 | -2.1% | ### 查看详细报表 [点击查看完整Excel](http://report-service.example.com/export?sessionabc123) ### ✉️ 发送此报告 [一键发送邮件](mailto:salescompany.com?subjectQ3销售报告body详见附件) 提示点击链接将自动触发对应服务无需手动复制粘贴。Python渲染引擎会自动将[点击查看...]转换为实际URL并注入唯一session_id用于审计追踪。用户点邮件链接Outlook自动填充收件人和主题——这才是真正的“自动化”。4.6 内存写入事务安全的双写保障每次Agent完成一轮交互必须保证记忆100%落库。我采用“双写校验”def save_to_memory_and_audit(user_input: str, agent_output: str, session_id: str): # 第一步写入主内存库SQLite save_message(AGENT, session_id, user_input, agent_output) # 第二步写入审计日志追加写入文本文件永不覆盖 with open(audit.log, a) as f: f.write(f[{datetime.now().isoformat()}] SESSION:{session_id} | INPUT:{user_input[:50]}... | OUTPUT_LEN:{len(agent_output)}\n) # 第三步校验写入一致性 conn sqlite3.connect(memory.db) c conn.cursor() c.execute(SELECT COUNT(*) FROM chat_history WHERE session_id ?, (session_id,)) count c.fetchone()[0] conn.close() if count 0: raise RuntimeError(Memory write failed! Audit log written but SQLite missing.)关键细节审计日志用纯文本追加写入不依赖数据库事务确保即使SQLite损坏至少有原始日志可追溯。这是金融、医疗类客户强制要求的底线。4.7 启动与监控让运维看得懂的健康检查Agent服务必须提供标准健康检查端点且返回信息对运维友好app.get(/healthz) def health_check(): # 检查工具服务连通性 tool_health {} for tool in TOOLS: try: resp httpx.get(fhttp://{tool.host}:{tool.port}/healthz, timeout2) tool_health[tool.name] UP if resp.status_code 200 else DOWN except Exception: tool_health[tool.name] DOWN # 检查内存库 try: conn sqlite3.connect(memory.db) conn.execute(SELECT 1) memory_status UP conn.close() except Exception: memory_status DOWN return { status: UP if all(v UP for v in tool_health.values()) and memory_status UP else DOWN, timestamp: datetime.now().isoformat(), components: { tools: tool_health, memory: memory_status, executor: RUNNING } }运维用curl http://localhost:8000/healthz就能看到所有依赖状态无需登录服务器查进程。这才是生产级Agent该有的样子。5. 实操全流程从克隆仓库到处理第一笔业务请求现在把所有模块串起来走一遍真实工作流。假设你要搭建一个“专利检索辅助Agent”目标是帮研发人员快速查某技术方案是否已被专利覆盖。5.1 准备工作三分钟初始化# 1. 克隆最小化Agent框架我已开源 git clone https://github.com/real-dev/lean-agent.git cd lean-agent # 2. 创建虚拟环境同前文 python3.11 -m venv venv source venv/bin/activate pip install -r requirements.txt # 仅含httpx/jinja2/pydantic # 3. 下载预置工具服务已打包为Docker镜像 docker pull lean-agent/crm-service:1.0 docker pull lean-agent/patent-api:2.1 docker pull lean-agent/email-gateway:0.9 # 4. 启动所有依赖服务 docker run -d --name patent-api -p 8002:8002 lean-agent/patent-api:2.1 docker run -d --name email-gw -p 8003:8003 lean-agent/email-gateway:0.95.2 配置Agent修改两个文件编辑config.yaml# config.yaml agent: model_endpoint: https://api.together.xyz/v1/chat/completions # Together.ai免费额度够用 model_api_key: YOUR_TOGETHER_KEY # 注册Together.ai获取 timeout: 12 tools: - name: search_patents openapi_url: http://localhost:8002/openapi.json # 专利服务OpenAPI地址 - name: send_email openapi_url: http://localhost:8003/openapi.json memory: db_path: ./memory.db编辑prompt_templates/patent_search.j2你是一个资深专利分析师。请严格按以下步骤执行 1. 从用户输入中提取技术关键词如锂电池、图像识别、申请人如宁德时代、华为、申请日期范围 2. 调用search_patents工具参数必须包含 - keywords: [{{ keyword1 }}, {{ keyword2 }}] - applicants: [{{ applicant }}] - date_range: [{{ start_date }}, {{ end_date }}] 3. 收到专利列表后用中文生成摘要包含 - 相关专利数量 - 最新公开号及标题 - 是否存在高风险专利权利要求含“方法”、“系统”字样 用户输入{{ user_input }}5.3 启动Agent并测试# 启动Agent主服务 python main.py --config config.yaml # 在另一个终端测试模拟用户请求 curl -X POST http://localhost:8000/ask \ -H Content-Type: application/json \ -d {session_id:TEST-001, input:查下‘固态电池电解质’相关专利申请人是宁德时代2023年之后的}预期返回截取关键部分{ session_id: TEST-001, response: ## 固态电池电解质专利分析宁德时代2023-2024\n\n- **相关专利总数**17项\n- **最新公开号**CN117887523A《一种固态电池电解质及其制备方法》\n- **高风险专利**3项含方法权利要求\n\n### 查看全部专利\n[下载完整PDF列表](http://patent-gateway.example.com/download?idsCN117887523A,CN117682341A)\n\n### 发送分析报告\n[邮件发送给张工](mailto:zhangcompany.com?subject固态电池专利分析body详见附件) }5.4 生产部署Kubernetes的极简配置别被K8s吓住核心就三个文件# deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: lean-agent spec: replicas: 2 selector: matchLabels: app: lean-agent template: metadata: labels: app: lean-agent spec: containers: - name: agent image: lean-agent/core:1.2 ports: - containerPort: 8000 env: - name: MODEL_API_KEY valueFrom: secretKeyRef: name: together-secrets key: api-key volumeMounts: - name: memory-db mountPath: /app/memory.db volumes: - name: memory-db persistentVolumeClaim: claimName: agent-memory-pvc# service.yaml apiVersion: v1 kind: Service metadata: name: lean-agent-svc spec: selector: app: lean-agent ports: - port: 80 targetPort: 8000 type: ClusterIP# ingress.yaml对接公司统一API网关 apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: lean-agent-ingress annotations: nginx.ingress.kubernetes.io/rewrite-target: / spec: rules: - host: agent.company.internal http: paths: - path: /v1/ask pathType: Prefix backend: service: name: lean-agent-svc port: number: 80实操心得K8s里最易错的是volumeMounts路径。我强制规定所有Agent容器的内存库路径必须是/app/memory.db且PVC必须用ReadWriteOnce模式——因为SQLite不支持多实例并发写入强行用ReadWriteMany必锁表。6. 常见问题与避坑指南血泪换来的十三条军规以下是我在七个落地项目中总结的高频问题按发生频率排序每条都附真实案例和根治方案。6.1 问题Agent调用工具后卡死日志显示“waiting for response”现象curl -X POST http://localhost:8000/ask一直挂起htop看CPU 0%netstat显示连接处于ESTABLISHED但无数据。根因工具服务如专利API返回了Transfer-Encoding: chunked但未正确结束chunk漏发0\r\n\r\n导致Agent的HTTP Client永远等待下一块数据。解决方案在Agent的HTTP Client中强制设置httpx.AsyncClient(timeout8.5, http2False)禁用HTTP/2和分块传输改用Content-Length校验。实测后故障率从37%降至0%。注意所有工具服务必须在OpenAPI文档中标明responses: {200: {content: {application/json: {schema: {...}}}}}Agent据此校验响应体完整性。6.2 问题记忆模块搜不到历史对话search_messages(U123, 报销)返回空现象用户明明昨天问过报销流程今天再问却得不到上下文。根因SQLite FTS5默认对短词3字符不索引而“报销”二字被当成停用词过滤。解决方案创建FTS5表时指定tokenizeunicode61 remove_diacritics 0并添加prefix2,3,4CREATE VIRTUAL TABLE chat_history USING fts5( user_id, session_id, timestamp, input_text, output_text, tokenizeunicode61 remove_diacritics 0, prefix2,3,4 );这样“报”、“销”、“报销”都会被索引搜索准确率100%。6.3 问题大模型输出格式错乱JSON解析失败现象Agent调用query_sales后大模型返回我帮你查到了结果如下 {customers: [{name: 某科技, sales: 1280}]}导致json.loads()报错因为前面多了废话。根因Prompt模板没加严格约束。大模型在“思考”时习惯性加解释性文字。解决方案在Jinja2模板末尾强制加一行...前面逻辑 请严格按以下JSON格式输出不要任何额外文字、不要json包裹、不要注释 {tool: query_sales, params: {time_range: [2024-07-01, 2024-09-30], product_line: 云服务}}6.4 问题工具服务间Cookie丢失登录态失效现象专利服务需要先调/login获取Cookie再调/search但Agent两次HTTP调用Cookie不共享。根因Agent用httpx.AsyncClient()每次新建实例Cookie不持久。解决方案全局复用一个httpx.AsyncClient实例并启用Cookie持久化# 在Agent初始化时 global_http_client httpx.AsyncClient( cookieshttpx.Cookies(), timeouthttpx.Timeout(8.5) )6.5 问题多用户并发时Session ID混淆现象用户A的请求意外触发了用户B的邮件发送。根因Session ID生成用了uuid.uuid4()但没绑定用户且工具调用时未透传Session ID。解决方案Session ID必须由前端生成如JWT中的jtiAgent全程透传所有工具服务在OpenAPI中声明x-session-idheader# openapi.json片段 /search: post: parameters: - name: x-session-id in: header required: true schema: type: string6.6 问题大模型幻觉生成不存在的工具名现象用户问“查专利”大模型返回{tool: search_patentzzz, params: {...}}Agent找不到对应工具。根因Prompt没限制工具名范围。解决方案在Jinja2模板中硬编码工具列表可用工具列表必须从中选择 - search_patents查询专利数据库 - send_email发送邮件 - generate_report生成PDF报告 请从以上列表中选择一个工具名不要发明新名字。6.7 问题时区混乱导致时间范围计算错误现象用户说“查今天数据”Agent调用时传了2024-05-20但专利库时区是UTC8实际查的是昨天。根因Agent服务器时区为UTC未统一转换。解决方案所有时间处理强制用datetime.now(timezone(timedelta(hours8)))且OpenAPI文档中所有时间字段注明timezone: Asia/Shanghai。6.8 问题长文本截断导致关键信息丢失现象专利摘要长达2000字Agent只取前500字漏掉权利要求关键句。根因Prompt模板里写了{{ abstract[:500] }}。解决方案用Jinja2的truncate过滤器并开启enddotsFalse{{ abstract|truncate(length500, enddotsFalse) }}这样截断到最近的句号不硬切。6.9 问题工具返回空数组Agent无法处理现象search_patents返回[]Agent直接崩溃没走“无结果”分支。根因状态机没定义HANDLE_EMPTY_RESULT状态。解决方案在HANDLE_TOOL_RESULT状态里增加分支if len(tool_result) 0: self.state GENERATE_EMPTY_RESPONSE else: self.state GENERATE_RESPONSE6.10 问题模型API限流Agent疯狂重试压垮服务现象Together.ai返回429Agent立即重试1秒内发10次请求。根因没实现指数退避。解决方案在HTTP Client封装层加退避import asyncio from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) async def call_model_api(prompt: str): ...6.11 问题