1. 为什么我要自己动手写一个 Agent市面上关于 Agent 的文章已经多到看不过来了但真正落到代码层面你会发现大部分内容要么停留在概念层面反复讲 ReAct、CoT、Tool Use 这些名词要么直接甩出一个庞大框架的文档链接让你自己去啃。我最初也是抱着“找个开源框架用用算了”的心态结果翻了好几个仓库之后发现要么抽象层太厚导致调试困难要么依赖太重跑起来一堆环境问题。后来我干脆决定自己参照一个轻量级 Agent 的实现思路从零搭一个最小可用的版本。这个想法听起来有点“重复造轮子”但实际做下来收获远超预期。当你亲手处理消息拼接、工具调用解析、循环终止条件这些细节时对 Agent 运行机制的理解会发生质变。这篇文章就是把我整个实现过程拆开来讲——不是教你调某个框架的 API而是从架构设计到代码落地完整走一遍一个小 Agent 的诞生过程。适合有一定编程基础、想真正搞懂 Agent 内部运转逻辑的开发者也适合那些用现成框架遇到瓶颈、想自己定制核心逻辑的人。我参照的核心思路来自一个叫 pi-agent 的轻量实现它的设计哲学很简单用最少的抽象层把 Agent 的核心循环暴露给开发者。下面我会先讲整体设计思路再逐层拆解关键模块最后给出可直接运行的代码和踩坑记录。2. 整体架构设计与核心思路拆解2.1 一个 Agent 到底需要哪些东西先把问题简化。抛开那些花哨的多模态、长期记忆、多 Agent 协作不谈一个能干活的最小 Agent 本质上就是三件事的循环接收输入、决定下一步动作、执行动作并把结果喂回去。这个循环听起来简单但每个环节都有设计决策要做。我最终确定的架构包含四个核心模块消息管理器负责维护对话历史包括系统提示、用户输入、助手回复、工具调用结果。它决定了每次请求发给模型时携带哪些上下文。模型接口层封装对语言模型的调用处理请求格式化和响应解析。这一层要屏蔽不同模型提供方的差异。工具注册与调度器管理可用工具的定义、参数校验和实际执行。工具是 Agent 与外部世界交互的唯一通道。主循环控制器串联以上所有模块控制“思考-行动-观察”的循环节奏决定何时终止。为什么这样分因为我在实际调试中发现Agent 出问题的时候故障点几乎总是落在这四个模块之一。分层清晰意味着出问题时你能快速定位是上下文拼错了、模型返回解析失败了、工具参数不对、还是循环条件写错了。如果所有逻辑揉在一个大函数里调试就是灾难。2.2 为什么选择“薄抽象”而不是“厚框架”这是整个项目最关键的决策。我试过用一些重型框架它们提供了链式调用、自动重试、内置记忆管理等一大堆功能。好处是上手快坏处是当你想改一个细节行为时往往要翻好几层源码才能找到对应的配置项。pi-agent 的思路正好相反核心循环就几十行代码所有可定制的地方都以最直白的方式暴露出来。我采纳了这个思路具体体现在几个方面消息历史就是一个普通的列表不做任何自动截断或摘要截断策略由使用者自己决定。工具定义就是一个字典键是工具名值是包含描述、参数 schema 和执行函数的对象。循环终止条件就是一个简单的判断模型不再请求调用工具或者达到最大迭代次数。这种设计的代价是你需要自己处理一些边界情况但收益是你对 Agent 的每一个行为都有完全的掌控力。对于想深入理解 Agent 的人来说这个代价完全值得。2.3 消息格式的设计考量消息格式是整个系统的骨架设计不好后面全是坑。我采用的是类似 OpenAI Chat Completions 的消息结构每条消息有role和content两个核心字段。role有四种取值system、user、assistant、tool。这里有个容易踩的坑当模型决定调用工具时它的回复里content可能为空但会带一个tool_calls字段。而工具执行结果需要以role: tool的消息追加回去并且要带上对应的tool_call_id来匹配。很多初学者在这里出错导致模型收到工具结果后无法正确关联到之前的调用请求。注意不同模型提供方对工具调用结果的消息格式要求略有差异。有的要求工具结果消息必须紧跟在助手消息之后有的允许中间插入其他消息。实现时最好严格按照目标模型的文档来构造。我选择用一个统一的内部格式然后在模型接口层做转换。这样切换模型时只需要改接口层的适配代码主循环逻辑完全不用动。3. 核心模块的详细实现与关键细节3.1 消息管理器上下文窗口的取舍艺术消息管理器看起来只是维护一个列表但实际写起来要考虑的事情不少。最核心的问题是上下文窗口有限历史消息无限增长怎么取舍我的做法是分三层处理第一层是系统提示永远保留在消息列表最前面。系统提示里定义了 Agent 的角色、能力边界、输出格式要求等。这部分内容通常不长但对 Agent 行为影响巨大。第二层是最近的对话轮次按时间倒序保留。我设置了一个可配置的max_recent_turns参数默认保留最近 10 轮完整对话。一轮指的是用户消息加上助手回复可能包含多次工具调用。第三层是更早的历史如果总 token 数超过阈值就从最老的消息开始丢弃。这里有个细节丢弃时要保证不破坏工具调用和工具结果之间的配对关系否则模型会报错。def trim_messages(messages, max_tokens, token_counter): system_msgs [m for m in messages if m[role] system] other_msgs [m for m in messages if m[role] ! system] # 从后往前保留直到超出预算 kept [] total sum(token_counter(m) for m in system_msgs) for msg in reversed(other_msgs): msg_tokens token_counter(msg) if total msg_tokens max_tokens: break kept.insert(0, msg) total msg_tokens # 确保不切断工具调用配对 kept fix_tool_pairing(kept) return system_msgs keptfix_tool_pairing这个函数的作用是如果保留列表的第一个消息是role: tool说明它的配对助手消息被截掉了需要把这条工具结果也去掉。反过来如果最后一条助手消息包含tool_calls但没有对应的工具结果也要处理。3.2 模型接口层请求构造与响应解析这一层的职责是把内部消息格式转换成目标模型 API 要求的格式然后把响应解析回内部格式。我以 OpenAI 兼容接口为例来说明。请求构造时需要注意几个点工具定义需要转换成模型要求的 JSON Schema 格式。每个工具包含name、description、parameters三个字段。description的写法直接影响模型能否正确选择工具。我踩过的坑是描述写得太简略模型经常选错工具或者参数填错。后来我把每个工具的描述都写成“什么场景下用这个工具 每个参数的含义和格式”准确率明显提升。temperature参数对 Agent 场景建议设低一些0 到 0.3 之间比较合适。太高会导致模型输出不稳定工具调用格式容易出错。响应解析主要处理两种情况普通文本回复和工具调用请求。工具调用请求里模型会返回工具名和参数 JSON 字符串。这里必须做 JSON 解析的异常处理因为模型偶尔会返回格式不正确的 JSON。def parse_tool_calls(response): tool_calls [] for tc in response.choices[0].message.tool_calls or []: try: args json.loads(tc.function.arguments) except json.JSONDecodeError: # 尝试修复常见格式问题 args try_fix_json(tc.function.arguments) tool_calls.append({ id: tc.id, name: tc.function.name, arguments: args }) return tool_callstry_fix_json是我后来加的一个兜底函数处理模型返回单引号、尾逗号、缺少引号等常见问题。虽然不优雅但实际运行中确实能救回不少失败请求。3.3 工具注册与调度让 Agent 真正能干活工具是 Agent 的手和脚。没有工具Agent 就只是一个聊天机器人。我的工具注册设计如下class Tool: def __init__(self, name, description, parameters, func): self.name name self.description description self.parameters parameters # JSON Schema self.func func def execute(self, **kwargs): return self.func(**kwargs) registry {} def register_tool(tool): registry[tool.name] tool每个工具本质上就是一个带元数据的函数。parameters用 JSON Schema 描述既用于生成给模型的工具定义也用于执行前的参数校验。我内置了几个基础工具来验证整个链路计算器接收一个数学表达式字符串返回计算结果。用来测试模型能否正确提取参数。当前时间无参数返回当前时间字符串。用来测试无参数工具的调用。文本统计接收一段文本返回字符数、词数、行数。用来测试多参数工具。工具执行时的关键细节第一参数校验不能省。模型可能传入多余参数、缺少必要参数、或者类型不对。我在execute前加了一层校验不通过就返回错误信息给模型让它重新尝试。第二工具执行要加超时和异常捕获。工具函数可能因为各种原因失败网络问题、文件不存在、除零错误等这些异常必须被捕获并作为工具结果返回给模型而不是让整个程序崩溃。第三工具结果要控制长度。如果工具返回大量文本直接塞进消息历史会迅速消耗上下文窗口。我的做法是对超过一定长度的结果做截断并附上提示说明被截断了。实操心得工具描述的质量直接决定 Agent 的可用性。我建议每个工具的描述都包含三部分功能一句话概括、适用场景说明、每个参数的详细解释。花十分钟写好描述能省掉后面几个小时的调试。3.4 主循环控制器Agent 的心跳主循环是整个 Agent 的引擎。它的逻辑用伪代码表示就是def run_agent(user_input, max_iterations10): messages.append({role: user, content: user_input}) for i in range(max_iterations): response call_model(messages) assistant_msg parse_response(response) messages.append(assistant_msg) tool_calls assistant_msg.get(tool_calls, []) if not tool_calls: return assistant_msg[content] for tc in tool_calls: result execute_tool(tc[name], tc[arguments]) messages.append({ role: tool, tool_call_id: tc[id], content: str(result) }) return 达到最大迭代次数任务未完成看起来简单但有几个关键决策点最大迭代次数设多少我默认设 10。太少了复杂任务做不完太多了万一模型陷入死循环会浪费大量 token。实际使用中可以根据任务复杂度调整简单问答 3 到 5 次足够复杂的数据处理任务可能需要 15 到 20 次。一次响应中有多个工具调用怎么处理我的做法是全部执行然后把所有结果都追加到消息历史中。模型下一轮会看到所有工具的结果再决定下一步。这里要注意执行顺序——如果工具之间有依赖关系模型通常会在不同轮次分别调用同一轮内的多个工具调用一般是无依赖的。循环终止后如果任务没完成怎么办我选择返回一个明确的提示而不是静默失败。这样使用者知道是迭代次数不够可以调整参数重试。4. 完整实操过程与代码实现4.1 环境准备与依赖安装整个项目只需要一个 HTTP 客户端库来调用模型接口其他都是标准库。我用的是requests你也可以用httpx或任何你习惯的库。pip install requests项目结构如下mini-agent/ ├── agent.py # 主循环控制器 ├── message.py # 消息管理器 ├── model.py # 模型接口层 ├── tools.py # 工具注册与调度 └── main.py # 入口我刻意保持文件数量少每个文件职责单一。这样你读代码时不需要在多个文件之间跳来跳去。4.2 模型接口层的完整实现先定义配置# model.py import json import requests class ModelClient: def __init__(self, api_base, api_key, model_name): self.api_base api_base.rstrip(/) self.api_key api_key self.model_name model_name def chat(self, messages, toolsNone, temperature0.1): payload { model: self.model_name, messages: messages, temperature: temperature, } if tools: payload[tools] tools payload[tool_choice] auto resp requests.post( f{self.api_base}/chat/completions, headers{ Authorization: fBearer {self.api_key}, Content-Type: application/json }, jsonpayload, timeout60 ) resp.raise_for_status() return resp.json()这里temperature默认设 0.1原因前面说过Agent 场景需要稳定输出。tool_choice设为auto让模型自己决定是否调用工具。如果你希望强制模型在特定场景必须调用工具可以设为required但一般不建议因为有些问题确实不需要工具。响应解析部分def parse_response(self, raw_response): choice raw_response[choices][0] message choice[message] result {role: assistant, content: message.get(content) or } if message.get(tool_calls): result[tool_calls] [] for tc in message[tool_calls]: try: args json.loads(tc[function][arguments]) except (json.JSONDecodeError, KeyError): args {} result[tool_calls].append({ id: tc[id], name: tc[function][name], arguments: args }) return result4.3 工具系统的完整实现工具基类和注册机制# tools.py import json import time import math class Tool: def __init__(self, name, description, parameters, func): self.name name self.description description self.parameters parameters self.func func def to_schema(self): return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters } } def execute(self, arguments): # 参数校验 required self.parameters.get(required, []) for param in required: if param not in arguments: return f错误缺少必要参数 {param} try: return self.func(**arguments) except Exception as e: return f工具执行出错{str(e)} # 全局注册表 _registry {} def register(tool): _registry[tool.name] tool def get_tool(name): return _registry.get(name) def all_schemas(): return [t.to_schema() for t in _registry.values()]内置工具的实现def calc(expression: str) - str: 安全的数学计算 allowed set(0123456789-*/.() ) if not all(c in allowed for c in expression): return 错误表达式包含不允许的字符 try: result eval(expression, {__builtins__: {}}, {}) return str(result) except Exception as e: return f计算错误{str(e)} register(Tool( namecalculator, description执行数学计算。当用户需要进行算术运算时使用此工具。参数 expression 是一个数学表达式字符串例如 2 3 * 4。, parameters{ type: object, properties: { expression: { type: string, description: 要计算的数学表达式支持加减乘除和括号 } }, required: [expression] }, funccalc )) def get_time() - str: return time.strftime(%Y-%m-%d %H:%M:%S) register(Tool( nameget_current_time, description获取当前日期和时间。当用户询问现在几点、今天日期等问题时使用。无需参数。, parameters{type: object, properties: {}, required: []}, funcget_time )) def text_stats(text: str) - str: chars len(text) words len(text.split()) lines text.count(\n) 1 return f字符数{chars}词数{words}行数{lines} register(Tool( nametext_stats, description统计文本的字符数、词数和行数。当用户需要分析文本长度时使用。参数 text 是要统计的文本内容。, parameters{ type: object, properties: { text: { type: string, description: 需要统计的文本内容 } }, required: [text] }, functext_stats ))注意calc函数里我用了eval但做了字符白名单过滤并且把__builtins__设为空字典。这是最小化的安全措施生产环境建议用更严格的表达式解析器。4.4 主循环的完整实现# agent.py from model import ModelClient from tools import get_tool, all_schemas class Agent: def __init__(self, client, system_prompt, max_iterations10): self.client client self.system_prompt system_prompt self.max_iterations max_iterations self.messages [] def reset(self): self.messages [{role: system, content: self.system_prompt}] def run(self, user_input): if not self.messages: self.reset() self.messages.append({role: user, content: user_input}) for i in range(self.max_iterations): raw self.client.chat( messagesself.messages, toolsall_schemas() ) assistant_msg self.client.parse_response(raw) self.messages.append(assistant_msg) tool_calls assistant_msg.get(tool_calls, []) if not tool_calls: return assistant_msg[content] for tc in tool_calls: tool get_tool(tc[name]) if tool is None: result f错误未知工具 {tc[name]} else: result tool.execute(tc[arguments]) self.messages.append({ role: tool, tool_call_id: tc[id], content: str(result) }) return 抱歉任务处理步骤过多未能完成。请尝试简化你的请求。入口文件# main.py from model import ModelClient from agent import Agent client ModelClient( api_base你的接口地址, api_key你的密钥, model_name你的模型名称 ) agent Agent( clientclient, system_prompt你是一个乐于助人的助手。你可以使用提供的工具来帮助回答问题。当需要计算时使用计算器工具当需要知道时间时使用时间工具。回答要简洁准确。, max_iterations10 ) while True: user_input input(你) if user_input.lower() in (exit, quit): break reply agent.run(user_input) print(f助手{reply})4.5 运行效果与关键日志分析跑起来之后我测试了几个典型场景。第一个是简单计算“帮我算一下 123 乘以 456 等于多少”。Agent 的行为是模型返回工具调用请求参数为{expression: 123 * 456}工具执行返回56088模型收到结果后生成最终回复“123 乘以 456 等于 56088”。第二个是复合任务“现在几点了另外帮我统计一下‘hello world’有几个字符”。这一轮模型返回了两个工具调用分别调用时间工具和文本统计工具。两个工具都执行完后模型综合两个结果给出了完整回复。第三个是边界测试“帮我查一下明天的天气”。由于我没有注册天气工具模型没有调用任何工具直接回复说它没有查询天气的能力。这说明系统提示和工具描述共同约束了模型的行为边界。注意如果你发现模型该调用工具的时候不调用首先检查工具描述是否清晰其次检查系统提示里是否明确引导了工具使用。这两个地方调整之后工具调用率通常会有明显改善。5. 常见问题与排查技巧实录5.1 工具调用失败排查表现象可能原因排查方法解决方案模型不调用工具工具描述不清晰检查 description 是否说明了使用场景补充场景说明和参数解释工具调用参数为空参数 schema 缺少 required检查 parameters 定义添加 required 字段工具结果模型不识别tool_call_id 不匹配打印消息历史检查配对确保 id 严格对应JSON 解析失败模型返回格式异常打印原始 arguments 字符串加 try_fix_json 兜底循环不终止模型反复调用同一工具打印每轮工具调用记录设最大迭代次数优化提示上下文超限历史消息太长统计 token 数实现消息截断逻辑5.2 我踩过的三个典型坑第一个坑工具结果消息格式不对。我最初把工具结果作为role: user的消息追加结果模型完全无法理解这是工具返回的内容表现得很混乱。后来改成role: tool并带上tool_call_id才正常。这个细节在文档里往往一笔带过但实际影响巨大。第二个坑系统提示写得太长。我一开始在系统提示里塞了大量规则和示例结果模型反而变得畏手畏脚该调用工具的时候不调用。后来精简到只保留核心角色定义和工具使用原则效果反而更好。系统提示不是越长越好关键是信息密度。第三个坑没有处理模型返回空内容的情况。有时候模型既没有返回文本内容也没有返回工具调用就是一个空回复。这种情况如果不处理程序会卡住或者返回空字符串。我的处理方式是如果连续两次收到空回复就返回一个提示信息并终止循环。5.3 提升 Agent 稳定性的几个实用技巧技巧一给工具调用加日志。每次工具调用都记录工具名、参数、执行结果和耗时。出问题的时候日志是最快的排查手段。我用一个简单的列表存这些记录调试时打印出来一目了然。技巧二参数校验要宽松但有底线。模型可能传入字符串类型的数字而不是数字类型这种可以自动转换。但如果缺少必要参数必须返回错误让模型重试不能默默用默认值填充。技巧三设置合理的超时。模型调用和工具执行都要设超时。模型调用我设 60 秒工具执行设 30 秒。超时后返回错误信息给模型让它决定是重试还是换一种方式。技巧四用低 temperature 加少量示例。在系统提示里给一两个工具调用的示例能显著提升模型输出格式的正确率。示例不需要多一个简单场景加一个复合场景就够了。5.4 扩展方向与后续优化思路这个最小 Agent 跑通之后可以往几个方向扩展。一是加记忆持久化把消息历史存到文件或数据库支持跨会话的上下文。二是加流式输出让模型回复逐字显示提升交互体验。三是加多轮工具调用的依赖处理比如前一个工具的输出作为后一个工具的输入。四是加工具的动态加载从配置文件或目录自动发现和注册工具。每个方向都可以独立展开但核心循环的逻辑不需要大改。这也是薄抽象设计的好处——扩展点清晰改动范围可控。我个人在实际操作中的体会是写 Agent 最难的不是代码本身而是对模型行为的理解和引导。同样的工具定义换一种描述方式模型的调用准确率可能从 60% 跳到 90%。这需要反复测试和调整没有一劳永逸的方案。建议你在自己的场景里多准备一些测试用例每次调整提示或工具描述后跑一遍用数据来指导优化方向。