做了半年AI Agent,我把踩过的坑和最终跑通的方案写下来。如果你正打算从零搭建一个自己的智能体,或者已经在折腾却总感觉差一口气,这篇内容应该能让你少走不少弯路。
先说清楚“Agent-Reach”是个什么东西。它的名字拆开看就挺直白:Agent是智能体,Reach是触达与覆盖。实际做下来,这套项目解决的核心问题就是“怎么让一个带目标的AI Agent真正把事情办完”,而不是聊到一半就断掉,或者答非所问地把流程带偏。它能做的具体事情包括:根据用户输入自动拆解任务、调用外部工具完成搜索或数据获取、把多轮对话的状态管理起来、最终输出结构化结果。适合三类人参考:想入门智能体开发的开发者、正在做自动化办公流程的产品经理,以及打算在团队内部落地AI工具的技术负责人。
下面是我从需求拆解、架构设计到最终落地全程的完整复盘。
1. 为什么做Agent-Reach:先想清楚要解决什么问题
1.1 项目定位与核心需求
我一开始做的那个Agent简直是灾难。给它一个简单任务,比如“帮我查一下最近三天某产品的用户反馈并整理成摘要”,它会先泛泛地说一堆“好的,我来帮您搜索相关信息”,然后就没有然后了。要不就是搜到一半突然开始聊它自己多聪明。这不是模型不行,是我压根没想清楚要做一个什么样的系统。
Agent-Reach立项时我给自己定了三条硬性需求:
- 任务必须能自动拆解。用户给一个模糊目标,系统内部要把它拆成可执行的子任务,而不是让模型凭感觉自由发挥。
- 工具调用必须可控。Agent不能想调什么就调什么,工具的注册、参数校验、结果回传都要有明确边界。
- 状态必须可追踪。每一轮对话、每一次工具调用、用户上下文的变化,都要有记录,否则长对话场景下必乱。
这三条写下来之后,“Agent-Reach”这个名字自然就出来了:Agent承担智能分析推理,Reach强调对结果的最终触达——每个任务都要能走到头。
1.2 方案选型:为什么不用现成的LangChain/CrewAI
立项的时候团队里有人提议直接用LangChain或者CrewAI,我的看法是不完全反对,但也不能无脑用。
现成的Agent框架确实省事,它帮你把工具调用、记忆管理、任务编排都封装好了。但问题恰恰出在这层封装上:一旦出现诡异Bug,你根本不知道是框架的问题还是自己逻辑的问题。我在测试阶段遇到过工具参数连续传错的情况,最后发现是框架内部的callback机制和自己代码的冲突,这类问题排查成本极高。
所以Agent-Reach的核心逻辑我全部手写,框架只作为底层模型调用的适配层。这样的好处是每一行逻辑都在自己掌控下,出任何问题都能顺着代码追下去。坏处是前期开发量确实大,但做工具的都知道,可控性大于一切。
具体选型如下:
- 核心语言:Python 3.10+
- LLM接入层:OpenAI SDK的异步接口,兼容本地部署的模型服务
- 工具注册机制:Python装饰器+函数签名自动解析
- 持久化存储:SQLite存会话状态,JSONL存完整交互日志
1.3 整体架构的演进过程
第一版我是按单Agent思路做的,所有逻辑塞在一个“大脑”里。跑起来就发现问题:任务一复杂,上下文窗口就不够用,而且职责混乱,用户问题解析、工具调用决策、结果校验全混在一起。
随后第二版我引入了“规划器+执行器”的双层结构。规划器负责理解用户目标、产出子任务列表;执行器负责逐个完成子任务并回传结果。这次跑通了很多场景,但仍有一个痛点:子任务之间如果有依赖关系,执行起来容易串行阻塞。
最终版借鉴了过去做微服务时的路由思路,把每个子任务当作一个独立的路由目标。规划器只产出任务描述和预期输出格式,不在过程中干预具体执行。这版才算真正跑顺了,代码结构也清爽:planner.py管思维编排,executor.py管工具调用,memory.py管状态记忆。
2. Agent-Reach的核心设计:一个能跑通的Agent骨架
2.1 工具注册机制:让Agent学会“使用双手”
Agent本质上是一个会思考的对话系统,但只有思考没有工具,它就只能纸上谈兵。Agent-Reach里的工具注册我完全模拟了微服务里API网关的模式。
# tools/registry.py from typing import Callable, Dict import inspect import json class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict] = {} def register(self, name: str, description: str): def decorator(func: Callable): signature = inspect.signature(func) params = [ { "name": p.name, "type": str(p.annotation), "required": p.default is inspect.Parameter.empty } for p in signature.parameters.values() if p.name != "self" ] self._tools[name] = { "function": func, "description": description, "parameters": params } return func return decorator def get_tool_schema(self): return [ { "name": name, "description": tool["description"], "parameters": tool["parameters"] } for name, tool in self._tools.items() ] def execute(self, name: str, **kwargs): if name not in self._tools: raise ValueError(f"Tool {name} not found") return self._tools[name]["function"](**kwargs) registry = ToolRegistry() @registry.register( name="search_web", description="搜索公开网页内容,返回与查询词相关的文本摘要" ) def search_web(query: str, top_k: int = 5): # 实际场景中可替换为搜索API或爬虫实现 results = [] # ... 具体实现省略 return results @registry.register( name="calculate", description="执行数学计算,支持四则运算和括号" ) def calculate(expression: str): try: result = eval(expression, {"__builtins__": {}}, {}) except Exception as e: return f"Error: {str(e)}" return str(result)这里有个关键点:我用inspect.signature自动函数签名解析替代了手写参数说明。过去在别的项目里需要为每个工具手写JSON Schema,维护起来比工具本身还累。现在写一个函数就自动生成Schema,Agent调用工具时能拿到精确的参数名和类型。实测下来,这套机制的精度和效率都很稳定。
2.2 规划器设计:任务拆解的逻辑
规划器解决的问题是:用户说了一句话,怎么把它变成一系列可执行步骤。这一步做不好,后面全乱。
我一开始尝试过让模型直接输出JSON格式的任务列表,效果不稳定,因为模型偶尔会漏掉任务或凭空创造不存在的任务。后面改成两步走:
- 先将用户输入解析为“意图+约束条件”的结构体
- 再基于意图匹配预置的任务模板,模板中有明确依赖关系
比如用户说“帮我调研一下最近三个月AI编程工具的排行,最好找有数据来源的资料”。第一步解析出意图是“调研排行”,约束是“三个月内”“需要数据来源”。第二步匹配到“调研型任务”模板,模板里定义了搜索、筛选、归纳三个子任务,且后两个依赖前一个的结果。
这比让模型自由生成任务列表可靠得多。因为预置模板保证了任务结构的完整性,模型只负责往模板里填具体内容,而不是从零开始想结构。
# planner.py 核心逻辑片段 from dataclasses import dataclass, field from typing import List @dataclass class Task: task_id: str description: str dependencies: list = field(default_factory=list) status: str = "pending" result: str = "" class Planner: def __init__(self, llm): self.llm = llm def create_tasks(self, user_input: str) -> List[Task]: # 第一层:解析意图和约束 parsed = self._parse_intent(user_input) # 第二层:从模板库中匹配任务结构 template = self._match_template(parsed["intent"]) tasks = [] for step_desc, deps in template.steps: task = Task( task_id=uuid4().hex, description=step_desc.format(**parsed["constraints"]), dependencies=deps ) tasks.append(task) return tasks def _parse_intent(self, user_input: str): prompt = f"""从用户输入中提取意图和约束条件。 用户输入:{user_input} 输出JSON格式,包含 intent 和 constraints 两个字段。""" # 调用LLM得到结构化结果 # 如果JSON解析失败,退回基于规则的匹配这里面有个细节值得提:退回策略。模型输出不一定每次都合法,如果解析失败就直接报错,用户体验会很差。我的做法是:LLM解析失败时,退回基于关键词的意图匹配。虽然不如前者聪明,但至少能把流程走下去。这个兜底设计在真实场景中救了不知道多少次。
2.3 执行器与状态追踪:Multi-Turn对话的根基
Agent系统绕不开的一个难题是多轮对话的状态管理。用户上一秒说“查一下A产品的资料”,下一秒说“顺便对比一下B”,如果系统不知道“A产品”指什么,第二个指令就会落空。
Agent-Reach里我为执行器配了一个结构化的Memory对象,负责维护三类信息:
- 当前用户目标(全局唯一的,不会被中间操作覆盖)
- 已完成任务列表(每个任务有状态和关键结果摘要)
- 用户偏好快照(比如用户偏好中文输出、需要详细来源等)
# memory.py class SessionMemory: def __init__(self, session_id: str): self.session_id = session_id self.goal = None self.completed_tasks = [] self.preferences = {} self.context = {} def set_goal(self, goal: str): self.goal = goal def add_completed_task(self, task: Task): self.completed_tasks.append({ "task_id": task.task_id, "description": task.description, "result_summary": task.result[:500] }) def to_system_prompt(self) -> str: """将记忆内容拼接到系统提示词中,给LLM提供上下文""" prompt_parts = [] if self.goal: prompt_parts.append(f"当前用户目标:{self.goal}") if self.completed_tasks: prompt_parts.append("已完成的任务:") for t in self.completed_tasks: prompt_parts.append(f"- {t['description']} => {t['result_summary']}") return "\n".join(prompt_parts)每个会话句柄会全程携带这个Memory,每次调用执行器前,都把Memory的内容注入系统Prompt。我实测过,这个做法比单纯把所有历史消息塞给模型要省钱省token。书读百遍其义自见,但Token烧了就是烧了,未必就能“其义自见”。
3. 实操过程:从零搭建Agent-Reach的完整记录
3.1 环境准备与项目初始化
先交代一下搭建环境,我用的是一台Linux服务器(配置大概4核8G),日常跑这个Agent系统没任何压力。如果你只是本地开发测试,Mac或Windows也能跑通,只要能正常安装Python依赖。
项目初始化步骤:
# 1. 创建项目目录 mkdir agent-reach && cd agent-reach # 2. 建虚拟环境 python3.10 -m venv venv source venv/bin/activate # 3. 安装核心依赖 pip install openai==1.30.0 pip install pydantic==2.7.0 pip install aiosqlite==0.20.0 # 4. 目录结构规划 mkdir -p agent_reach/{core,tools,memory,api} mkdir -p logs touch agent_reach/{__init__.py,main.py,config.py}依赖安装有个需要注意的地方:OpenAI SDK版本千万不能乱装最新版。1.x版本API风格变化大,你要是照着老教程写代码,大概率会碰到Response对象不一样的问题。我锁在1.30.0这个版本,是因为我测过它和自定义Agent循环的兼容性最稳,不追求新功能就别往上走了。
3.2 核心循环逻辑:Agent运行时的调度中枢
Agent-Reach的运行循环是整个系统的心脏,它的职责是让规划器、执行器、记忆系统各司其职地配合起来。
# agent_reach/core/agent_loop.py import asyncio class AgentReach: def __init__(self, planner, executor, memory_factory): self.planner = planner self.executor = executor self.memory_factory = memory_factory async def run(self, user_input: str, session_id: str = None): # 1. 获取或创建会话记忆 memory = self.memory_factory.get_or_create(session_id) # 2. 更新用户目标 memory.set_goal(user_input) # 3. 规划任务 tasks = self.planner.create_tasks(user_input) results = [] for task in tasks: # 4. 检查依赖是否已经满足 if not self._dependencies_satisfied(task, tasks): continue # 5. 执行任务(注入当前记忆作为上下文) task_prompt = self._build_task_prompt(task, memory) tool_results = await self.executor.execute(task_prompt) # 6. 更新任务状态与记忆 task.result = tool_results task.status = "completed" memory.add_completed_task(task) results.append(tool_results) # 7. 汇总最终答案 final_answer = await self._summarize_results(results, memory) return final_answer这套主循环看着简单,但调整了很多次才稳定。核心要点是每个任务在执行时,都从Memory里取“当前用户目标+已完成任务列表”作为上下文,这保证Agent干活时不会忘记最初的指令。用户问的是A,执行到子任务时被B带偏,这种情况我在早期版本遇到太多次了。
_build_task_prompt方法里还有一层细节:不是把全部上下文塞进去,而是只取前两轮的关键信息,并明确拼接当前任务描述。太长反而干扰模型判断,啥都给等于啥都没看。
3.3 工具实战:让Agent接入搜索和计算
为了让Agent不是空壳子,我实现了一个模拟搜索工具、一个真实可用的计算工具,以及一个日期时间工具。这三个工具覆盖面够了,能跑通常见的“查信息+算数据+看时间”组合场景。
日期工具的实现细节值得一说,Agent经常需要知道今天是几号来回答“近三天”这类相对时间问题。我通过工具返回精确日期,比让模型自己猜要靠谱得多。
# tools/others.py import datetime @registry.register( name="get_current_date", description="获取当前日期和星期,格式如2025-02-14 星期五" ) def get_current_date(): now = datetime.datetime.now() weekdays = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"] weekday = weekdays[now.weekday()] return f"{now.strftime('%Y-%m-%d')} {weekday}"讲一个我在实际测试中经常用的场景,大家能直观感受这套系统的工作方式。用户发来一句话:
“帮我查一下今天深圳的天气,然后计算摄氏26度等于多少华氏度。”
Agent的规划器把这句话拆成两个子任务:查天气、算换算。先执行日期工具拿到当日日期,再用搜索工具查天气关键词,同时用计算工具执行算式。执行完毕,汇总模块把“天气数据”和“华氏度数值”拼成自然语言答案返回。这个场景虽然简单,但完整走通了“目标解析-任务拆解-工具调用-结果汇总”的链路,用来做系统验收特别合适。
3.4 会话接口:怎么把Agent暴露出去
光有一个核心循环还不够,得像Web服务一样给外界暴露访问入口,否则没法在真实业务里用。我用FastAPI封装了一个极简的HTTP接口:
# agent_reach/api/server.py from fastapi import FastAPI from pydantic import BaseModel import uuid app = FastAPI(title="Agent-Reach API") agent_instance = AgentReach(...) # 注入已初始化的实例 class ChatRequest(BaseModel): message: str session_id: str = None class ChatResponse(BaseModel): reply: str session_id: str task_count: int @app.post("/chat", response_model=ChatResponse) async def chat(req: ChatRequest): session_id = req.session_id or uuid.uuid4().hex reply = await agent_instance.run(req.message, session_id) task_count = len(agent_instance.memory_factory.get(session_id).completed_tasks) return ChatResponse(reply=reply, session_id=session_id, task_count=task_count) @app.get("/health") async def health(): return {"status": "ok"}接入Web前后的差异很大。之前写死命令行直接调用,很多边界情况测不出来。接上HTTP接口后,并发请求、重复session、超时这些真实业务里的问题全暴露了,对系统健壮性的提升非常明显。
4. 常见问题与排查技巧实录
4.1 工具调用失败的三个高频原因
把Agent-Reach放到更多人手里跑之后,反馈回来的问题集中在工具调用环节,这里列一下出现频率最高的三类情况。
第一类是参数格式错误。模型输出了字符串,工具期望的是数字类型,直接传进去必然报错。解决方案是在工具注册时加一层“类型校正”逻辑,用pydantic把传入值按Schema定义做一次强制转换,转换失败才真正报错。
第二类是超时无响应。搜网工具偶尔卡在某次请求上,几秒没返回会阻塞整个流程。解决思路是给每个工具调用加asyncio.timeout包装,超时就返回“工具超时,请重试或换一种方式”,不能让一个工具卡死全流程。
第三类是递归死循环。模型发现第一次工具返回的结果不满足条件,会尝试再补一次搜索;如果第二次结果仍不满足,它可能会继续调第三次、第四次……我在早期版本没加次数限制,某次测试中模型连调了11次工具,Token烧得心都在滴血。
解决办法是给执行器加一个循环上限(我设为最多5次),超过上限就停止工具调用,直接基于当前已有信息生成回答。这个机制上线后,再没出现“无限循环烧Token”的惨案。
4.2 Prompt设计的微妙之处:语气与格式的影响
实践下来发现,同一个Agent能力上限的30%取决于代码,70%取决于你给它的系统Prompt。
刚开始我写的系统Prompt过于简化:“你是一个智能助手,请完成用户的任务。”这种话等于没写。后续我在Prompt里加了三类关键信息:
- 角色定义:跨领域研究助理,在收到任务后先拆解再执行
- 边界约束:不能编造来源;如果工具没有返回数据,如实说明未找到,不得编造
- 输出偏好:中文回答,核心结论放开头,细节放在后段,引用工具结果时要标明来源
# agent_reach/core/prompts.py SYSTEM_PROMPT = """你是Agent-Reach智能助理,具备任务拆解和工具调用能力。 工作流程: 1. 根据用户目标拆解为具体步骤。 2. 不要跳过中间步骤,按依赖顺序执行。 3. 每个工具调用前,判断是否是完成当前任务所必需的手段。 4. 无论结果如何,最终都要整合信息给用户一个明确的回复。 约束: - 必须基于工具返回的数据回答;没有数据时明确告知用户,不要编造。 - 回答使用中文,在开头给出结论,再补充细节。 - 工具调用失败时不要反复重试同一个动作,最多更换策略后再试一次。"""这段Prompt上线后,整体回答质量肉眼可见地提升。尤其是“不要编造”和“反复重试”这两条约束,直接减少了模型幻觉和Token浪费。后来我又根据反馈微调过几个版本,比如在“工具调用失败时”那句后面,加上了“更换策略后再试一次”的引导词,模型的表现更稳定了。
4.3 上下文膨胀与Token成本:省钱技巧
Agent系统部署久了最容易面临的隐形问题就是上下文膨胀。因为你的Memory系统会往Prompt里拼接已完成任务摘要、用户偏好、工具执行记录等,时间一长总量非常可观。
实测数据:一次完整调研型任务,如果包含6个子任务,完成时的对话总Token消耗大约在12K-20K之间,其中Memory注入的部分约占总量的三到四成。这意味着你每跑一轮,都要为“过去”付钱。
省钱技巧有两个。
一个是摘要分层。不要直接拼接原始子任务结果,而是用一个“摘要模型”把每个结果压缩到100-200字,再注入Prompt。三级压缩后,上下文总量能省四成左右,而最终回答质量和直接拼接原文几乎无差。
另一个技巧是“只保留对当前目标有关键价值的信息”。我给Memory增加了一个“相关性打分”的步骤,每完成一个子任务后,先让规划器评估该结果和最终目标的相关度,相关度低于阈值的直接存进SQLite,不注入本轮Prompt。这让上下文总量进一步缩减,同时不失关键信息。
4.4 实战中的两个小技巧
第一个技巧:给工具函数名取一个语义明确的名字。工具名其实会直接影响模型的理解,search_web比execute_search_procedure_001效果好得多。模型看到清晰的名字,调用起来才有信心。
第二个技巧:所有工具的说明用中文,不要用中英混搭。虽然模型英文能力强,但这会让它在生成工具参数时出现莫名其妙的语言切换。统一用中文描述,成功率反而更高,尤其当参数说明中包含“这个参数表示……”这类语义描述的时候,中文表达更符合模型的任务上下文。
这两个技巧不是我拍的脑袋,是在排查模型调用习惯后发现的。有一次我连续跑五个场景,发现三个场景中模型都倾向于调用名称包含get_且描述简洁的工具,而不是那些描述冗长的工具。想提升调用率,就把工具说明减到三行以内,只保留最核心的信息。
Agent-Reach这套项目做下来,我最大的感受是:智能体开发的难点不在于“让模型变聪明”,而在于“把模型的能力有条理地引导到目标任务上”。规划、执行、记忆、工具,四个环节环环相扣,哪一环松了都跑不出稳定结果。所以如果你也在做类似的Agent项目,建议不要急着堆功能,先把这四块骨骼搭稳。骨架撑住了,皮肉是可以慢慢长的。以后有机会,我再把多Agent协作、人机审核介入这些扩展点展开聊聊。