手写200行代码:构建最小AI智能体核心循环与工具调用实战
1. 200行代码的底气智能体的最小架构1.1 智能体到底在说哪件事先别被“AI智能体”这个包装得金光闪闪的词唬住。剥掉概念外壳你面对的本质上就是一个循环控制问题喂给大模型一个任务模型说“我要调用某个工具”你帮它执行这个工具把结果塞回给它它再决定下一步干什么如此反复直到它给出最终答案。这个“循环”就是智能体的内脏。风口上的Agent、工作流、自动化不管外面套了什么壳核心都绕不开这一圈。我见过很多张口就是“LangChain多智能体编排”“Graph状态机”的人真把那个循环逻辑抽出来问两句反而卡壳。说残酷一点如果他们手里的框架明天消失可能连一个能跑通“问天气→查到结果→回复用户”的对话都拼不完整。所以这篇文章要干的事很直接手写成形的最小智能体核心代码量控制在200行上下跑通“感知→决策→行动→观察→再决策”的完整闭环然后把“这套东西为什么值钱”的底层逻辑给你拆开。适合谁看不只适合正在学大模型开发的工程师也适合产品、算法转岗、自由接单的开发者。你不需要懂复杂的强化学习只要会Python基础、调过大模型接口就能跟上。1.2 为什么不是50行也不是1000行肯定有人问200行这个数字哪来的底气把智能体核心拆开真正必要的部分只有四块一个循环体负责控制“模型-工具”之间的信息流转一个工具注册表让模型知道有哪些工具可用、参数是什么一个输出解析器从模型返回的文本里稳定地抽出“该调哪个工具、传什么参数”一个极简的记忆容器让多轮交互不“失忆”。每块的平均实现量大概50行加起来就是200行。少于这个体量说明你把解析或记忆砍掉了那叫半成品Demo超过这个体量说明你已经进入了工程化地带比如状态持久化、多Agent拓扑、可观测性埋点这些不是“核心循环”的职责而是框架层的加分项。这个数字本身就是个很好的判断标尺。我面试候选人的时候不会问“你用没用过XX框架”只问“如果现在禁网白板画一个智能体交互时序再让我看一个不漏的状态转移代码”。能稳定画出来的人对Agent的理解不会差。反过来天天对着框架文档抄代码的人往往在这个环节原形毕露。2. 手搓一个能感知、能决策、行动的最小智能体2.1 核心循环代码逐行拆解下面这段代码是我保留的“最小核心版”没有数据库、没有消息队列、没有可视化面板纯Python 一个兼容OpenAI协议的模型调用接口。为了让没有API Key的人也能直接跑通我先用一段模拟输出把模型层“占位”跑完再把真实接入方案给你。import json import re from datetime import datetime # ---------- 1. 工具定义 ---------- def get_current_time(**kwargs): 返回当前时间 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def calculate(expression: str): 计算表达式注意这里是安全的eval壳子生产环境请用ast或数字解析 try: return str(eval(expression, {__builtins__: {}}, {})) except Exception as e: return f计算错误: {e} def get_city_weather(city: str): 模拟查询城市天气实际场景在这里接气象API mock_weather { 北京: 晴25℃, 上海: 小雨22℃, 深圳: 多云28℃, } return mock_weather.get(city, f未收录{city}的天气数据) TOOL_REGISTRY { get_current_time: {func: get_current_time, desc: 获取当前时间无参数}, calculate: {func: calculate, desc: 计算数学表达式参数expression为字符串如 2*34}, get_city_weather: {func: get_city_weather, desc: 查询城市天气参数city为城市名}, } # ---------- 2. 模拟模型调用占位实现后续替换为真实大模型 ---------- def mock_llm(messages, temperature0): # 这个函数仅用于演示循环逻辑真实场景请替换为你的大模型客户端调用 text Thought: 用户想知道北京天气我需要调用天气工具 Action: get_city_weather Action Input: {city: 北京} return text # ---------- 3. 核心循环 ---------- def extract_action(text): 从模型输出中解析 工具名 和 参数 action_match re.search(rAction:\s*(\w), text) input_match re.search(rAction Input:\s*(\{.*?\}|.), text, re.S) if not action_match: return None, None action action_match.group(1) raw_input input_match.group(1).strip() if input_match else {} try: action_input json.loads(raw_input) except json.JSONDecodeError: action_input {query: raw_input} return action, action_input def run_agent(user_query, max_iterations5, verboseTrue): history [{role: user, content: user_query}] for step in range(max_iterations): response mock_llm(history) if verbose: print(f[Step {step1}] 模型输出:\n{response}\n) thought re.search(rThought:\s*(.), response, re.S) if Final Answer in response: final_answer re.search(rFinal Answer:\s*(.), response, re.S) return final_answer.group(1).strip() if final_answer else response.strip() action, action_input extract_action(response) if action is None or action not in TOOL_REGISTRY: return f模型给出了无法识别的动作: {response} observation TOOL_REGISTRY[action][func](**action_input) history.append({role: assistant, content: response}) history.append({role: tool, content: fObservation: {observation}}) if verbose: print(f[Step {step1}] 工具 {action} 返回: {observation}\n) return 已达最大迭代次数任务未完成。 if __name__ __main__: result run_agent(帮我查下北京天气, verboseTrue) print(f最终结果: {result})这段代码的逻辑很直白但每一处都有讲究。while True式的for step in range(max_iterations)循环是智能体的发动机。为什么要限制max_iterations因为模型在开放场景下完全可能反复调用同一个工具不收敛一次对话烧掉你几百次Token还出不了结果。给它一个硬上限本质上是在“任务质量”和“成本兜底”之间取了平衡。我见过不设上限的Demo跑一个物流调度问题模型来回调了27次“查询库存”最后把上下文撑爆了。限定迭代次数是所有工程级Agent都会做的第一层防御。2.2 让大模型“遵循指令”的格式约定不知道你注意到没有上面的解析逻辑依赖一个铁律模型输出必须包含Thought:、Action:、Action Input:这三个字段。这设计是有学名的叫做ReAct模式Reason Act思路是让模型先输出思考过程再输出具体行动然后把行动结果以Observation的形式回灌给模型。你现在可能想问直接让模型输出JSON不行吗也行但我在实践里测下来ReAct的纯文本标记解析成功率高过JSON原因在于大模型对“逐行语义”的把控能力比对严格JSON语法的把控能力强得多。JSON只要少一个引号、多一个逗号整个json.loads就报错而文本标记即使稍微漂移我的正则还是能兜住。这就是老鸟和新手处理同一件事时的明显区别新手执着于“最优雅”老鸟执着于“最不容易出错”。但注意我在extract_action里做了第二层容错当Action Input不是合法JSON时我不会让进程崩掉而是把它塞进一个{query: raw_input}的结构里。这叫兜底策略。你做个简单Demo可能永远触发不到这条分支一旦部署给真实用户使用模型偶尔会抽风输出不带引号的残缺字段这时兜底策略直接决定你的服务是“好好报错”还是“整个链路瘫痪”。2.3 接入真实模型的关键替换点上面的mock_llm只是占位接真实模型时你只需要替换这一个函数核心循环完全不用动。替换方式非常简单from openai import OpenAI client OpenAI( api_key你的API_KEY, base_url你的服务商地址 # 国内很多兼容OpenAI协议的服务商填这里的地址即可 ) def llm_call(messages, temperature0, max_tokens512): resp client.chat.completions.create( model你的模型名, messagesmessages, temperaturetemperature, max_tokensmax_tokens ) return resp.choices[0].message.content接入真实模型后第一件震惊你的事就是原来mock时一路顺风的流程真模型经常不按格式出牌。它可能在Action:后面多打一个冒号可能在Action Input:后面输出一段散文甚至在系统提示没给清楚的情况下直接说“我不知道怎么调用工具”。这不是模型的智商问题而是格式约束不够强。我踩过的调整经验按优先级排序一是system提示词里给死格式明确写“你只能输出Thought/Action/Action Input这三行禁止输出其他内容”二是temperature必须往低调最好设成0让采样空间变窄模型会更乖三是max_tokens不要给太少有些模型思考步骤长输出到Action就截断了。这三个点按顺序调能解决80%的解析失败问题。3. 给智能体装工具和记忆从玩具到可用3.1 工具注册表为什么用字典而不是一堆if上面的TOOL_REGISTRY用的是字典映射。这是很多人容易忽略但极其值得说明的一个点与其写几十个if action xxx分支不如直接用字典把“动作名”和“函数对象”绑在一起。这样加一个新工具只需要加一个新函数 在字典里加一行注册全流程零侵入。我们顺手把它扩展成一个更通用的注册方式def register_tool(name, func, desc): TOOL_REGISTRY[name] {func: func, desc: desc} def get_tools_prompt(): lines [可用工具列表] for name, meta in TOOL_REGISTRY.items(): lines.append(f- {name}: {meta[desc]}) return \n.join(lines)get_tools_prompt()的作用是把所有工具注册信息拼成一段提示词系统级喂给大模型。这一步是整个智能体的“能力边界”来源——模型只能调用你注册进来的工具没注册的它想破头也调用不了。这就是所谓“给AI装手”的方式。实际项目里我见过往这个注册表里塞了三十几个工具的团队有查数据库的、发邮件的、调报表的效果比想象中的好因为模型确实能从描述里理解该调哪个工具。但你也别狂塞。工具描述不清晰或者参数定义混乱模型很容易选错。我见过一个实际翻车案例某个工具的描述里没有写清楚“这个接口只支持国内手机号”模型给一个海外号码用户悄悄调了这个工具返回了一串错误然后模型还很自信地告诉用户“发送成功”。这个坑提醒我们注册表里每一项描述必须包含边界条件这比你多写两个工具更能减少线上事故。3.2 对话记忆的朴素实现我在上面的核心循环里已经通过history列表实现了最简单的记忆每一轮的模型输出和工具观察结果都会追加进去。这个方案能用但藏着一个大隐患Token会越滚越长。用户问10个问题每个问题带一段工具返回五轮以后上下文可能就溢出窗口限制了。一个朴素但有效的做法是只保留最近N条消息def trim_history(history, max_messages20): if len(history) max_messages: return history return history[-max_messages:] def run_agent_with_trim(history, max_iterations5): for step in range(max_iterations): history trim_history(history) response llm_call(history) ...窗口截断是最廉价的记忆方案适合所有Demo和MVP。但它的代价也很直观太早的信息会被丢掉用户问你“我第一句话说了什么”时你就会失忆。进阶路线有三条供你按需选择第一是摘要压缩每次截断前把旧消息用模型总结成一段话塞回上下文第二是向量检索把历史消息向量化根据当前问题召回相关内容第三是结构化记忆把用户偏好、关键事实单独存成KV记录。这三种方案都能和上面的循环结合但它们属于“记忆架构优化”范畴核心循环依然不用变。3.3 两个容易被忽略的核心能力错误恢复与多步规划工具调用不可能永远成功。第三方接口挂了、传参格式错了、数据源返回空值这些在Demo里不会发生生产环境天天发生。处理方式很简单不要只把成功结果塞回给模型失败信息也要塞回去让模型自己消化异常、换一条路继续走。代码只需要这样try: observation TOOL_REGISTRY[action][func](**action_input) except Exception as e: observation f工具执行失败错误信息: {e}请尝试其他方式这个看似不起眼的改动是把智能体从“玩具”推向“能用”的关键一步。因为真实世界的不确定性根本不应该由你的代码去穷举处理应该把异常反馈给决策层——也就是大模型——让它基于当前观察重新规划。多步规划能力就藏在循环里模型第一次调用工具拿到Observation后会再次生成Thought可能是“天气查到了接下来需要计算温差”于是继续调用计算器然后第二次收到Observation最后给出Final Answer。这是智能体和普通聊天机器人本质上的分水岭它不是一次生成答案而是“按需多步推理直到条件满足”。你做客服自动化、数据分析、自动接单本质上都是在复刻这个循环。4. 从手写200行到工程框架框架到底补了什么4.1 框架不是魔法是把边界补齐手写跑通核心循环后你会更客观地看待LangChain、LangGraph这类框架。它们解决的不是“能不能跑”的问题而是“能不能规模化地跑”的问题。我把框架真正做的事列给你看第一状态管理。手写版本里一切状态都在内存进程一重启就全没。框架会帮你把Agent状态序列化、存库、恢复这是长时间运行任务的刚需。第二消息流编排。多智能体场景下A Agent的输出要作为B Agent的输入还要支持条件分支、并行执行手写这套又是一大摊代码。第三可观测性。每次工具调用耗时多少、Token烧了多少、哪一步开始偏离预期框架自带追踪面板手写就得自己埋点。第四缓存与重放。同样的工具请求来了直接返回缓存不在话下。打个比方手写核心循环相当于你手工和了一团面、擀了一张皮能包饺子了。框架则像一台压面机加一条流水线帮你把“包饺子”这件事规模化配上自动计量、质检、封装。但问题是一样的你只有先能手擀皮才知道流水线哪个环节出了问题。4.2 什么时候该手搓什么时候该上框架我给一条自己的判断准则不一定适合所有人但屡试不爽。需要集成外部生态、批量跑多个异构Agent、团队多人协作维护时直接上框架省得自己写那套状态持久化和UI面板。反过来你只是做个垂直场景Demo、公司内部自动化小工具、或者教学验证那就手写核心循环依赖少、调试快、出问题一眼能看懂。我踩过一个典型坑某次做一个内部工单自动分类小项目为了“显得专业”一上来就套了框架。结果项目过半发现绝大多数语法都在处理框架版本升级后的接口变动真正的业务逻辑反而不到三百行。后来我把核心循环手写整个项目直接删掉一半无关代码上线效率反而翻倍。框架是为你服务的不是让你去服务的。选型标准永远是“当前问题的复杂度”而不是“竞品技术方案里用了什么”。5. 百万年薪的底层逻辑不是代码值钱而是“能解决不确定问题”值钱5.1 稀缺能力的三个构成标题里那个“百万年薪”确实有夸张成分但背后的信号是真实的能把大模型从“聊天玩具”变成“业务工具”的人目前在市场上极其稀缺。我拆解过这类高薪岗位的能力要求发现它们高度聚合成三个底层能力不是某个框架的熟练度。第一范式迁移能力。传统编程是“规则确定性”范式你把逻辑写清楚机器照做。智能体开发是“概率决策”范式你写的不是步骤而是“循环 边界 反馈回路”具体怎么解决问题是模型自己决定的。能完成这种心智切换的工程师本来就筛选掉了一大半人。第二业务问题翻译能力。客户说“想要一个能自动回消息的机器人”平庸的实现是写一堆关键词匹配好一点的实现是套一个提示词模板值钱的实现是把它拆解成一个智能体循环意图识别工具、订单查询工具、退款策略工具、人工升级条件每一步都嵌进那个循环框架里。能把模糊业务需求准确翻译成“工具集 循环策略”的人价值不在代码上在于决策质量上。第三容错与评估能力。传统软件出错可以复现Bug然后修复智能体出错往往是“概率性漂移”你得建立评估集、抽样评测、灰度上线、兜底降级。这种“在不确定系统里做确定性交付”的能力是最少人具备的也是最值钱的。5.2 一个真实的工作场景对照举一个我实际参与过的通用场景来对照。某公司的客服团队之前处理退款咨询靠人工一条一条回复高峰期排队严重。他们找技术团队做自动化第一版方案是规则引擎写了两百多条if-else覆盖了80%的常见问法但用户一句“我上周买的东西不合适想退了钱什么时候能退回来”词法一变规则直接失配全部转人工。后来重构时核心思路换成了智能体循环注册了一个“订单查询”工具一个“退款进度查询”工具一个“发起退款申请”工具系统提示词里写好边界与话术规范模型自主决定调哪个工具、要不要转人工、怎么回复。整个核心循环不到两百行但业务指标的变化是决定性的自动化解决率从32%跃升至71%平均响应时间从4分钟降到8秒。这才是“百万年薪”逻辑的真实版本不是写了两百行代码就值一百万而是这两百行代码替代了原先需要养一整个客服班组才能达到的服务吞吐量。薪资的本质是对“你撬动的业务价值”的定价。你要做的不是焦虑标题而是找到那个“智能体循环能重构的低效业务环节”把它做出可量化的提升。5.3 给入行者的三条建议基于我接触过的成长路径给正在转型的人三条不算新的、但极少有人真正照做的建议。先把核心循环写熟。不需要急着学框架你对着上面那段代码改三轮自己调过三次工具解析的Bug之后对Agent的理解会超过很多背文档的人。找一个垂直小业务闭环练手。别做“万能助理”做“某个城市的天气播报机器人”“某个小团队的周报自动汇总助手”范围越小你能把控的变量越少越容易做出稳定效果。建一个评测集哪怕只有20条测试用例每次改动都跑一遍记录成功率曲线。很多调参行为最后沦为玄学就是因为没有基准线。你把基线立住了迭代速度反而不是最大的最珍贵的是你知道“改了什么导致效果变了”的掌控感。6. 实操避坑我踩过的问题与排查速查6.1 问题一Action Input解析频繁失败表现模型正常输出了Action字段但是输入参数解析不出来程序报JSONDecodeError。排查思路先看原始输出是不是被转义字符污染了。很多模型服务商返回的字符串里带转义反斜杠json.loads会直接爆炸。还有一种常见情况是模型自作聪明把Action Input:的值用单引号包裹这也不是合法JSON。我的解法分两层第一层解析之前先做一次字符串清理把多余反斜杠去掉第二层正则匹配时加一个模式接受“直到行尾的非JSON文本”解析失败时按纯字符串塞入。双保险之后解析成功率能从85%提到99%以上。6.2 问题二智能体陷入死循环表现模型反复调用同一个工具每次拿到相同的结果然后继续调用同一个工具直到max_iterations耗尽。这种情况发生的前提是工具返回的Observation没有给模型提供增量信息。比如你查库存第一次返回“120件”第二次还返回“120件”模型如果规划能力不强就会卡在原地反复查。解法有效的是两步第一步把Observation里塞进时间戳或者“本次是第几次查询”的通告让模型感觉到变化第二步在系统提示词里直接写死一条规则“如果你连续两次调用同一个工具且结果相同必须改变策略或给出Final Answer。”这种“规则注入”比依赖模型自觉可靠得多。6.3 问题三上下文超限导致行为退化表现对话前几轮还很正常越往后模型越笨经常忘掉早前指令甚至开始乱答。这基本都是history膨胀惹的祸。我在3.2里给过窗口截断方案这里补充一个更细节的执行经验截断时机不要放在每轮开头而是在一轮完整交互结束之后。否则工具调用链条中间被打断Observation没来得及回灌模型下一轮就断了上下文。更激进一点的方案是“摘要替换”当历史消息超过阈值时用一次单独的模型调用把这些历史总结成一段两三百字的摘要然后把摘要放回系统提示词里。这样既保留了关键信息又控制了Token增量。代价是多一次模型调用但在长会话场景里非常值得。6.4 报错速查表现象可能原因快速解法JSON解析报错模型输出非法JSON转义清理 正则兜底 非JSON回退工具名匹配失败模型加了前缀或大小写不一致匹配前正则提取纯单词统一小写模型不遵循格式系统提示词约束不足格式示例写进prompt temperature调0上下文超限历史消息膨胀窗口截断 / 摘要压缩工具执行异常参数类型不对或服务不可用try-except捕获异常并反馈给模型迭代耗尽模型多步规划能力弱规则注入 单步最大迭代检查Final Answer缺失模型到Action截断调大max_tokens或检查截断标志这张排查表是我在实际调试里沉淀出来的基本覆盖了手写智能体上线后90%的故障类型。每次遇到新问题先回看模型原始输出再对照这张表问题定位基本不会超过十分钟。关于手写代码和框架的关系我再多说一句个人体会写这200行代码的过程真正让你获得的不是那几行代码本身而是你开始用“循环 工具 反馈”的眼光去拆解业务问题。这个视角一旦形成无论你之后是用现成框架还是继续手写决策质量都会完全不同。市场对这个岗位的定价买的正是这种“看得透问题本质”的能力而不是某个具体的工具链。