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数据这步,我硬编码
timeout=8.5秒(因为监控显示CRM平均响应7.2秒,留1.3秒缓冲),失败后自动切到缓存快照,绝不让错误蔓延。 - 可观测性:Executor每执行一个Task,立刻写入结构化日志(JSON格式),包含
task_id、start_time、end_time、input_hash、output_truncate_200、error_type。运维同事用Grafana看一眼error_type=="API_TIMEOUT"的曲线飙升,就知道是CRM那边出问题,而不是怀疑Agent逻辑。 - 轻量性:整个调度核心代码仅387行,无任何第三方依赖。客户审计时要求提供全部源码,我直接打包一个.py文件,他们用
python -m py_compile agent_executor.py就能验证没藏后门。
有人问:“不用框架,那记忆、工具、规划这些能力怎么实现?”答案是:只封装最必要、最稳定的原子能力。记忆模块用SQLite+全文索引(不是向量库),因为客户明确要求“所有对话记录必须100%可检索、可导出、不依赖GPU”;工具调用统一走OpenAPI 3.0规范的HTTP Client,所有插件必须提供/openapi.json,Agent启动时自动加载并校验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(timeout=8.5) as client: try: resp = await client.post("https://crm-api.example.com/v2/sales", json=q, 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的向量近似,全不符合。我用SQLite+FTS5(全文搜索扩展)实现:
-- 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.py:
import 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( name=op["operationId"], description=op["summary"], method=method.upper(), url=f"http://localhost:{get_port_by_service(op['servers'][0]['url'])}{path}", request_schema=op["requestBody"]["content"]["application/json"]["schema"], response_schema=op["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(loader=FileSystemLoader("prompt_templates")) template = env.get_template("crm_query.j2") prompt = template.render( user_input="查下2024年Q3华东地区云服务销售额Top3客户", time_start="2024-07-01", time_end="2024-09-30", product_line="云服务", region="华东" )实操心得:模板里所有变量必须有默认值(如
{{ region|default('全国') }}),否则用户输入缺字段时整个Prompt崩溃。我强制要求每个模板文件配套schema.json,用JSON Schema校验传入参数,不合法直接抛异常,绝不让错误进入大模型。
4.3 执行编排:手写状态机的七种状态
Agent调度器核心是有限状态机(FSM),我定义七种状态,每种状态有唯一入口、唯一出口、超时保护:
| 状态 | 触发条件 | 执行动作 | 超时 | 下一状态 |
|---|---|---|---|---|
| INIT | 启动 | 加载工具列表、初始化内存连接 | 2s | PARSE_INPUT |
| PARSE_INPUT | 收到用户输入 | 调用NLP模块提取实体(时间/产品/地区) | 1.5s | VALIDATE_ENTITIES |
| VALIDATE_ENTITIES | 实体提取完成 | 校验时间格式、产品线是否存在 | 0.5s | GENERATE_TOOL_CALL |
| GENERATE_TOOL_CALL | 实体校验通过 | 渲染Prompt模板,调用大模型API | 12s | EXECUTE_TOOL |
| EXECUTE_TOOL | 大模型返回tool_call | 解析JSON,调用对应HTTP工具 | 工具自身timeout | HANDLE_TOOL_RESULT |
| HANDLE_TOOL_RESULT | 工具返回 | 判断是否需重试/降级/终止 | 1s | GENERATE_RESPONSE |
| GENERATE_RESPONSE | 工具结果就绪 | 渲染回复模板,保存记忆 | 3s | IDLE |
状态流转代码(简化):
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(f"Invalid 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_threshold=5, timeout=60): 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?session=abc123) ### ✉️ 发送此报告 [一键发送邮件](mailto:sales@company.com?subject=Q3销售报告&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(f"http://{tool.host}:{tool.port}/healthz", timeout=2) 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?ids=CN117887523A,CN117682341A)\n\n### 📩 发送分析报告\n[邮件发送给张工](mailto:zhang@company.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(timeout=8.5, http2=False),禁用HTTP/2和分块传输,改用Content-Length校验。实测后故障率从37%降至0%。 注意:所有工具服务必须在OpenAPI文档中标明
"responses": {"200": {"content": {"application/json": {"schema": {...}}}}},Agent据此校验响应体完整性。
6.2 问题:记忆模块搜不到历史对话,search_messages("U123", "报销")返回空
- 现象:用户明明昨天问过报销流程,今天再问却得不到上下文。
- 根因:SQLite FTS5默认对短词(<3字符)不索引,而“报销”二字被当成停用词过滤。
- 解决方案:创建FTS5表时指定
tokenize="unicode61 remove_diacritics 0"并添加prefix='2,3,4':
这样“报”、“销”、“报销”都会被索引,搜索准确率100%。CREATE VIRTUAL TABLE chat_history USING fts5( user_id, session_id, timestamp, input_text, output_text, tokenize='unicode61 remove_diacritics 0', prefix='2,3,4' );
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( cookies=httpx.Cookies(), timeout=httpx.Timeout(8.5) )
6.5 问题:多用户并发时Session ID混淆
- 现象:用户A的请求意外触发了用户B的邮件发送。
- 根因:Session ID生成用了
uuid.uuid4()但没绑定用户,且工具调用时未透传Session ID。 - 解决方案:Session ID必须由前端生成(如JWT中的jti),Agent全程透传,所有工具服务在OpenAPI中声明
x-session-idheader:# openapi.json片段 "/search": post: parameters: - name: x-session-id in: header required: true schema: type: string
6.6 问题:大模型幻觉生成不存在的工具名
- 现象:用户问“查专利”,大模型返回
{"tool": "search_patentzzz", "params": {...}},Agent找不到对应工具。 - 根因:Prompt没限制工具名范围。
- 解决方案:在Jinja2模板中硬编码工具列表:
可用工具列表(必须从中选择): - search_patents:查询专利数据库 - send_email:发送邮件 - generate_report:生成PDF报告 请从以上列表中选择一个工具名,不要发明新名字。
6.7 问题:时区混乱导致时间范围计算错误
- 现象:用户说“查今天数据”,Agent调用时传了
"2024-05-20",但专利库时区是UTC+8,实际查的是昨天。 - 根因:Agent服务器时区为UTC,未统一转换。
- 解决方案:所有时间处理强制用
datetime.now(timezone(timedelta(hours=8))),且OpenAPI文档中所有时间字段注明timezone: "Asia/Shanghai"。
6.8 问题:长文本截断导致关键信息丢失
- 现象:专利摘要长达2000字,Agent只取前500字,漏掉权利要求关键句。
- 根因:Prompt模板里写了
{{ abstract[:500] }}。 - 解决方案:用Jinja2的
truncate过滤器并开启enddots=False:
这样截断到最近的句号,不硬切。{{ abstract|truncate(length=500, enddots=False) }}
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_RESPONSE"
6.10 问题:模型API限流,Agent疯狂重试压垮服务
- 现象:Together.ai返回429,Agent立即重试,1秒内发10次请求。
- 根因:没实现指数退避。
- 解决方案:在HTTP Client封装层加退避:
import asyncio from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10)) async def call_model_api(prompt: str): ...