1. 项目缘起为什么需要 Agent-Reach做了这么多年 AI 应用我一直觉得大模型的能力被低估了。ChatGPT 们确实能写文章、改代码、做翻译但真正到了落地环节你会发现它更像一个“坐在办公室里的顾问”——你说什么它答什么你不说它就不动。可现实世界的任务往往是需要主动触达、反复确认、多方协作的。比如让 AI 帮我查一下某家公司的工商信息再对比三家供应商的报价最后起草一封邮件发出去——这种“要干活”的需求传统对话式 AI 根本接不住。Agent-Reach 这个项目就是冲着这个痛点去的。它的核心目标只有一个让大模型从“被动应答”变成“主动执行”。说得更直白一点我要构建的是一个具备“手”和“眼”的智能体——它不仅能理解你的意图还能自己调用搜索引擎、读取数据库、操作第三方工具甚至按照流程一步步完成任务最后把结果整理好交给你。整个过程你只需要下指令剩下的事情它自己跑。这个项目适合谁参考如果你正在做 AI 应用开发或者公司里已经有业务需要接入大模型但苦于“只会聊天”又或者你想深入了解 Agent 架构、工具调用、上下文管理这些工程细节那这篇文章应该能给你不少启发。我会把整个设计思路、实现过程、踩过的坑以及最后的评估结果原原本本拆给大家看。先交代一下背景Agent-Reach 是在 2024 年底启动的最初只是一个内部试验项目后来逐步演变成一套可以对接实际业务的智能体框架。整个系统基于 Python 构建核心依赖有 LangChain、FastAPI、Redis、PostgreSQL模型层用的是国内可稳定调用的商用大模型 API具体厂商不展开各家能力差距不大。项目名称里的“Reach”取的是“触达”的意思——我们希望这个 Agent 能触达一切外部系统真正做到手眼并用。2. 整体设计与架构思路2.1 为什么不能直接调 API 完事很多人一听到“做个 Agent”第一反应是那不就是写个循环让模型自己决定调哪个函数吗理论上没错但实际做起来远没那么简单。直接调模型 API 的话你会很快遇到几个实际问题。第一模型输出不稳定。你让它“调用搜索工具查一下最新的 AI 新闻”它可能返回一段 JSON也可能返回一堆废话甚至可能一本正经地说“我已经帮你查了”——实际上什么都没查。如果不对输出做结构化约束和校验Agent 就是个摆设。第二多步任务的状态管理。比如用户要求“先查 A 公司的信息再找三家竞争对手最后生成对比表格”这个过程中间有依赖关系第二步要用第一步的结果第三步又要用第二步的结果。如果你只是机械地循环“模型输出 - 调用工具 - 再喂给模型”那上下文里塞满了中间结果很快就把 token 窗口撑爆而且模型会越聊越糊涂。第三工具调用出错时的恢复机制。外部 API 总有不稳定的时候搜索引擎可能超时数据库可能锁表。如果 Agent 没有重试、降级、纠错的机制一个工具挂了整个任务就断了。所以 Agent-Reach 从一开始就定了一个原则不要把 Agent 做成一个纯“模型的壳”而是要做一个有干预能力、状态可控、可观测的调度系统。模型负责“思考”系统负责“执行”两者之间有一层标准化的协议在撑着。2.2 整体架构里的四个角色Agent-Reach 的架构可以拆成四个核心模块各司其职第一个是意图理解层。负责解析用户指令把自然语言转成结构化的任务描述。比如用户说“帮我盯一下友商的官网有更新就发到群里”这个模块要能识别出这是“周期性监控任务”拆出“目标URL”“更新检测”“通知渠道”三个要素。这里我用了两阶段的处理方式先用小模型做意图粗分类再交给大模型生成 JSON 格式的任务参数兼顾速度和准确率。第二个是任务编排层。这是整个系统的心脏。它把复杂的用户请求拆解成一个 DAG有向无环图式的步骤流每一步要么是“调用哪个工具”要么是“让模型推理一下”要么是“把上一步的结果传给下一步”。这一步我们用了类似 LangGraph 的思路但更轻量因为核心业务里没有太多图遍历的需求一个按依赖顺序执行的有向图就够了。第三个是工具接入层。它管理所有 Agent 可以调用的外部能力比如 web 搜索、网页抓取、数据库查询、API 调用、文件读写甚至 Slack/企微通知。每个工具都注册成统一的接口格式输入参数、输出结构、错误类型。这样模型不需要关心底层实现细节只要知道“这个工具能做什么、参数是什么、返回什么”。第四个是记忆与上下文层。这层负责三件事短期记忆当前任务的中间状态、长期记忆用户的偏好、历史交互、以及上下文压缩把过长的对话历史摘要化。没有这层Agent 多轮对话必崩这是踩了无数次坑之后的血泪结论。2.3 关键设计决策让模型“格式化输出”而不是“自由发挥”我在 2.1 里提到过模型输出不稳定这是整个项目里最让人头疼的问题。后来我们的解法是所有模型输出必须走 JSON 格式并且在 system prompt 里明确告诉模型你只能输出 JSON输出里只允许包含type、content、tool_name、tool_args四个字段绝不输出任何多余的文字。这个约束听着简单但实际执行时还有很多细节。比如有些模型对 JSON 格式的理解不够严格会输出“type”: “tool_call”这种带全角引号的内容一解析就报错。所以我们写了一个容错解析器先尝试标准json.loads失败后会自动修正全角引号、去掉注释、截取 JSON 片段等把能救的尽量救回来。实测下来商用大模型配合这套约束输出结构合规率能稳定在 98% 以上。剩下的 2% 也不是死路我们会在校验失败后把解析错误信息连同原始输出一起喂回给模型让它“自我修正”一次。这个“主动纠错”机制比直接报错退出不知道高到哪里去了。3. 核心模块拆解与实现细节3.1 工具调用协议把每个工具变成“函数”工具接入层是整个 Agent-Reach 的基础设计得好后续扩展新工具几乎不用改业务代码。我们给每个工具都定义了一套标准的注册协议包含以下几个要素name工具的唯一标识比如web_search、fetch_url、query_databasedescription给模型看的自然语言说明描述这个工具能干什么、什么时候该用它parametersJSON Schema 定义说明参数名、类型、是否必填、枚举范围handler实际执行函数接收一个字典参数返回一个 JSON 可序列化的结果每次调用时Agent 需要按照参数规则把工具需要的参数传进来。为了让模型更容易生成正确的参数我会在 description 里附带示例“比如你可以这样调用web_search(queryA公司 工商信息, limit5)”。这个细节特别有用模型确实更倾向于模仿你给出的示例格式。一句话总结工具协议的核心是让“人机对话”变成“机机对话”。每个工具对模型来说就是一个有明确签名的函数模型不需要理解 HTTP、不需要知道 API key 怎么传只需要决定“调用哪个函数、传什么参数”。这就大大降低了模型出错的空间。3.2 上下文工程别让 Agent 迷路Agent 执行多步任务时最怕的是上下文被中间结果淹没。一开始我直接把搜索结果的原始 HTML 文本塞进对话记录结果第三步的时候模型已经不知道自己最初要干嘛了。后来的做法是这样的每个任务的完整上下文由三部分拼接而成任务描述 步骤状态 关键结果摘要。任务描述在一开始就固定下来不让它被后续内容覆盖。步骤状态维护一个 JSON记录“第几步已完成、第几步执行中、下一步要做什么”。关键结果摘要则是一条条精简过的信息每条最多 50 字由小模型负责从工具原始返回中提炼。比如用户要求“查 A 公司信息再找三家竞争对手”到了第二步时上下文里放的摘要不是一坨几千字的搜索结果而是“A公司成立于2015年主营智能仓储设备已完成B轮融资竞争对手暂未查询”。模型清楚知道自己在哪一步、已经知道了什么、接下来要做什么。这里我特别想强调一点摘要的质量直接影响最终结果的质量。第一次做的时候我图省事直接截取工具返回的前 500 字当摘要结果模型经常漏掉关键信息比如翻到第三页才出现的融资信息。后来改为让一个独立的摘要模型专门处理工具返回才把信息丢失率降下来。系统支出增加了一点但效果提升非常明显。3.3 记忆管理短期记忆和长期记忆分家记忆模块听起来简单但真正实现好要处理两个问题记忆的写入和记忆的召回。写入方面我们做了“关键事件记录”而不是“全量记录”。比如用户中途纠正了一次参数这是一个关键事件要记下来用户问了三次同一个问题说明他没听懂答案这也是关键事件可以触发解释句式的调整。这样做的好处是长期记忆不会越攒越乱。召回方面我们做了一个“记忆检索器”每轮对话开始时先用户的历史交互数据做向量相似度检索召回最相关的 3~5 条记忆片段拼进上下文中。这里我用的向量库是 Redis 的 RediSearch 模块对于几万条的规模完全够用没必要上 ES 或 Milvus。当然记忆管理也踩过坑。最典型的一个是把用户所有历史对话全部塞进上下文以为“记得越多越好”结果模型被大量无关信息干扰答非所问。后来做了“按相关性召回”而非“全部加载”之后这个问题就解决了。4. 完整实操从零搭一个 Agent-Reach 原型4.1 环境准备与依赖安装下面这个原型我尽量保持精简但保留了 Agent-Reach 的核心骨架。你可以把它直接跑通体验一下“指令 - 意图拆解 - 工具调用 - 结果汇总”的完整链路。建议在 Python 3.10 的环境下运行。创建一个项目目录然后用 pip 安装依赖pip install fastapi langchain langchain-openai redis openai如果你没有 Redis 环境可以先去掉记忆模块用内存字典写一个简单的替代实现等需要时再切 Redis。4.2 核心代码骨架让 Agent 会思考、能动手我们先定义工具的注册协议和内置的两个工具一个是“假搜索”模拟外部搜索一个是“假数据库查询”模拟查内部数据。生产中把这两个函数的实现替换成真实调用即可。import json import time from typing import Callable, Dict, Any # ---------- 工具注册 ---------- class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict[str, Any]] {} def register(self, name: str, description: str, parameters: Dict[str, Any], handler: Callable[[Dict[str, Any]], Any]): self._tools[name] { description: description, parameters: parameters, handler: handler } def get_schema_prompt(self) - str: lines [] for name, info in self._tools.items(): lines.append( f工具名: {name}\n f描述: {info[description]}\n f参数: {json.dumps(info[parameters], ensure_asciiFalse)}\n ) return \n.join(lines) def call(self, name: str, args: Dict[str, Any]) - Any: if name not in self._tools: raise ValueError(f未知工具: {name}) return self._tools[name][handler](args) registry ToolRegistry() # 模拟搜索引擎 def fake_search(args: Dict[str, Any]) - str: query args.get(query, ) time.sleep(0.5) # 模拟网络延迟 return f【模拟搜索结果】关于“{query}”的资讯A公司发布了新产品B公司获得融资C公司计划扩张。 registry.register( nameweb_search, description搜索互联网信息。当用户需要了解最新新闻、公司信息、行业动态时使用。, parameters{query: {type: string, description: 搜索关键词, required: True}}, handlerfake_search ) # 模拟查询内部数据库 def fake_db_query(args: Dict[str, Any]) - str: table args.get(table, ) time.sleep(0.3) return f【模拟数据库】表 {table} 查询到 3 条记录记录1、记录2、记录3。 registry.register( namequery_database, description查询内部业务数据库。当用户需要客户数据、订单、库存等结构化数据时使用。, parameters{table: {type: string, description: 表名, required: True}}, handlerfake_db_query )跑一个测试调用验证注册链路是通的result registry.call(web_search, {query: AI Agent 发展趋势}) print(result)4.3 Agent 主循环拆解指令、调用工具、汇总结果接下来是实现 Agent 的核心循环。它的工作方式是把“工具清单 用户指令 历史上下文”拼成一个 prompt 发给模型让模型输出一个 JSON。如果是tool_call就执行工具调用的 handler把结果追加到上下文再继续循环如果是final_answer就说明任务完成汇总输出给用户。from openai import OpenAI client OpenAI() # 生产环境替换为你的模型配置 SYSTEM_PROMPT_TEMPLATE 你是一个任务执行助手。你的工作方式是 1. 你需要根据用户的指令决定调用哪个工具或者直接回答。 2. 你只能输出一个 JSON 对象不要输出任何其他文字。 3. JSON 对象必须符合以下格式之一 - 当需要调用工具时{{type: tool_call, tool_name: 工具名, tool_args: {{...}}, reason: 简要说明为什么调用这个工具}} - 当所有工具调用完成、可以回答用户时{{type: final_answer, content: 你的回答内容}} 可用工具如下 {tool_schema} 请确保所有输出都是合法的 JSON。 def run_agent(user_query: str, max_iterations: int 5): messages [ {role: system, content: SYSTEM_PROMPT_TEMPLATE.format( tool_schemaregistry.get_schema_prompt() )}, {role: user, content: user_query} ] for step in range(max_iterations): response client.chat.completions.create( modelgpt-4o-mini, # 替换为你的实际模型 messagesmessages, temperature0.2, response_format{type: json_object} ) raw_output response.choices[0].message.content parsed json.loads(raw_output) if parsed[type] tool_call: print(f[Step {step1}] 调用工具: {parsed[tool_name]}原因: {parsed.get(reason, )}) tool_result registry.call(parsed[tool_name], parsed[tool_args]) # 将工具结果以用户消息形式追加回上下文 messages.append({ role: user, content: f[工具 {parsed[tool_name]} 返回结果]\n{tool_result} }) elif parsed[type] final_answer: print(f[Step {step1}] 任务完成输出最终结果。) return parsed[content] else: raise ValueError(f未知输出类型: {parsed}) return 已达最大迭代次数任务可能未完全完成。 # 测试运行 print(run_agent(帮我搜一下 A公司 B轮融资 的最新消息))这里面有几个关键点值得注意。第一response_format{type: json_object}是一个强有力的约束大部分 OpenAI 兼容协议的服务商都支持这个参数能极大提高 JSON 输出的规范性。第二工具结果是以“用户消息”追加进上下文而不是直接塞进 system这样模型能区分“这是新的事实输入”而不是“系统指令”。第三max_iterations是一个保险丝防止 Agent 陷入死循环无限调用工具生产环境我会设置成 8~10 步再长就要考虑是不是任务拆解出了问题。4.4 用 FastAPI 包一层 HTTP 服务既然是为了对接实际业务放内存里跑肯定不行。我用 FastAPI 把 Agent 包成了一个 HTTP 服务方便其他系统调用。每路请求进来后创建独立的会话 ID用 Redis 存储对话历史实现多轮上下文保持。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): query: str session_id: str default app.post(/agent) def agent_endpoint(req: QueryRequest): result run_agent(req.query) return {session_id: req.session_id, answer: result}启动服务之后你可以用 curl 直接测curl -X POST http://127.0.0.1:8000/agent \ -H Content-Type: application/json \ -d {query: 查一下公司数据库里的客户表, session_id: test001}响应里返回的就是 Agent 最终的答案。这一步做完Agent-Reach 的 MVP 就通了一条自然语言指令进来Agent 自己决定调什么工具、怎么组合结果最终输出一个完整的答案。5. 评估与调优怎么知道 Agent 真的变强了5.1 建立评测集没有评测就没有优化很多做 Agent 的人忽略评测都是“感觉差不多就上线了”。但“感觉”是最靠不住的。Agent-Reach 项目里我花了不少力气搭了一套评测机制主要分三类指标第一类任务完成率。给 Agent 布置一批标准任务看它最终是不是给出了正确结果。任务从简单到复杂分成三个难度等级单工具调用、多工具顺序调用、需要条件判断的多分支任务。第二类步骤有效率。统计 Agent 每次工具调用的价值这次调用是必需的还是白白浪费 token有没有绕弯路这个指标能直观反映任务编排层写得好不好。第三类格式合规率。每次模型输出的 JSON 是否一次解析通过。这个指标直接暴露 Prompt 写的稳不稳模型换版本时尤其要盯紧。拿这套指标去回归测试每次优化 Prompt 或调整编排逻辑都能看出真实变化而不是靠脑子“觉得变好了”。5.2 实测环节跑一组真实任务下面我们看一组实测数据任务是“查找 A 公司的工商信息并找出其 3 个竞争对手”。这是个典型的多步任务需要先搜索、再判断、再搜索。跑完输出的日志大致是[Step 1] 调用工具: web_search原因: 需要获取A公司的基本信息。 [Step 2] 调用工具: web_search原因: 基于A公司的主营业务寻找可能的竞争对手。 [Step 3] 任务完成输出最终结果。最终的回答是A公司成立于2015年主营业务是智能仓储机器人已完成B轮融资。 其竞争对手主要包括B公司主营AGV、C公司主营机械臂分拣、D公司主营仓储管理系统。这一步完成得漂亮。但有对比才有伤害。同样的任务有一次模型只调用了一次搜索工具然后直接靠训练数据里的“记忆”编造了竞争对手导致结果里出现了一个不存在的公司。这个案例非常典型说明工具调用次数不是越多越好但也不是越少越好关键是模型要知道“什么时候该用工具”。后来我们在 Prompt 里加了一句强约束“当你无法确认真实信息时必须调用搜索工具获取最新资料不允许直接基于预感回答”这个问题才基本解决。5.3 调优的三板斧Prompt、参数、容错调优过程里我用得最多的三板斧值得分享一下。第一板斧是 Prompt 迭代。每次发现 Agent 在某个环节表现不好先别急着改代码先看 Prompt 里的描述是否足够明确。比如“不允许编造信息”这种描述太虚改成“当你不知道答案时必须搜索后再回答如果搜索仍无法得到答案请明确告知用户‘未找到相关信息’”就具体多了。第二板斧是模型参数调整。temperature对 Agent 的影响极大。做创意写作可以用 0.9但做工具调用、JSON 输出我基本锁死在 0.1~0.2。温度太高你会发现模型偶尔“灵光一闪”输出格式不规范的 JSON或编造一个不存在的工具名。把温度降下来稳定性和可复现性都上来了。第三板斧是容错设计。即使上面都做了模型也不是 100% 听话。一次它输出了一个不存在的工具名web_search_engine解析器直接抛 “未知工具” 异常整个任务就断了。后来我加了一套模糊匹配如果工具名不在注册表里就先做一次相似度比对找到最接近的真实工具并提示模型“你是不是想调用 web_search”让它重新确认。这个机制把类似错误的存活率降到了 1% 以下。5.4 数据说话评测结果汇总跑到第三周时评估集上的数据终于像样了。简单任务完成率 96%中等任务完成率 78%复杂任务 61%。格式合规率稳定在 98% 以上。步骤有效率从最初的 62% 提升到 84%意味着平均每个任务少调了 1~2 次没必要的工具。这个成绩谈不上完美但对于一个 MVP 项目来说已经是可用的状态。最让我欣慰的是评测集真的指出了优化的方向——比如复杂任务容易卡在条件分支的判断上后来把“分支决策”单独抽出来做成一个带格式约束的推理步骤完成率立刻涨了近 10 个百分点。没有数据支撑这些改进只能靠瞎猜。6. 常见问题与避坑指南6.1 模型输出解析失败怎么处理这是遇到最多的问题没有之一。即使加了response_formatjson_object偶尔还是会拿到残缺 JSON。我的处理方案是写一个三级容错解析器。第一级直接json.loads。第二级如果失败尝试用正则提取 JSON 片段再解析因为模型有时会在 JSON 前后加一些解释性文字。第三级把解析错误信息加上模型原始输出一起喂回给模型“你之前的输出格式有误{error}。请重新输出只输出合法 JSON。”实测下来第三级自救成功率大概在 70% 左右大部分情况下模型能意识到自己错了并给出合规输出。6.2 工具参数总是传错怎么办有两种典型的传错一种是参数名字写错比如工具定义的是query模型输出的是keyword另一种是参数格式不对比如需要数字传了字符串。名字写错的解法很简单在 Prompt 里的工具描述中把最容易被“同义转写”的词都列出来。比如query后面加一句“也可理解为 keyword、search_term、关键词”。格式不对的解法是加一层类型转换器在调用工具之前先按 JSON Schema 的规则做一次参数校验和类型强转。这个转换逻辑放工具调用入口统一处理能拦截掉大部分低级错误。6.3 Agent 陷入死循环怎么破最极端的情况是模型发现自己“信息不够”就一直调用搜索工具但搜索回来的结果又让它不满意于是再搜……直到 token 烧完。对策主要有两个第一个是max_iterations硬上限超出后强制返回“任务超时建议简化指令”第二个是重复检测如果模型连续两次调用同一个工具且参数完全一样就判定为“没有进展”打断它并把“已有信息摘要”再加到上下文里明确告诉它“你已经有了这些信息请基于现有信息作答不要重复搜索”。6.4 上下文爆炸怎么控制多轮任务里工具返回结果越攒越多迟早超过模型窗口。我的处理方案是“分层摘要”每执行完一步工具调用就对全部上下文做一次压缩把旧的工具结果压缩成不超过 200 字的摘要只保留最近两步的完整细节。这个策略的整体效果是30 轮以内的多轮对话token 消耗基本稳定在 5k 以内。代价是细节有轻微损失但换来的稳定性完全值得。6.5 常见问题速查表问题现象可能原因解决方案模型不调用工具直接编答案Prompt 里对“必须搜索”的表述不够强添加强制约束“不确定时必须搜索”工具结果的 JSON 里有非法字符部分模型对 JSON 的严格性不足容错解析器逐级修复Agent 反复调用同一个工具对已有结果不满足陷入循环加重复检测打断循环上下文越来越长响应变慢历史记录未做压缩分层摘要机制只保留最近两步多轮对话上下文丢失会话状态未持久化接入 Redis 保存 session 状态复杂任务失败了 50% 以上任务拆解粒度太粗把分支决策独立成专用推理步骤7. 写在最后的经验之谈Agent-Reach 做到现在我最深的体会是Agent 系统的难点不在于模型本身而在于工程兜底。模型的“智能”是上限但系统的“可靠性”才是下限。没有评测集、没有容错机制、没有上下文管理的 Agentdemo 演示时惊艳全场一上真实业务立刻现出原形。另外有一个小技巧想分享给大家调试 Agent 时不要只看最终输出一定要保留每一步的日志——模型当时调用什么工具、为什么调用、拿到了什么结果、下一步基于什么信息决策全部记下来。你会发现大部分问题在翻日志时一眼就能定位根本不用瞎猜。Agent-Reach 项目里我在日志系统上花的功夫回报率远超任何一次 Prompt 优化。最后再补充一个扩展方向。现在 Agent-Reach 的执行链路是“串行为主、少量并行”但真实业务里很多步骤没有强依赖完全可以并行调度。比如“查三家公司信息”这三路搜索互不干扰如果改成并发调用整体响应时间可以压缩一半以上。我后续正在尝试把任务编排层升级成支持并行分支的执行器到时候单独写一篇分享。希望这篇文章能给你一些启发如果你想做 Agent别急着堆功能先把评测、容错、上下文这三件地基打牢。