AI Agent 开发工程师(二):核心组件深度解析——功能、工具、记忆
上一篇回顾为什么要把 Agent 拆成四要素上一篇我们把 Agent 定义为数学意义上的「思考 → 行动 → 观察 → 再思考」的 ReAct 循环并给了一张四要素表规划 / 工具 / 记忆 / 执行。这篇我们把四根柱子逐一打磨到能写代码的程度。读完你会得到两样东西对每个组件「到底是什么、存哪、怎么取」的精确理解一个不依赖任何框架、纯 Python 标准库就能跑起来的最小 Agent 骨架——把工具调用、短期记忆、ReAct 循环串成一个真正能执行任务的程序。信我写完这个骨架你回头再看 LangChain/LangGraph 的封装会突然觉得它没那么神秘——那些框架只是把这个骨架做得更健壮、更工程化。关于代码本文所有代码块均为完整可运行的最小示例Python 3.10只需pip install openai一次。为方便你在无 API Key 时也能看效果骨架里会同时给「真实调用」和「模拟调用」两条路。1. Function Calling模型到底是怎么「选中一个工具」的先解决最大的一个谜模型是大模型它怎么知道该调哪个工具、参数填什么1.1 它不是一个黑盒开关而是一次「带格式约束的补全」很多人以为 Function Calling 是模型内部发生了什么魔法开关。其实没有。它的本质是你把一组「工具的描述 参数的 JSON Schema」作为结构化文本和用户问题一起塞给模型模型在你给的这组候选里输出一个「它想调用的函数名 参数」的 JSON。换句话说工具列表就是提示词的一部分只是它有一套约定俗成的格式约定让模型输出机器可解析的结构化结果。模型的幻觉依然可能发生选错工具、填错参数所以才需要工程上的校验和兜底。看一个最简单的例子。我们定义两个工具importjson TOOLS[{type:function,function:{name:get_weather,description:查询某个城市的实时天气,parameters:{type:object,properties:{city:{type:string,description:城市名如 北京}},required:[city],},},},{type:function,function:{name:add,description:两个整数相加,parameters:{type:object,properties:{a:{type:integer},b:{type:integer},},required:[a,b],},},},]当你把TOOLS传给模型的tools参数并问「北京的天气怎么样」模型会返回类似{name:get_weather,arguments:{\city\: \北京\}}而不是直接回答「我不知道」——因为tools把它框定在了「你只能从这两个里选」的空间里。1.2 真正的函数执行循环最重要的一节关键来了模型负责「生成结构化的调用请求」——它输出一段 JSON我想调哪个函数 参数填什么而真正跑那个函数、拿到真实结果的是你的代码。完整一轮调用是这样用户问题 │ (1) 发给模型带着 tools 描述 ▼ 模型返回: { name: get_weather, arguments: {city:北京} } │ (2) 你的代码在 TOOLS 里 find 到 name 对应的真函数 ▼ 执行该函数拿到 { 城市:北京, 天气:晴, 温度:28 } │ (3) 把函数返回的结果作为新的消息再次发给模型 ▼ 模型拿到结果回答一句人话: 北京今天晴28 度。注意第 (3) 步函数调用的结果不是直接拼进用户回答而是作为一个roletool的消息回传让模型看到结果后再组织语言。这是 ReAct 的「观察」环节的落地。用消息时序图再看一遍这轮交互tool 消息回环是关键LLM工具函数LLM你的代码LLM工具函数LLM你的代码请求(messages tools 描述)assistant turn: tool_calls [{name, arguments}]根据 name 路由并执行 fn(**args)真实结果(天气/行情…)回传 roletool 消息(tool_call_id result)最终自然语言回答返回给用户下面给一个能把这三步真正跑起来的小函数用 OpenAI 兼容 API任何支持 tool calling 的模型都行importos,jsonfromopenaiimportOpenAI# 末尾可换自己的 key/base为演示留了哨兵clientOpenAI(base_urlos.environ.get(OPENAI_BASE_URL),api_keyos.environ.get(OPENAI_API_KEY))defget_weather(city:str)-str:# 真实场景这里去调天气 API演示返回固定值returnjson.dumps({city:city,weather:晴,temperature:28})defadd(a:int,b:int)-int:returnab NAME2FN{get_weather:get_weather,add:add}defrun_once(user_msg:str)-str:messages[{role:user,content:user_msg}]respclient.chat.completions.create(modelgpt-4o-mini,# 换成你的模型messagesmessages,toolsTOOLS,# 上面定义的工具列表)choiceresp.choices[0].messageifnotchoice.tool_calls:# 模型觉得不需要调工具直接给回答returnchoice.contentortcchoice.tool_calls[0]# 只取第一个工具调用多工具见下fnNAME2FN[tc.function.name]# arguments 是 JSON 字符串解析成关键字参数argsjson.loads(tc.function.arguments)resultfn(**args)# 关键把模型选择 函数结果都回传给模型让它观察后继续messages.append({role:assistant,content:None,tool_calls:[{id:tc.id,type:function,function:{name:tc.function.name,arguments:tc.function.arguments}}],})messages.append({role:tool,tool_call_id:tc.id,content:json.dumps(result)})replyclient.chat.completions.create(modelgpt-4o-mini,messagesmessages)returnreply.choices[0].message.contentor这段代码就是 tool calling 的最小可运行形态。你理解它就理解了 80% Agent 框架里invoke_tool/executor到底在干什么。1.3 多工具并行与 Tool Choice一个常见问题是「模型一次选了三个工具怎么办」。答案tool_calls是一个数组你可以循环逐个执行、把结果逐个回传。这对并行获取多个信息再综合的场景很关键——比如同时查天气、查机票再规划。而tool_choiceauto让模型自己决定调或不调tool_choicerequired强制至少调一次某些场景我们希望固定调用某个指定函数相当于路由容差可以用tool_choice{type:function,function:{name:get_weather}}。工程细节当tool_calls是数组时每条tool_call_message都要有独立的tool_call_id回传时roletool的消息就要带上对应的tool_call_id。一条错后端就报tool_call_id not found——这是新人最容易踩的坑。2. 记忆Agent 的「记事本」到底存哪、怎么取记忆是所有 Agent 工程里被低估最多、却又直接影响体验的部分。我们分两种来看。2.1 短期记忆会话上下文是什么当前这轮任务从开始到现在所有的消息历史messages数组。怎么办简单就是一直带着——把历史消息作为上文反复塞给模型。但它有个硬上限上下文窗口有限且越长越贵。所以工程化时会截断只保留最近 N 条把过老的对话丢掉摘要压缩超过阈值后让模型把前面的对话压缩成一段摘要用摘要替换旧上下文窗口之争控制在工具结果回传不撑爆窗口尤其长工具输出。一个极简的最近 K 条记忆deftrim_history(messages,max_len12):returnmessages[-max_len:]2.2 长期记忆向量 / 知识库跨会话、可检索的本子是什么Agent 不可能把每个任务都重新学一遍。它需要把跨会话、跨任务的经验与领域知识持久化并且基于相关度而不是顺序去取回。怎么办标准做法是嵌入Embedding检索——把一段经验/知识chunk起来送入 embedding 模型变成向量存进向量库如chromadb、faiss、milvus、或简单的numpy数组进来一个新问题时把问题也嵌入成向量用余弦相似度找最接近的 K 条把这 K 条当作上下文喂给模型。一个最小可跑、不依赖重型向量库的长期记忆用 numpy 算余弦相似度importhashlibclassTinyMemory:def__init__(self):self.items[]# 每个元素: (text, vec)def_encode(self,text:str):# 真实场景换成真实 embedding 模型这里用一个确定性的“伪向量”演示# 思路把文本哈希成 8 字节转成 8 维 0~255 的向量并归一化hhashlib.sha256(text.encode()).digest()[:8]# 8 字节raw[bforbinh]# 8 个 0~255nsum(x*xforxinraw)**0.5# 模长return[x/(n1e-9)forxinraw]# 归一化成单位向量defadd(self,text:str):self.items.append((text,self._encode(text)))defcos(self,a,b):returnsum(x*yforx,yinzip(a,b))defrecall(self,query:str,k:int3):qself._encode(query)scoredsorted(self.items,keylambdat:-self.cos(q,t[1]))return[tfort,_inscored[:k]]# 演示mTinyMemory()m.add(用户在巴西出差偏好清淡口味)m.add(项目上线要求周五前完成内测)m.add(本周会议记录放在公司 wiki)formatchinm.recall(下周上线安排,k2):print(match.strip())提示真正的工程里你会用chromadb这类现成库把add/recall换一行即可。这里手写是为了让你懂它背后就是「向量存起来 算相似度检索 K 条」。关键区分面试也常问维度短期记忆长期记忆存什么本轮会话上下文跨会话的事实/经验/知识结构消息数组向量 文本 chunk检索方式顺序按时间相似度 Top-K上限上下文窗口存储容量典型实现messages 列表chromadb / faiss / 数据库3. 规划模型「思考」的几种范式与取舍规划决定 Agent 的聪明程度。说几种主流3.1 CoT思维链让模型一步一步想把问题引导模型逐步推理而非直接给结论。工程上通常就是提示词里加一句请一步一步思考再给最终答案或让模型输出thought → answer结构。优点简单、几乎零成本、效果显著。局限不是真正的行动,没有工具一旦中间想错无法圆回。3.2 ReAct推理 行动交替上篇的主角Thought想这一步该怎么做→Action调用某个工具→Observation看到结果→ 循环。这是工具型 Agent 的默认范式。我们在第一节的循环代码就是这个范式的具体化。3.3 Plan-and-Execute先规划、再执行先让模型把大目标分解成一串子步骤Plan然后逐步走每步可能复用 ReAct 循环比 ReAct 更擅长需要长链路的任务且能让规划和执行分开监控。defplan_and_execute(goal,planner_fn,executor_fn):planplanner_fn(goal)# 让模型输出步骤清单results[]forstepinplan[steps]:results.append(executor_fn(step))# 每步单独执行 收集结果returnresults3.4 规划范式速查表范式适合场景不确定度CoT纯推理、不需要外部工具低ReAct单步调用工具、需要环境反馈中Plan-and-Execute长任务、多子目标、需全局顺序高工程建议先 CoT 到顶不行再加 ReAct任务明显多步骤时直接上 Plan-and-Execute别一上来就整最复杂的。复杂度也要成本 失败面。4. 执行与循环把四要素串成一个真正会跑的 Agent最后把第一节的函数调用、第二节的记忆、第三节的 ReAct 循环合成一个最小可用 Agent 骨架。它不带任何框架只做三件事有工具能调、会记上下文、循环直到结束。importjsonfromdatetimeimportdatetimeclassMiniAgent:def__init__(self,model_client,tools,name2fn,max_rounds6):self.clientmodel_client self.toolstools self.name2fnname2fn self.messages[]# 短期记忆消息历史self.max_roundsmax_roundsdefchat(self,user_msg:str)-str:self.messages.append({role:user,content:user_msg})forround_noinrange(self.max_rounds):respself.client.chat.completions.create(modelgpt-4o-mini,messagesself.messages,toolsself.tools)msgresp.choices[0].messageifnotmsg.tool_calls:# 没有工具要调 收敛return 最终回答finalmsg.content self.messages.append({role:assistant,content:final})returnfinal# 有工具调用 逐个执行并回传self.messages.append(msg)fortcinmsg.tool_calls:argsjson.loads(tc.function.arguments)resultself.name2fn[tc.function.name](**args)self.messages.append({role:tool,tool_call_id:tc.id,content:json.dumps(result)})return达到最大轮数未收敛。用法给 MiniAgent 塞上「查天气」和「加法」两个工具让它处理一个需要先查、再算的任务——比如问到「北京的天气多少度然后温度再加 10 是多少」它会自动走get_weather(北京) → 看到温度 → add(温度,10) → 拼出最终回答agentMiniAgent(client,TOOLS,NAME2FN)print(agent.chat(北京的天气多少度温度再加 10 告诉我结果。))你不需要写任何一行流程编排循环和工具分发都是MiniAgent.chat内部完成的。无 API Key 也想看效果把get_weather返回固定值代码里已是演示固定值再看下面这段MiniAgent的模拟驱动——只要把client.chat.completions.create换成一段按关键词返回固定 tool_calls 的桩函数同样的循环代码就能在没有网络的情况下演示完整 ReAct 过程。框架层面的循环逻辑和真实调用完全一致。4.1 循环终止与防死锁max_rounds 上限一定设上限否则碰到一个反复调工具不停止的场景会无限循环烧 token。停止条件没有 tool_calls即视为任务处理完成。这是最常用的终止条件。异常兜底工具执行抛异常时把异常文本作为roletool回传让它自己重试或用别的手段。try:resultself.name2fn[tc.function.name](**args)exceptExceptionase:resultf工具执行失败:{e}# 让它看到错误自我修正4.2 多轮与短期记忆的一致性self.messages就是 ReAct 的多轮短期记忆。每一轮的工具调用 结果都被吞进messages,下一次请求带着完整上下文。这也是为什么之前说记忆messages 数组。总结与下一步这节我们把四要素逐个拉细到可写代码的程度Function Calling 工具描述进提示词 模型输出结构化调用 你自己执行 roletool回传让模型观察。记忆 短期用 messages 列表 截断/摘要长期用 embedding 向量 余弦检索。规划 CoT / ReAct / Plan-and-Execute 三选一按任务长度拆解。核心骨架一个能跑的最小 ReAct Agent 用MiniAgent从代码到概念背都可以自己扩。下一篇《实战用代码造出第一台能用的工具型 Agent》会把你写出的MiniAgent接上一个真实模型做成一个能自己查资料、做总结的完整工具型 Agent——你就正式完成首次能干活的运行里程碑了。在动手前如果你读到这里仍觉得roletool那块绕可以再回去看第 2 节那段run_once把注释逐条读一遍——那是理解整个 Agent 的钥匙。