ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

可运行源码是硬门槛:手写Agent最小闭环系统

可运行源码是硬门槛:手写Agent最小闭环系统 简介面向希望从零搭建智能体系统的开发者这是一份可直接运行的源码包聚焦将笔记系统从离线版升级为联机版并实现AI搜索、报告生成与自动笔记整理等核心能力。包内共9个文件体积仅15KB以4个Python脚本为主干分别对应研究者、编辑者、笔记记录者三个角色模块另含依赖清单、环境变量示例、说明文档、版本忽略与云端运行配置结构清晰便于快速定位和理解工作流程。资源还展示了三角色分工协作的具体案例研究者调用搜索工具采集信息编辑者汇总生成报告笔记记录者自动归档内容并配有可运行脚本供直接修改复用。目前已有148人学习下载适合具备一定Python基础、想通过完整代码理解RAG与智能体协作机制的开发者上手实践。可运行源码是硬门槛Agent系统的最小闭环先说一个我观察了很久的现象很多同学聊起Agent来头头是道什么感知、规划、执行、反思框架名词一套一套的。但真让他们把一段用户请求跑通到一个Agent自动调用工具并完成任务十有八九卡在不知道怎么把循环控制起来这一步。我踩过同样的坑。最开始在框架示例上直接改跑倒是能跑但遇到问题就像在迷宫里找门——框架帮你隐藏的那些细节恰恰就是出问题的地方。后来我痛下决心从零手写了一个最小可运行的Agent系统不依赖任何Web框架不搞花哨编排就一个纯粹的模型工具循环骨架。跑通之后很多概念瞬间就通了tool_calls怎么触发、工具结果怎么回填、上下文怎么不爆炸、错误怎么不把整个Agent搞挂。这篇文章就把这套骨架拆给你看包含一份可以直接复制运行的源码结构以及我实测踩过的坑和排查思路。我不会绑定某个特定框架底层调用你可以任意换成自己的模型服务封装。适合的人群很明确想真正理解Agent而不是只会调API的同学以及准备在自己的项目里从零搭建Agent、需要一个稳定起点的开发者。可运行源码是硬门槛Agent系统的最小闭环1.1 一个Agent系统到底由什么组成很多人理解的Agent是一个大模型一个Prompt其实这不是Agent这是聊天机器人。Agent系统的关键差异在于模型要能主动决定调用什么工具、看到工具结果后继续推理直到完成任务。所以它的最小闭环至少包含四样东西一个可调用的模型接口一组注册在案的工具一段循环控制逻辑一套消息上下文体系我见过不少人在第一步就直接用框架全家桶结果连模型为什么调用这个工具都解释不清。框架不是不能用但前提是你得知道它在背后做了什么。从零手写一个最小系统不是为了重复造轮子而是为了将来用框架的时候你能读懂它的日志和报错。1.2 项目结构刻意保持精简这是我实际使用的最小目录结构没有用任何框架也能直接跑agent_demo/ ├── agent/ │ ├── __init__.py │ ├── core.py # Agent核心循环 │ ├── tools.py # 工具注册表与分发器 │ └── memory.py # 上下文管理与压缩 ├── main.py # 入口接收一条用户问题 └── requirements.txt # 依赖openai等我刻意把所有逻辑放在四个文件里目的是让你能在半小时内读完所有代码。等你理解透彻后再往里面加向量检索、多智能体协作都不迟。1.3 为什么能跑的hello agent比漂亮的设计稿重要这里我想强调一个观点Agent开发中最有价值的事情是先让一个最朴素的循环跑通。因为Agent的难点从来不在单次调用模型而在多次调用之间的控制流和数据流转。工具返回格式不规范怎么办模型连续三次调同一个工具怎么终止上下文被撑爆了怎么压缩这些坑只有代码真正跑起来才会暴露。可运行的源码就像一个可以对照调试的基准线。你后面遇到的每一个诡异问题都可以问一句我把这个新改动去掉回到能跑的状态问题还在不在有了基线排查效率会高非常多。核心循环是怎么跑起来的一条用户问题的完整生命周期2.1 核心循环代码解读下面这段是整套系统的灵魂我把它完整放出来。逻辑很直接循环里调模型模型说我要调用工具就执行工具并把结果回填模型说可以回答了就返回最终文本。# agent/core.py from openai import OpenAI from .tools import TOOL_SCHEMAS, dispatch_tool client OpenAI() def run_agent(user_query: str, max_steps: int 10) - str: messages [{role: user, content: user_query}] for step in range(max_steps): resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOL_SCHEMAS, ) msg resp.choices[0].message # 模型没有请求调用工具说明可以输出最终答案了 if not msg.tool_calls: return msg.content # 把模型的工具调用请求追加到消息历史 messages.append(msg.model_dump()) # 逐个执行工具调用 for tool_call in msg.tool_calls: fn_name tool_call.function.name fn_args tool_call.function.arguments result dispatch_tool(fn_name, fn_args) # 把工具执行结果按 tool 角色回填 messages.append({ role: tool, tool_call_id: tool_call.id, content: str(result), }) return 已达到最大步数任务提前终止这段代码有一个非常关键的点助手消息和工具结果必须成对出现。模型返回的tool_calls里有id工具结果的tool_call_id必须对应上否则模型会认为消息历史错乱。这是很多初学者最容易忽略的细节——把工具结果当普通文本塞进去模型就失忆了。2.2 控制流里的退出条件为什么不能只靠模型自觉注意循环里有两个出口一个是模型不再请求工具返回最终答案另一个是步数上限。第二个出口不是摆设。我在实测中遇到的典型情况是模型在一个问题上反复调用同一个工具每次都拿到相同结果但就是不说我查完了。这种死循环如果不加步数上限你的账单会先教你做人。步数上限是Agent系统的安全阀不是可选项。你可以把整个循环理解成一个实习生向主管汇报工作的过程主管模型每下达一个指令你系统执行完再汇报结果主管根据新情况决定下一步。如果主管迟迟不拍板你就必须有个最多汇报几次的规矩。工具接入不是堆积木注册机制与错误传播最容易被忽略3.1 用装饰器维护一个统一的工具注册表工具层设计得好不好直接决定Agent能扩展到哪里。我推荐用装饰器注册模式每个工具函数上标注名字、描述、参数结构框架自动收集成OpenAI能识别的tools列表。# agent/tools.py TOOL_SCHEMAS [] TOOL_HANDLERS {} def register_tool(name, description, parameters): def decorator(func): TOOL_SCHEMAS.append({ type: function, function: { name: name, description: description, parameters: parameters, } }) TOOL_HANDLERS[name] func return func return decorator register_tool( get_weather, 查询指定城市的当前天气, { type: object, properties: { city: {type: string, description: 城市名} }, required: [city] } ) def get_weather(city: str) - str: return f{city} 当前天气晴25摄氏度 def dispatch_tool(name: str, args_json: str) - str: import json args json.loads(args_json) handler TOOL_HANDLERS.get(name) if handler is None: return f错误未找到工具 {name} return handler(**args)这个模式最大的好处是新增工具零侵入。你只要写一个函数、加一个装饰器Schema和分发器自动就位。我在实际项目里扩展到了十几个工具核心循环一行没改。3.2 工具报错的正确姿势让错误变成模型看得见的观察这是我想重点展开的部分。我见过的很多Agent项目工具函数里一抛异常整个Agent循环就崩了。很多人在搜索时都见过类似的报错agent execution terminated due to error。这个错误本质上就是执行器捕获到未处理异常后无法继续循环只能强行终止。排查过一次之后我给所有工具调用包了一层安全执行器def safe_call(handler, *args, **kwargs): try: result handler(*args, **kwargs) return result if isinstance(result, str) else str(result) except Exception as e: return f工具执行失败{type(e).__name__}: {e} def dispatch_tool(name: str, args_json: str) - str: import json try: args json.loads(args_json) except json.JSONDecodeError as e: return f工具参数不是合法JSON{e} handler TOOL_HANDLERS.get(name) if handler is None: return f错误未找到工具 {name} return safe_call(handler, **args)关键思路是把异常转成字符串作为工具结果发回给模型。模型看到工具执行失败KeyError: xxx它是能理解并采取下一步行动的——比如换一种参数写法、换一个工具或者直接告诉用户出了什么问题。异常信息对模型来说不是程序bug而是观察结果的一部分。这里也顺便回应skill和agent的区别这个话题工具更像是skill的载体是Agent可调用的能力单元而Agent本身是决策和执行的调度者。把能力的实现和调度逻辑解耦正是这个注册模式的核心价值。记忆与上下文Agent记不住和记太多同时存在4.1 短期记忆与长期记忆的边界Agent的记忆问题用大白话说就是既怕它忘记不住前面干了啥又怕它吃太多上下文窗口爆掉。短期记忆就是上面代码里的messages列表。它记录当前任务从开始到现在的所有消息是Agent理解任务上下文的基础。但问题在于OpenAI等模型都有token窗口限制比如4k、8k、128k一旦messages总长度超过窗口调用直接报错就算不报错模型也会因为输入过长而忘记开头的内容表现就是答非所问。长期记忆则是跨任务的持久化信息。我建议在最简单的骨架阶段先不引入向量数据库用一个JSON文件就够了。只有当你发现这个工具结果下次任务还要用时才值得把关键信息抽出来存到长期记忆里。4.2 上下文压缩让长对话不爆炸的最实用策略我最推荐的轻量方案是摘要压缩。当messages总长度超过阈值时保留最早的系统提示和最近的几轮对话把中间的历史对话交给模型生成一段摘要然后开一轮新对话。# agent/memory.py SUMMARY_SYSTEM 你是一个对话总结助手把下面这段对话压缩成简洁的中文纪要保留关键事实和已执行过的工具结果。 def compress_context(messages, max_tokens6000): total sum(len(str(m)) for m in messages) if total max_tokens: return messages recent messages[-4:] # 保留最近的4条 history messages[1:-4] summary summarize(history) compressed [ messages[0], {role: system, content: f历史纪要{summary}}, ] recent return compressed这里的summarize本质上就是一次把历史翻译成摘要的模型调用。压缩成本很低但能让你在长任务中保持Agent不失忆。我踩过的坑是把摘要放在system角色里后模型偶尔会混淆System里说的是过去的事和用户现在说的是新需求。解决办法是在摘要前缀明确标注历史纪要三个字实测能显著降低混淆概率。4.3 记忆不是越大越好很多Agent新手以为只要把窗口撑大就行。我的经验是窗口越大模型越容易忽略重要信息。与其无脑扩大窗口不如有意识地精简该放弃的历史就放弃该过滤的细节就过滤。Agent的聪明更多来自对关键信息的聚焦而不是把所有信息都塞进大脑。从能跑到稳跑我实测里踩过的坑与排查链路5.1 坑一工具异常直接杀死整个Agent这是最常见的问题也直接对应很多人搜到的agent execution terminated due to error。复现路径很简单让Agent调用一个会抛异常的工具比如查询一个不存在的城市ID。症状终端输出一长串TracebackAgent进程直接退出。观察异常发生在工具执行阶段而异常没有被任何地方捕获。定位检查dispatch_tool发现原始版本没有try/except。修复引入safe_call把异常转成字符串。验证重新运行同一问题时Agent会看到工具执行失败然后自己决定换一种写法或工具任务继续。这条链路看起来简单但在项目里排查时很容易绕弯路。最容易误导人的是有些Agent框架的日志里会先打印Calling tool xxx然后才报错让人误以为问题出在模型API调用上。你一定要先分清异常发生的位置是工具内部、工具分发器、还是模型调用层。5.2 坑二模型返回了非法JSON模型的tools参数返回的arguments是一个JSON字符串。理论上没问题但模型偶尔会输出一个抖机灵的JSON比如在末尾加逗号、单引号包裹key、甚至直接返回散文而不是JSON。一旦json.loads失败循环就挂了。这个坑的排查链路比较有意思症状同一段提示词在一批请求中偶发失败不是必现。观察失败请求和成功请求的输入几乎一样只是模型温度不同。定位在dispatch_tool入口打印args_json发现有的返回{city: 北京}key没加引号。修复不直接交给json.loads先用json.loads试失败后用demjson3或正则做兜底修复。我在生产代码里还加了一道查无此key就返回默认值的兜底宁可让模型多走一步也不让循环中断。5.3 坑三循环失控费用飙升还有一个在实际运行中非常值得警惕的坑Agent在循环里反复调用同一工具每次都返回相同结果却迟迟不输出最终答案。最严重的一次我本地跑一个任务max_steps设置成50实际跑了40多步才被截断token费用是预估的五倍。排查链路是这样症状任务耗时异常长账单变更令人心疼。观察日志显示同一个工具被调用了8次返回内容一模一样。定位给循环加连续相同结果计数器超过3次直接终止。修复除了全局max_steps再加一个局部去重机制如果某工具连续返回相同结果N次就在消息里注入注意这个结果你已经拿过多次了请换一种方式处理。验证同一任务现在最多在5步内收敛且最终结果质量没有下降。这个问题的根源不在模型笨而在于工具结果可能确实没有提供新信息。让模型意识到这条路走到底了它才会主动换路。5.4 坑四网络抖动和外层API不稳定最后是一个纯基础设施层的问题模型API偶发超时或返回5xx。这个不处理Agent循环照样会崩。症状偶发Timeout错误重试后就能跑通。定位给client调用加三层重试每次退避等待。修复用tenacity这样的库包装调用或者自己写for循环重试。验证连续压测200次请求0次崩溃。这层跟Agent本身的智能没有关系但它决定了Agent在真实环境里能不能稳。我见过太多Demo级Agent在演示环境跑得好好的一上生产就被网络波动干掉。重试、超时、熔断这些基础工作必须在骨架里就位。上线前的最后一道坎安全、成本与可观测性6.1 工具权限的最小化原则Agent能力强意味着它调用的工具权限也大。如果给Agent的工具列表里有一个执行shell命令的工具理论上模型能让Agent执行任何命令。我在安全测试中最常见的教训就是权限给得太大方了。我现在的做法是三级权限分层只读工具查天气、查数据库Agent可以自由调用。有副作用的工具发邮件、写文件必须经过二次确认。高危工具删数据、执行远端命令默认不注册需要显式打开开关。这里没有银弹但是有一个自查清单可以帮你快速过一遍这个工具被模型调用时最坏的结果是什么如果最坏结果不可接受那就加确认或加白名单校验。6.2 成本控制给Agent上一个看得见的账单Agent循环的步数上限是成本控制的第一道闸但不是唯一一道。我还在核心循环里加了一个简单的token统计每一步统计输入输出token数累加到全局计数。usage resp.usage total_tokens usage.total_tokens这个计数器能干什么它能帮你快速回答这个Agent完成任务花了多少钱这类问题。在Agent上线初期我建议把每次调用的token数打到日志里跑一整天后你会对哪些任务值钱、哪些任务烧钱有非常直观的认知。没有数据的优化都是空谈。6.3 可观测性没有日志Agent就是黑盒Agent出问题时最怕什么最怕你不知道它中间干了什么。我在core.py里加了最简日志记录每一步是调用了工具还是输出答案、工具名、入参、耗时。就这么几行日志帮我解决过大量为什么Agent给出这个结论的困惑。更进阶的做法是接入OpenTelemetry链路追踪把Agent循环里的每次模型调用和工具调用都记录成一个span。但说实话对大多数项目来说结构化日志就够用了。关键不是工具多高级而是你有没有养成每一步都有据可查的习惯。从骨架到生产你可以这样继续扩展写到这里骨架和坑都讲得差不多了。我最后分享一个自己的体会这个最小系统跑通之后最大的价值不是代码本身而是你终于有了一个可以信任的基线。之后再引入向量检索、多Agent协作、各种框架都是在已知地基上盖楼出了问题你知道往哪一层排查。如果你在搭建过程中也遇到了程序明明能跑但Agent就是表现不对的困惑我建议按这个顺序检查先看工具注册表是否完整再看工具结果有没有正确回填再看上下文有没有被截断最后看循环的退出条件是否合理。这四个环节覆盖了Agent系统90%以上的运行问题。源码我已经整理成上面四份文件的样子尽量保留了最精简的结构。你可以在自己的项目里建同样的目录把核心循环代码复制进去替换模型封装后直接体验一次完整的Agent生命周期。跑通之后再去折腾那些更炫酷的框架你会发现自己看文档的速度完全不一样了。本文还有配套的精品资源点击获取
返回列表