从零搭建一个可用可跑的 AI Agent 系统:以 Agent-Reach 为例聊聊我踩过的坑
“Agent-Reach”这类名字这两年特别多,但很多人聊 Agent 都停留在概念层。我今年实际把一个名为 Agent-Reach 的 AI 代理系统从前端交互、后端调度、工具注册、上下文管理到部署上线完整跑通了。这篇文章把整个设计思路、核心实现、以及我在实际开发中踩过的坑,一次性说清楚。文章面向的是想真正动手实现 Agent 系统的开发者,无论你是刚接触大模型调用,还是已经在做 RAG、插件化工具集成,我这里的内容都能给你一些实际参考。
Agent-Reach 这个名字拆开来看,重点在“Reach 触达”上——让 AI 代理真正触达业务系统、外部API、数据库和用户终端。它不是聊天机器人演示,而是能自主拆解任务、调用工具、处理多轮上下文的完整系统。目标就是让你输入一个自然语言指令,Agent 能自主规划并执行,直到目标达成或需要人工介入。
下面直接进入正题,从系统设计、核心细节、实操实现到问题排查,一个一个环节讲。
1. 整体架构与核心设计思路
1.1 为什么 Agent 不能只靠一个 Prompt 撑起来
很多人第一步想到的是,写一个“你是智能助手”的 System Prompt,然后不断把用户消息灌进大模型,靠模型自己“悟”调度。早期 Demo 可以这么跑,但要接真实业务就崩。原因有几个:
- 上下文长度有限,多轮对话 + 工具返回结果很容易直接把 Token 打满。
- 模型缺乏对业务系统状态的真实感知,它不知道订单是否存在、库存是否充足、API 是否可用。
- 一旦流程中途出现异常,没有状态管理和恢复机制,整个任务就断在那里。
所以 Agent-Reach 在设计时采用了一个更工程化的思路:Agent 只负责决策和意图拆解,具体能力全部由注册在工具列表里的函数完成,状态则由独立的上下文管理层维护,对话与执行状态分离。这个思路借鉴了业界主流的 ReAct 模式——模型通过 Thought(思考)、Action(行动)、Observation(观察)循环完成推理,每次行动都走工具调用通道,而不是让模型“猜”业务结果。
ReAct 的核心价值在于把模型的长处(语言理解、任务分解)和短处(不具备实时数据感知)分开,模型只负责判断下一步动作,实际取值都来自工具的返回结果。这种设计的好处是,如果模型选错了工具,返回的错误信息可以立即反馈给模型,它有能力自我修正,而不是一路错到底。
1.2 模块划分:调度、记忆、工具、执行
Agent-Reach 从代码结构上拆成了四个核心模块,每个模块职责单一,边界清晰:
| 模块 | 核心职责 | 关键点 |
|---|---|---|
| 调度器(Agent Core) | 解析用户意图、调用大模型、维护推理循环 | 决定“下一步做什么” |
| 记忆管理器(Memory Manager) | 管理短期上下文与长期会话记录 | 控制 Token 消耗与信息检索 |
| 工具注册中心(Tool Registry) | 工具列表注册、参数描述生成、函数路由 | 决定 Agent 能“干什么” |
| 执行器(Executor) | 真正调用业务函数/API、获取结果 | 将 Action 翻译为可执行指令 |
调度器是大脑,工具注册中心是四肢,记忆管理器是短期记忆,执行器是血肉。为什么要把工具注册中心单独拿出来?关键在于大模型 Function Calling 的机制:模型本身不直接调用函数,而是输出一个结构化的、包含函数名和参数的 JSON。我们拿到这个 JSON 后,再由本地代码去找到真正的函数并执行。工具注册中心就是把“模型能认识的描述”和“本地能运行的函数”绑定在一起的桥梁。
这个绑定关系如果耦合在调度器内部,每加一个新工具就得改核心代码;单独拆出来之后,新增能力只需要往注册中心加一条记录,调度器完全不动。我在开发过程中把工具数量从 3 个扩展到 30 多个时,这个拆分的收益非常明显。
1.3 技术选型与关键依赖
Agent-Reach 的基座模型我用了 OpenAI 系的 GPT-4o-mini 和 GPT-4o 双档位,但实际上整个系统对基座模型并不绑定,接口层做了抽象,理论上可以换成 Claude、国产通义千问或开源模型。工具调用走的是 Function Calling 协议,如果基座模型不支持原生 Function Calling,也可以退化为让模型输出 JSON 格式的 Action 文本,再用 Pydantic 做校验,效果会差一些但也能跑。
编程语言选的 Python + FastAPI。FastAPI 用在这里很合适,原因有三:一是它对异步原生支持好,工具执行时遇到 IO 密集操作可以直接 async 并发;二是它有完善的 Pydantic 集成,模型返回的 JSON 参数可以自动做类型校验;三是自带 OpenAPI 文档,调试工具调用时可以直接在 Swagger UI 里手动触发接口。
向量数据库用的是 ChromaDB,负责长期记忆中的相似度检索。这块后面会细讲。
2. 核心细节解析:从工具定义到记忆管理
2.1 工具注册:Function Calling 的终极封装
工具注册是整个系统里最细致、也最影响稳定性的环节。每个工具注册时需要提供的信息包括:函数名、描述、参数定义(JSON Schema)、执行权重(何时适合用这个工具)。这些信息最后会被组装成大模型能理解的 tools 参数。
一个典型工具注册的代码结构,这里我以查天气为例,展示核心思路:
from pydantic import BaseModel, Field class WeatherQueryParams(BaseModel): city: str = Field(description="城市名称,如 北京、上海") date: str = Field(default="today", description="查询日期,默认今天") async def get_weather(city: str, date: str = "today"): # 这里调用真实天气 API return {"city": city, "date": date, "weather": "晴", "temperature": 26} REGISTRY = { "get_weather": { "name": "get_weather", "description": "查询指定城市的天气情况,当用户问天气、温度、是否下雨时使用", "parameters": WeatherQueryParams.model_json_schema(), "handler": get_weather, } }这段代码看起来简单,注意几个细节:
- 参数的 description 要写清楚,因为模型就是靠这个决定该填什么值。
- handler 可以是同步函数也可以是 async 函数,注册中心在调用时统一按 async 处理,这样遇到 IO 型任务不会阻塞整个 Agent 循环。
- 返回结构尽量统一,建议都返回字典,后续统一序列化给模型看。
工具描述的质量直接决定了模型选对工具的概率。我做过测试,同一个查询接口,描述写得模糊(“查询天气”)时,模型经常误选成别的工具;写成触发条件式(“当用户问天气、温度、是否下雨时使用”),准确率从 60% 提升到 95% 以上。这个提升不花一分钱,全靠描述打磨。
2.2 参数填充:模型不擅长的事别让它做
模型真正擅长的是语义理解,而不是精确计算。举一个典型场景:用户说“帮我查下后天深圳的天气”。模型如果要填充日期参数,它需要知道“后天”是哪一天。如果你没有任何干预机制,模型可能会填一个格式错误的日期,或者干脆填个字符串“后天”。解决措施是:在让模型调用工具之前,调度器先做一次参数预处理。
具体做法是,维护一个“动态参数注入器”:
def inject_params(func_name: str, params: dict, user_context: dict): if "date" in params: if params["date"] == "后天": from datetime import datetime, timedelta params["date"] = (datetime.now() + timedelta(days=2)).strftime("%Y-%m-%d") elif params["date"] == "今天": params["date"] = datetime.now().strftime("%Y-%m-%d") return params这种硬编码方式看起来有点“土”,但在工程上极其有效。还有一类参数需要从用户历史会话中继承,比如用户说“查一下最近订单的物流状态”,他之前的对话里已经指定了“最近订单”的编号,这时候就需要从记忆管理器里把订单号取出来填进去,模型本身不知道订单号是什么。这个逻辑其实就是领域知识的注入,把业务规则放在 Agent 之外,比让模型猜测要可靠得多。
2.3 上下文窗口管理:Token 不够时的三层降级策略
上下文管理是 Agent 落地时第一个会遇到的硬墙。我刚开始做的时候,连续五轮对话之后模型就开始“失忆”。根本原因是基础模型的上下文窗口有限,而工具返回结果通常很长,天气接口还好,但如果是数据库查询结果,可能一次返回几百行。
Agent-Reach 的上下文管理采用三层策略,按优先级依次处理:
- 第一层:摘要压缩。当会话窗口超过阈值时,把最早的历史消息发送给模型做摘要,用一段 200 字以内的概括替换掉原始对话。代价是丢失细节,但保住宏观意图。
- 第二层:关键信息提取。每一轮工具返回结果中,提取核心字段(比如订单状态、金额、地址等),过滤掉冗余字段后再进入上下文。这个需要每个工具返回时自己定义 summarizer 逻辑。
- 第三层:向量化检索。超过一定轮次的旧消息,不再进上下文,而是写入 ChromaDB 向量库。当新问题涉及历史信息时,先做相似度检索,把最相关的 3~5 条片段捞回来。
我用一个例子说明为什么第三层很关键:用户上午问过“我的笔记本电脑订单什么时候发货”,下午问“那台电脑的发票能开吗”。如果不做检索,模型根本不知道“那台电脑”指的哪个订单。记忆管理器检索后,把上午的订单消息重新插入上下文,模型立刻理解指的是同一件事。
三层策略组合之后,我在实际压力测试中跑了 50 轮连续对话,模型始终能平稳理解意图,不再出现“失忆式回答”。
2.4 长时记忆与向量检索的经验之谈
ChromaDB 我用下来最大的感受是:检索质量的好坏,不取决于向量数据库本身,而取决于你存进去什么。如果只是把原始对话文本直接塞进去,检索效果会很差——因为日常对话里充满了代词、省略、模糊表达。
建议做法是:在写入向量库之前,先让模型把这段对话改写为一段规范化的摘要,提取成包含主语、时间、对象、状态的短句。比如用户说“我昨天那个订单好像还没到货”,摘要改写为“用户订单 X(创建于昨天)当前物流状态未签收”。这样向量化之后的匹配精度明显提高。代价是多一次模型调用,但这个成本完全值得。
另外,建议在每条记忆里附加一个 metadata,包含:会话 ID、创建时间戳、记忆类型(意图/实体/事实)。检索时用 metadata 做前置过滤,可以大幅降低无关内容的干扰。
3. 实操过程:把 Agent-Reach 真正跑起来
3.1 搭建最小可运行的 Agent-Reach 骨架
一个最小可运行版本其实不需要太多代码。核心只有三块:模型接入、工具注册、推理循环。我先把最小骨架贴出来,后面再逐步增加细节。
import json import asyncio from openai import AsyncOpenAI client = AsyncOpenAI(api_key="your-key", base_url="your-endpoint") INSTRUCTIONS = """ 你是一个智能代理,可以调用工具解决用户的问题。 你必须严格按照工具描述输出调用请求,格式为 JSON: {"name": "工具名", "parameters": {"参数名": "参数值"}} 如果判断不需要调用工具,直接输出自然语言回答。 """ async def run_agent(user_input: str, history: list = None): history = history or [] messages = [{"role": "system", "content": INSTRUCTIONS}] + history + [ {"role": "user", "content": user_input} ] formatted_tools = [ {"type": "function", "function": { "name": tool["name"], "description": tool["description"], "parameters": tool["parameters"], }} for tool in REGISTRY.values() ] resp = await client.chat.completions.create( model="gpt-4o", messages=messages, tools=formatted_tools, tool_choice="auto", ) choice = resp.choices[0] if choice.message.tool_calls: # 解析工具调用 call = choice.message.tool_calls[0] func_name = call.function.name params = json.loads(call.function.arguments) handler = REGISTRY[func_name]["handler"] result = await handler(**params) # 把结果回填给模型 messages.append(choice.message) messages.append({ "role": "tool", "tool_call_id": call.id, "content": json.dumps(result, ensure_ascii=False), }) # 让模型基于工具结果生成最终回答 resp2 = await client.chat.completions.create( model="gpt-4o", messages=messages, tools=formatted_tools, ) return resp2.choices[0].message.content return choice.message.content这是最核心的路由逻辑。你可以看到,模型返回工具调用后,我们先执行函数,再把结果以 role="tool" 的身份回传给模型,模型拿到结果后再生成最终回复。整个过程在一个循环里,如果一次执行后模型认为还需要调用其他工具,它会再返回新的 tool_calls,代码里应该再加一层while循环来处理多轮工具调用。
3.2 交互层与前端通信
Agent-Reach 的后端我用 FastAPI 暴露一个 WebSocket 接口,前端通过 WebSocket 发送用户消息,后端按流式方式推送进度事件。为什么用 WebSocket 而不是 HTTP 轮询?因为 Agent 执行一个多步骤任务可能耗时 5~20 秒,用户需要实时看到“正在调用天气接口”“正在查询数据库”“正在生成回答”这些中间状态,体验会好很多。
@app.websocket("/ws/chat") async def chat_endpoint(ws: WebSocket): await ws.accept() while True: user_msg = await ws.receive_text() await ws.send_json({"type": "status", "content": "正在分析意图..."}) answer = await run_agent(user_msg, history=session_history) await ws.send_json({"type": "answer", "content": answer})前端收到 status 事件后可以渲染成“步骤卡片”,告诉用户当前执行到哪一步。这里有个经验:即使工具执行失败,也要把失败原因回传给用户看,不要让系统死寂。透明感比完美感重要得多。
3.3 Agent 循环中的限流与并发控制
当你把 Agent 的能力接入真实业务后,很快会撞上两个问题:大模型 API 的 QPS 限制、以及工具本身可能带来的副作用(比如重复下单)。我在这块踩过比较深的坑,总结下来有三条铁律。
第一,所有对大模型的调用必须走一个带信号量限流的封装层。一个用户同时发起多个任务,或者多个终端接入,如果把几十个请求同时打给模型,很容易触发 429 限流错误。用一个asyncio.Semaphore(5)把并发数控制在 5 以内,整体吞吐不降反升,因为 429 重试导致的等待比主动排队更长。
第二,敏感的写操作工具必须加确认机制。Agent 推理得再准确,也不可能让它在没有用户确认的情况下去调用“下单”“转账”“删除数据”这类工具。实现方式是在注册中心给工具配置一个require_confirm: True的标记。调度器一旦发现模型要调用的工具带这个标记,它会先停止执行,向用户发送确认卡片,用户点了确认之后才真正执行。这一步你可以在框架层面统一处理,不需要每个工具自己实现。
第三,超时机制。工具执行可能因为下游接口卡住而长时间不返回。我给每个工具调用设置 15 秒超时,超时直接把失败结果回填给模型,让模型告诉用户“查询超时,请稍后再试”。虽然多花了一点 Prompt Token,但避免了整个 Agent 卡死。
4. 常见问题与排查技巧
4.1 模型选错工具或参数频出乱填
这个是最常见也最让人头大的。排查时我第一个看的是工具描述。如果描述里没有写明“什么场景下用”“触发条件是什么”,模型自然容易选错。建议:每个工具的描述以“当用户想……时使用”开头,参数描述里尽量写清楚边界和示例。
第二个排查点是参数格式化问题。GPT-4o 系列对 JSON Schema 里的 enum 约束遵守得比较严格,但对字符串格式就没那么可靠。比如定义了一个参数date类型为string且格式要求YYYY-MM-DD,模型仍然可能输出4/5这种不规范的日期。解决办法不是去抱怨模型不听话,而是在参数注入层增加格式化函数,强制把非标准格式转换成系统标准格式。
还有一个容易被忽视的问题:工具数量超过一定规模后,模型的选择准确率会明显下降。我的经验值是 15~20 个工具是一个临界点,超过之后就要考虑对工具做分组路由。比如一级路由判断“用户问题涉及订单还是物流”,再在二级路由里从该组的 5 个工具中做选择。这样虽然多了一次模型调用,但准确率稳得多。
4.2 工具返回结果不完整导致 Agent“幻觉”
这里说的幻觉不是模型编造内容,而是模型拿到的工具返回结果本身缺字段。比如天气接口返回了温度、风力,但没返回湿度,模型可能就会根据已有信息“推断”一个湿度,看起来有理有据,实际是错误的。
解决办法是在工具返回内容里,明确标注哪些字段值不确定。我在注册中心给每个工具增加一个return_schema字段,执行器在调用完函数后,对返回结果做校验,缺失的字段自动补上"unknown": true的标记。这样模型在处理时能够识别“该字段未知”,不会强行补全。这一步是血泪教训换来的,初期没有校验时,Agent 在回答里一本正经地编造了库存数量,直接被业务方抓包。
4.3 上下文被工具返回结果撑爆
事后来看这是最经典的坑。我的工具返回结果如果是个大 JSON,一次性全塞给模型,几十轮之后上下文直接爆炸。虽然有三层降级策略兜底,但最好还是从源头控制。我现在规定:工具返回结果超过 2048 字符时,执行器自动调用一个压缩器,把大 JSON 转成关键字段列表。
压缩器本质上是再一次调用模型,让模型提取“对这个任务有帮助的字段并输出精简摘要”。听起来有点蠢,但这相当于让专家模型做了一次信息降维,效果远好于硬截断。实测中,数据库查询返回 5000 行的结果集,压缩后变成 300 字以内的提要,模型完全能理解核心信息。
4.4 多轮对话中的指代消解失效
“帮我查下上海的天气”“那边呢?”——第二句里的“那边”指代什么?模型是知道指代上海的,但如果我们在进入工具调用循环前就把上下文截断了,模型就拿不到“那里是上海”的信息。这会直接导致工具参数填错,用户问“那边”,Agent 默认填了北京。
这个问题的排查思路分两层:一是检查上下文管理时是否保留了指代消解所必需的前轮信息。摘要压缩策略会破坏指代关系,所以我会在摘要里刻意保留主实体信息(地名、商品名、订单号等)。二是兜底方案:在工具参数填充器里增加“指代未知时向上文检索”的逻辑,如果参数值看起来是代词(长度小于 5 且无实体特征),则回查前两轮消息补全。
4.5 常见问题速查表
| 现象 | 根因 | 解决方案 |
|---|---|---|
| 工具选择错误率高 | 描述缺少触发条件 | 改写描述为“当用户…时使用” |
| 日期参数乱填 | 模型对日期计算不可靠 | 参数注入层统一格式化 |
| Agent 编造工具返回值 | 返回字段不完整 | 执行器校验缺失字段并标记 unknown |
| 上下文爆炸 | 工具返回结果过大 | 2048 字符压缩 + 摘要降维 |
| 指代无法消解 | 上下文被摘要破坏 | 摘要保留主实体信息 |
| 请求频繁 429 | 并发无限制 | 信号量限流 + 指数退避重试 |
| 写操作多发 | 模型误解用户意图 | require_confirm 确认机制 |
5. 落地部署与稳定性保障
系统开发到上线之间还有一段路要走,我把它分成三个关键动作:预发布压测、回归脚本、门店部署。
5.1 预发布压测
Agent 系统的压测跟普通接口不一样,不能只看 QPS,更要看“多流程成功率”。我写了一个模拟脚本,把 100 个典型用户消息按时间段投喂到系统里,统计三种结果:完整成功完成、中途模型修正后成功、失败。这个成功率指标才是 Agent 真正可靠性的度量。
压测中我发现一个有意思的现象:工具调用链越长,失败率越高。单次工具调用的成功率如果是 90%,两次调用就是 81%,五次调用只剩 59%。解决方案有两个方向,一是尽量简化流程设计,把能合并的查询合并,减少中间的模型往返;二是增加“失败自动重试一次”机制,实测中一次重试可以把五次链路成功率从 59% 拉回到 82%。
5.2 每轮会话状态可回溯
Agent 系统最难调试的问题是“同样的输入,不同的输出”。模型有随机性,这没问题,但出问题时你必须有办法复盘。Agent-Reach 做了一个 Session 日志系统——每一轮交互、每次工具调用的请求参数和返回结果、每次模型的原始输出,全部记录到 MongoDB 里,附带一个全局唯一的 trace_id。
排查时直接按 trace_id 查全链路,从用户输入到最终回答,每一步的中间状态都看得见。刚开始觉得这个设计多余,真正排查线上问题的时候才发现,没有它只能靠猜。
5.3 灰度发布与模型回退
大模型接口的升级是我们无法控制的,你可能会在某次更新后发现模型对tool_call的输出格式改变了。我遇到过一次 GPT-4o 更新后,模型的 tool_calls 参数里name字段多了几个空格,直接导致注册中心查找失败。所以针对模型输出的所有字段,我在解析层统一做了一次strip()清洗,这个习惯强烈建议保留。
同时保证模型版本可回退:给每个用户会话标记该会话使用的模型版本。如果新模型在灰度期间表现异常,可以通过一个后台开关一键切换到旧模型,不用改代码。
6. 一个完整的执行案例复盘
拿一个用户实际走过的流程举例:用户说“帮我对比一下上海和杭州今天的最低气温,然后告诉我哪里更冷。”
调度器解析后先调用天气工具查询上海,拿到结果后接着查询杭州,然后模型对比两处的数值,最后生成回答。整个过程涉及 2 次工具调用和 2 次模型生成。这里有一个细节:模型第一次拿到上海气温后,会认为还需要查杭州才发起第二次工具调用,而不是把上海的结果直接作为最终答案输出。这说明工具循环设计是有效的,模型没有“偷懒”提前结束任务。
这个案例看似简单,但已经把 Agent 的核心流程完全覆盖了:意图理解、参数抽取、工具调用、结果观察、二次决策、自然语言合成。你把任何一个环节拆开,都能定位到前面讲的某个模块和机制。
我在实际部署中还遇到过模型返回空参数的情况——比如用户只问“上海天气”,模型在调用时可能给 date 赋了一个today字符串,但 JSON Schema 里 date 的默认值是today,注册中心在填充时发现缺省就引入了默认值,所以没出问题。如果参数 schema 里没有默认值,就需要在工具执行时自己判断是否必需参数缺失。
再分享一个数据层面的细节。Agent 执行完工具调用后,工具返回结果一般是原始 JSON,这个 JSON 里的字段可能很多,比如天气接口可能返回空气质量、紫外线指数、湿度、风向等一大堆。模型如果看到这么多字段,它不一定知道哪些对用户重要,有时候会把无关的信息也放进回答里。我的处理方案是:在工具返回前,利用注册中心的 return_schema 定义“主回答字段”和“备选字段”,主回答字段保留在结果中,备选字段压缩到扩展信息里。模型优先基于主回答字段生成,需要时可以看扩展信息。这个策略有效提升了回答的简洁性和准确度。
写到这里,Agent-Reach 从设计到落地的完整路径已经讲得很清晰了。最后说一点个人体会:Agent 系统本质上是一个复杂的分布式决策系统,它的难点不在模型调用本身,而在于工程化的状态管理、容错设计和工具交互。不要指望模型能解决所有问题,你真正要做的是把系统边界梳理清楚,让模型在它擅长的地方发挥价值,把不擅长的地方牢牢封住。
如果你也正在折腾类似的 Agent 项目,建议从最小工具集跑起,先把 5 个以内的工具链路打通,再逐步扩展。工具数量上去之后,你会发现架构设计的重要性远超过模型选型。希望这篇内容对你走通自己的 Agent 系统有实质性的帮助。