从零搭建可运行的Agent系统:核心架构、源码实现与避坑指南
发布时间:2026/9/25 23:35:42来源:尧图网络
简介这是一套面向软件开发者的智能体系统可运行源码与搭建指南适合具备一定编程基础、希望从零学习检索增强生成与智能体结合实践的读者。压缩包共9个文件整体约15KB其中包含4个Python脚本、依赖清单、环境变量示例与说明文档目录结构清晰代码量精简便于快速理解与二次开发。作者以笔记系统从离线版升级为联机版的案例为线索展示接入AI搜索、自动生成报告与整理笔记的完整流程并给出可直接运行的代码骨架覆盖了研究者、编辑者、笔记记录者三个角色的分工协作。目前已有150人学习下载可帮助读者快速掌握多角色协作机制、搜索工具与笔记工具的集成方法以及如何将本地工具改造成在线智能服务的基本路径。1. Agent系统没有你想的那么玄先搞清它在解决什么问题“搭建Agent系统”最近几乎成了大模型应用里最热闹的方向但你去搜相关教程看到的文章大多在讲概念感知、规划、记忆、工具调用真正拿到手能跑的可运行源码却很少。我给一个更直接的判断Agent系统的本质是给LLM装上一个“能操作外界的循环”——模型每说一步系统就去执行工具、把结果喂回去再让模型决定下一步做什么。这个循环一旦跑通你手里那个只会聊天的模型就变成了能查数据、发消息、写文件的干活工具。本文目标很明确从零搭出一套可运行源码中间拆到执行链路、工具注册、记忆槽和评估方法。适合已经会写Python、调过LLM API但还没把“Agent”落成代码的工程师。2. Agent系统的最小可用架构模型、工具、记忆、循环怎么串起来Agent不是一个新的模型也不是一个单独的函数调用它是一条被组装起来的执行链路。你真的把它想清楚了后面写代码只是在做翻译。常见实现里最小闭环包含四部分一个能理解任务并输出动作的LLM一组可以被LLM“点名”调用的工具一段存得下对话历史的记忆以及一个决定何时停下来的循环控制。我见过不少新手把Agent直接等同于“调一次带tools的chat接口”跑出来的结果是模型礼貌地回复“好的我来帮你查天气”但根本没有工具被调用。这不是模型笨而是你的系统里压根没有“调用工具”的执行路径。Agent和普通问答的关键区别在于模型输出的不是最终答复而是“下一步动作”。2.1 Agent和LLM、AI模型到底差在哪多一次“操作”的能力三个词先分清。LLM大语言模型是根据输入生成文本的模型AI模型范围更宽分类模型、语音模型、向量模型都算Agent则是一个运行时系统拿模型当“大脑”同时握着工具、记忆和循环。最常见的误解是把“调一次LLM接口”等同于“搭了Agent”——这就像把发动机直接叫成汽车缺了底盘、轮子和方向盘。对比一个具体场景。你用LLM做翻译工具输入“把这句话翻成英文”输出英文一次结束这是单次问答。但如果你让它“分析这份PDF并生成摘要邮件”LLM自己没法打开PDF它要先调用解析工具拿到文本再调摘要工具做压缩最后调邮件接口发出去。整个过程中模型只是在一轮一轮地“说下一步做什么”真正干活的永远是外部代码。这些外部代码的集合就是工具层。从用户角度看Agent带来的价值不是“更聪明”而是“更会干活”。它把你原本在代码里写死的if-else判断变成了模型在运行时动态决定。代价是你不再能确定它会走哪条路。所以循环控制、超时、预算、校验这些环节从第一天起就得跟着加进来不能等跑挂了再补。顺着这个思路再回答一个高频问题模型怎么知道自己有哪些工具靠function calling。你把每个工具的名字、描述、参数JSON Schema传给模型模型根据用户需求自己选择调哪个。工具描述的写法在这里极其重要写太泛“查询数据”和写太细“查询用户订单表中的订单状态入参为order_id字符串”的效果能差出一大截。2.2 一条完整的Agent执行链路感知、规划、调用、反馈行业内流行的ReAct模式翻译成大白话就是“思考-行动-观察”循环。一次完整执行大致是四条腿走路感知接收用户输入和系统提示词、历史消息、可选的长期记忆一起组装成上下文。规划LLM基于上下文输出动作。动作只有两种——调用某个工具并给出参数或者直接给最终回答。调用系统解析模型输出的工具名和参数在本地注册表里找到对应函数并执行。反馈把执行结果作为tool消息追加到对话历史再交给LLM让它根据结果决定下一步。终止模型输出最终答案或循环次数耗尽、超时、触发预算上限Agent停止。列个表看普通对话和Agent循环的差别会更清楚环节普通LLM调用Agent循环输入一条用户消息用户消息系统提示历史工具Schema工具结果输出文本文本或工具调用指令执行无有代码实际去跑反馈无有工具结果回灌模型终止单次返回多轮循环直到答案或达到终止条件这条链路里最容易被低估的是“反馈”这一步。很多初版Agent跑不起来不是模型选得不好而是工具执行结果没有正确回灌。比如工具返回的是一个Python dict你没序列化成字符串就往历史里塞模型看到的是一段格式残缺的内容只能凭空猜答案于是下一步就越走越偏。第5章的踩坑记录里我会专门展开这类问题。Agent的规划能力上限取决于模型的推理水平。做复杂任务时我一般优先选带reasoning能力的模型它们在做多步拆解时明显更稳。代价是更慢、更贵。务实做法是简单任务用快模型一旦进入工具调用循环再考虑切强推理模型。如果用的国产模型提供OpenAI兼容接口MiniAgent的client也能直接对接只需要改base_url和api_key。2.3 Agent记忆怎么分层短期上下文、长期摘要和永久存储记忆是搭建Agent时最容易拍脑袋的部分。从实现角度可以切三层短期记忆就是当前对话的message列表直接随对话拼接给模型长期记忆是从历史对话里提炼出的摘要或关键事实对话开始时随系统提示注入永久记忆是跨会话存在的实体关系、用户偏好、业务知识通常落到数据库或向量库需要时再拉取。我建议第一个可运行版本先只做短期记忆把message列表管好能正确裁剪长度就够了。这里有个常见误区以为上下文越长越好。模型能装下的上下文是有限的塞太多历史反而会降低对当前指令的遵循准确率。等系统真正要跨会话服务用户时再加长期摘要每N轮对话后调一次LLM把这段对话压成200字以内的摘要单独存一个summary角色下次任务开始时放回上下文。永久记忆则需要先想清楚“哪段记忆在哪个环节被哪一层用到”再决定存储方案不要一上来就上向量库后面第3章会附一个可落地的摘要实现。3. 用Python手写一个能运行的Agent核心最小源码与参数调法有了结构现在把它落成代码。我不会一上来就上复杂框架先让你看懂一个最小Agent循环的实现。这个版本走OpenAI兼容的chat接口和function calling核心只有一个类加一个工具注册示例。你把它跑通后再往里面加记忆、加并发、接框架就不会被黑匣子卡住出了错也知道该去哪里看。3.1 最小可运行源码一个带工具调用循环的Agent类核心代码尽量少封装所有环节都露在外面方便打断点观察每一步发生了什么。项目结构也不复杂一个agent.py放核心类一个demo.py放工具和启动逻辑。import json from openai import OpenAI SYSTEM_PROMPT 你是运行在用户本地的Agent。你需要根据用户请求 在可用工具中选择合适的工具并调用最后给用户一个清晰的中文回答。 如果工具结果不足以回答问题继续调用其他工具如果已经足够直接回答。 class Tool: 工具封装模型侧看到的是schema执行侧看到的是Python函数。 def __init__(self, name, description, parameters, func): self.name name self.description description self.parameters parameters self.func func # 实际执行的函数 def to_schema(self): 转成OpenAI function calling需要的JSON Schema结构。 return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters, }, } class MiniAgent: def __init__(self, client, modelgpt-4o-mini, toolsNone, max_steps8, temperature0.2): self.client client self.model model self.tools tools or [] self.max_steps max_steps self.temperature temperature self.messages [{role: system, content: SYSTEM_PROMPT}] def find_tool(self, name): for t in self.tools: if t.name name: return t return None def run(self, user_input): self.messages.append({role: user, content: user_input}) for step in range(1, self.max_steps 1): print(f--- step {step} ---) resp self.client.chat.completions.create( modelself.model, messagesself.messages, tools[t.to_schema() for t in self.tools], temperatureself.temperature, ) msg resp.choices[0].message if not msg.tool_calls: # 没有工具调用说明Agent决定给最终答案 self.messages.append({role: assistant, content: msg.content}) return msg.content # 有工具调用先把模型输出原样放回历史再逐个执行 self.messages.append(msg) for tc in msg.tool_calls: print(f[call] {tc.function.name} args{tc.function.arguments}) tool self.find_tool(tc.function.name) if tool is None: result fError: 工具 {tc.function.name} 不存在 else: try: args json.loads(tc.function.arguments) result tool.func(**args) except Exception as e: result fError: {e} self.messages.append({ role: tool, tool_call_id: tc.id, content: str(result), }) # 继续下一轮让模型看到工具结果 return Reached max_steps without final answer. def get_current_time(): 示例工具返回当前时间字符串。 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def run_demo(): client OpenAI() # 读取OPENAI_API_KEY环境变量 tools [ Tool( nameget_current_time, description获取当前日期和时间返回字符串, parameters{type: object, properties: {}}, funcget_current_time, ), ] agent MiniAgent(client, modelgpt-4o-mini, toolstools, max_steps5) answer agent.run(现在几点) print(answer) if __name__ __main__: run_demo()这段代码的逻辑并不复杂。Tool类做了一层封装模型只看到to_schema生成的描述实际执行的是func这个Python函数。好处是以后加新工具不用改循环往tools列表里塞一个对象就行。MiniAgent.run方法里的for循环就是上一章说的ReAct循环先调一次模型有tool_calls就执行工具并把tool消息回灌没有tool_calls就返回内容。有一个细节必须强调self.messages.append(msg)这一行不能省。OpenAI的调用规范要求工具调用消息必须原样放回历史后面每条tool消息要靠tool_call_id关联前一次的工具调用。很多初版代码漏掉这行模型看到的是“上一条调用记录缺失”轻则参数对不上重则直接报错终止。参数按下面这几条去调能少踩一半坑model模型名直接决定推理能力和价格。先用gpt-4o-mini这类便宜模型把循环调通再换更强模型验证效果。想接国内模型的把OpenAI的base_url和api_key换成对应厂商的兼容接口即可。max_steps整个循环的最大轮数。建议先设5到8足以覆盖大多数简单任务也不会让异常场景无限烧钱。要调高之前先问自己这个任务真的需要十几步很多时候是工具返回结果质量太差导致模型反复重试。temperature控制在0到0.3之间。Agent要的是稳定决策不是创意发挥温度太高大概率出现“换着说法反复调同一个工具”的翻车现场。我默认给0.2。工具描述质量虽然不在参数列表里但它对工具选择准确率影响最大。参数名写清楚能用枚举就别用自由文本比如查询类型给[today, range]模型就不容易编造值。3.2 怎么验证它真的在“思考”而不是乱跑跑上面代码的时候打开终端盯着print输出。正常轨迹类似用户问“现在几点”模型调get_current_time工具返回时间模型给最终答案。如果看到模型反复调同一个工具、参数不停变或调一个不存在的工具说明上下文或工具Schema出了问题。第一次跑通后可以做个小压力测试给Agent挂两个工具一个正常返回一个永远抛异常。观察它看到Error: xxx之后会不会自己换工具或者直接向用户说明失败原因。如果它假装错误不存在继续用同样的参数调同一个工具那就得在系统提示里补一句“工具可能执行失败失败时停止重试并向用户说明原因”。反馈链路通不通这个测试几分钟就能看出来。3.3 给Agent加记忆槽短期对话上下文与长期摘要并存上面的MiniAgent目前只有对话内记忆messages在run内不断增长但每次run之间是独立的。对可用的Agent系统来说这不够。至少需要两种短期记忆是当前任务的上下文对应代码里的messages列表要做的是“裁剪”。简单做法是设置最大轮数超过后把最早的非必要消息移除更好的做法是把早期对话压成摘要再放回上下文。长期摘要记忆的代码不复杂一次对话结束后调一次LLM即可def summarize_history(messages, client, modelgpt-4o-mini): 把一段对话压成200字以内的摘要保留事实、偏好和结论。 text \n.join( f{m[role]}: {m.get(content) or m.get(tool_calls)} for m in messages if m.get(content) ) resp client.chat.completions.create( modelmodel, messages[ {role: system, content: 请把上述对话压缩成200字以内的中文摘要。 只保留用户的目标、关键事实、已确认的偏好和最终结论。}, {role: user, content: text}, ], temperature0, ) return resp.choices[0].message.content这段代码的记忆选型思路是短期记忆管当下长期摘要管跨轮次。注意temperature必须给0摘要场景不允许创造性发挥否则会把事实总结歪。压缩后的摘要建议存SQLite或JSON文件字段带上user_id、created_at、summary_text方便跨会话检索。再往上是永久记忆那就要上向量库做相似度召回把Top-K条拼进系统提示。但一个可运行的Agent系统往往用JSON加SQLite的组合就能覆盖八成需求先把这一层做好更重要。4. 从手写代码到框架编排Agent框架选型与编排边界手写最小循环的意义在于理解执行过程。但真实项目里你不太会愿意自己维护工具调用的并发、重试、状态持久化、多模型切换这些事。这时候就需要框架。需要提醒的是框架不是越多越好选型之前先想清楚你缺的是“运行时支撑”还是“决策编排”。4.1 Agent框架和harness的区别先分清编排层和运行时“harness和agent区别”最近问的人很多这是个好问题。harness指的是运行Agent所需的外围支撑系统包括工具注册、会话状态、LLM调用接口、错误处理、日志链路这些。agent框架更偏“决策逻辑的编排”比如多步计划的生成、子任务分发、多角色协作。换句话说只做一个单Agent工具调用你需要的其实是harness几十行循环就够了。要做多Agent协作、人机审批、动态规划才需要框架级的编排能力。不少团队选型翻车是因为把两者混在一起拿着一个主打harness的框架硬写复杂编排或者拿着编排框架却自己重新造了一套harness两边能力都浪费。先判断你的核心复杂度在哪再选对应层级的工具。4.2 常见框架选型对照什么场景自己写什么场景用现成按我接触的常见做法给一个选型参考。纯手写循环适合原型期或工具链很短的项目它能让你对每次调用成本、每步状态完全可控。任务一旦复杂再考虑这几类LangGraph这类状态图框架适合流程要画成节点图、节点失败要回退重试、中间状态要持久化的场景把多步动态规划显式画出来。AutoGen、CrewAI这类多Agent框架适合明确要把任务拆给多个角色并行处理的场景。Dify、Coze这类平台化产品适合快速给业务方搭一个带界面的Agent原型内置知识库、工作流但定制深度有限复杂逻辑容易顶到天花板。选型的一个实际判断标准是你的流程是确定性多还是模型决策多。如果八成路径是人能预判的固定流程用轻量框架加状态机反而更稳如果八成路径是模型现场决定的才需要完整的Agent框架。不要因为某个框架在热搜上就套进项目Agent项目的复杂度大多来自业务不来自你用了多炫的框架。4.3 多Agent协作的两种模式主从编排与对等协商多Agent协作是所有Agent方向里听着最诱人、落地最容易翻车的一块。生产里用得最稳的是主从模式一个主Agent负责拆解任务把子任务发给多个专用Agent再汇总结果。这种模式好控制、好排查主Agent卡住了你知道是哪个子任务的问题。另一种是对等协商模式多个Agent以相同身份围绕一个目标互相追问适合头脑风暴、方案评审。但这种模式必须在消息协议上做严格限制比如规定每条消息必须带“结论依据”否则上下文很快被无效对话塞满。经验之谈能用主从解决的不要上对等多Agent的通信成本是成指数涨的两三个角色的对话噪音也会比你想的重得多。搭建多Agent时每个子Agent还是一个独立循环但它的tools要收窄。我给每个子Agent单独配置允许调用的工具列表不让所有Agent共享全部工具否则模型会跨Agent乱调工具你不知道哪里把状态改脏了。外部工具如果走的是MCP这类协议也可以做一个适配层把远端工具映射成内部的Tool对象对主Agent来说接口保持一致就行。4.4 编排层最容易漏掉的成本控制与降级策略框架层给了你便利也容易让你忽略预算。手写循环时你对每次调用敏感上了框架反而容易放开跑。我一般会在编排层做三件事单Agent任务设置总步数上限和单步超时每次LLM返回后记录usage累加超过预算阈值直接终止给关键工具调用设置失败降级路径不让一个工具挂了拖垮整条链路。5. Agent搭建避坑5个让我反复翻车的常见问题与排查这一章是血泪经验。每个问题都写成“现象→原因→解决”你照着对号入座就行。5.1 工具调用报“agent execution terminated due to error”先查工具返回体现象模型调用链路没有问题工具一旦报错整个Agent任务立刻终止日志里出现类似agent execution terminated due to error.的提示没有下一步重试也没有兜底回答。原因工具函数内部异常直接向上抛执行线程退出。还有一种情况是工具返回内容过大单条消息超过上下文窗口或网关限制运行时直接判定为致命错误。解决给每个工具执行外层包try/except异常时返回结构化错误而不是抛出异常。常见做法是让所有工具返回统一格式def safe_call(func, *args, **kwargs): try: data func(*args, **kwargs) return json.dumps({status: ok, data: data}, ensure_asciiFalse) except Exception as e: return json.dumps({status: error, error: str(e)}, ensure_asciiFalse)同时把工具返回内容截断单条tool消息控制在2000字符内。再在系统提示里写明“工具返回可能带Error遇到Error时分析原因后决定下一步”模型会按照提示继续运行而不是束手无策。5.2 Agent死循环max_steps设了还是跑满问题在哪现象max_steps设置到8任务还是跑满8轮不返回最终答案日志里全是工具调用最后一行是Reached max_steps without final answer.。原因最常见的是模型每次都被工具结果带跑不停发起新调用始终没有走进“给最终答案”的分支。另一个隐蔽原因是循环计数的口径不对有的实现把一次“模型思考工具执行”拆成两段各计一次实际模型调用次数等于max_steps的两倍。解决先把每轮step号真正打印出来确认计数口径是否一致。再检查系统提示明确要求模型“信息足够时必须在最后一条消息里直接给最终答案”。还可以加一个硬保护同一工具连续调用超过3次就强制终止直接返回“该工具连续失败无法完成任务”。5.3 记忆污染Agent“记住”了不该记忆的内容越用越傻现象长期摘要里出现了用户某次随口说的临时值几天后Agent把这个过期值当成了用户偏好或者工具返回的噪音数据被摘要当成业务结论写进去。原因摘要生成没有过滤规则LLM在压缩时无法区分“用户明确表达的偏好”和“本次任务临时生成的中间值”。解决摘要提示词里明确限定范围“只允许记录用户明确的偏好、事实性结论和待办事项工具原始输出不属于摘要记录范围”。对可能过期的记录加expires_at字段召回时过滤掉过期内容。这一步不做Agent服务时间越长上下文里的“幻觉历史”越多。5.4 并行工具调用结果串台tool_call_id对不上现象模型一次输出多个tool_call并行执行后A工具的结果被当成B工具的结果回填模型的下一步决策建立在错误数据上。原因并行执行时按完成顺序收集结果没有按tool_call_id把结果和调用一一配对。解决无论工具以什么顺序执行完回填时都按id匹配results {} for tc in msg.tool_calls: results[tc.id] run_tool(tc.function.name, tc.function.arguments) for tc in msg.tool_calls: self.messages.append({ role: tool, tool_call_id: tc.id, content: results[tc.id], })这样结果与调用严格对应不依赖执行顺序排查日志时也直观得多。5.5 预算失控一个Agent任务跑出几十美元的排查现象任务看起来不复杂账单却很吓人。查日志发现同一工具被调了几十次每轮还带着越来越长的历史消息。原因工具反复报错但模型不放弃重试策略没有上限上下文没有裁剪越到后面的轮次token消耗越大。解决在循环里累加拿到的usage超阈值直接终止工具结果截断加同一工具连续调用上限。还有一个低成本技巧在系统提示里告诉模型“如果工具连续两次返回相同错误停止重试并向用户说明失败原因”。这个提示能挡掉大部分重试循环先试这个再谈框架级限流。6. 用Evals和Skill注册表给Agent质量上保险两个进阶技巧Agent跑通只是开始真正的问题是你改了一行提示词怎么知道整体质量没倒退我现在的习惯是任何Agent项目都先建一条golden path评估集。不需要多十条约等于一个端到端场景每条固定输入、固定期望。每次改动后跑一遍规则断言能挡掉大部分回归。EVAL_CASES [ {input: 现在几点, expect: 时间, tool: get_current_time}, {input: 查一下订单123状态, expect: 订单, tool: query_order}, ] def run_evals(agent): for case in EVAL_CASES: output agent.run(case[input]) passed case[expect] in output print(f{case[input]}: {passed})规则断言先做起来后续要测更开放的任务再用“LLM as judge”让另一个模型按维度打分。评估集的价值是当你换了模型、改了提示词、动过工具描述后它能立刻告诉你哪些能力悄悄掉了。第二个技巧是给Agent做一个skill注册表。工具少于10个时全量暴露没问题工具多了模型会陷入“选择困难”反复挑错工具。做法是每个工具挂一个触发关键词列表用户输入进来先做一轮轻量匹配只把命中的工具Schema暴露给模型。这样模型每次面对的工具列表从几十个缩到三五个选型准确率会明显提升上下文token也省下一截。这个技巧和上文的MiniAgent完全兼容注册表只影响tools参数的组装不影响循环本身的执行逻辑。当你把Skill注册表和Evals结合起来Agent系统才算真正进入可维护状态。这两个习惯帮我避开了很多次上线前的翻车希望帮到你。本文还有配套的精品资源点击获取
网站建设高端定制企业官网