1. 从零理解 Agent 到底在做什么
很多人第一次接触 Agent 这个概念时,脑子里浮现的都是科幻电影里那种能自己思考、自己行动的智能体。但真到了动手写代码的阶段,反而会陷入一种迷茫:Agent 和普通的函数调用、和一条简单的 API 请求到底有什么区别?我当初也是带着这个疑问去翻了不少开源实现,其中 pi-agent 这个项目的结构给了我很大启发——它没有堆砌复杂的框架,而是用相当克制的代码量把 Agent 的核心骨架讲清楚了。
所谓 Agent,本质上就是一个"能自己决定下一步做什么"的程序循环。你给它一个目标,它不会像普通函数那样一次性返回结果,而是会经历"观察当前状态 → 思考下一步动作 → 执行动作 → 观察执行结果 → 再思考"这样一个反复迭代的过程。这个循环就是 Agent 的心脏。普通程序是"输入→处理→输出"的直线流程,而 Agent 是"输入→循环决策→输出"的螺旋流程。理解这一点,后面所有的代码结构都会变得顺理成章。
pi-agent 这个参照对象之所以值得学习,是因为它把 Agent 拆成了几个非常清晰的模块:一个负责和大模型对话的客户端、一个管理工具调用的调度器、一个维护对话历史的记忆模块,以及一个驱动整个循环的主控逻辑。这四个部分各司其职,没有过度设计,也没有为了炫技而引入不必要的抽象层。对于想真正搞懂 Agent 内部机制的人来说,这种"够用就好"的实现方式反而比那些大而全的框架更有学习价值。
这篇文章适合两类人:一类是已经用过某些 Agent 平台、但对其内部原理一知半解,想自己动手写一个的开发者;另一类是刚接触这个概念,想找一个最小可运行示例来建立直觉的新手。我会从最基础的环境准备讲起,一步步把每个模块的实现逻辑拆开,补充我在实际编码中踩过的坑和总结的技巧。读完之后,你应该能独立写出一个能跑起来、能调用工具、能维持多轮对话的小 Agent,并且清楚每一行代码背后的意图。
需要提前说明的是,Agent 开发涉及的核心技术点包括:大模型 API 的调用方式、工具(Tool)的定义与注册机制、对话上下文的组织与管理、以及循环终止条件的判断。这些概念听起来抽象,但落到代码上其实都很具体。我会尽量用生活化的类比来解释,让没有相关背景的读者也能跟上节奏。
2. 动手前的环境与依赖准备
2.1 运行环境的选择与版本约束
在开始写代码之前,先把运行环境理清楚。我推荐使用 Python 3.10 或更高版本,原因有两个:一是这个版本之后对类型注解的支持更加完善,写工具定义和函数签名时会舒服很多;二是很多大模型官方 SDK 已经把最低版本要求提到了 3.10,用低版本会遇到各种依赖冲突。如果你本地还停留在 3.8 或 3.9,建议先用虚拟环境隔离出一个新版本,不要直接在系统 Python 上折腾。
虚拟环境的创建方式我用的是最朴素的 venv,没有上 conda 或 poetry,因为这个小项目的依赖非常少,没必要引入额外的包管理复杂度。具体操作就是在一个空目录下执行python -m venv .venv,然后激活它。Windows 和 macOS/Linux 的激活命令不一样,这个细节很多人第一次会搞混,我列在下面方便对照。
# macOS / Linux source .venv/bin/activate # Windows PowerShell .venv\Scripts\Activate.ps1 # Windows CMD .venv\Scripts\activate.bat激活之后,命令行前面会出现(.venv)的标识,说明你已经进入隔离环境了。这一步看起来简单,但我见过太多人因为忘了激活虚拟环境,把包装到了全局,后面排查半天才发现问题所在。
2.2 依赖清单与安装策略
这个小 Agent 的依赖可以用"极简"来形容。核心只需要两样东西:一个 HTTP 请求库用来和大模型 API 通信,一个环境变量管理库用来存放密钥。我选的是requests和python-dotenv,前者足够稳定且文档丰富,后者能让我把密钥写在.env文件里而不污染代码。
pip install requests python-dotenv如果你打算用某个官方 SDK 而不是直接发 HTTP 请求,那就把requests换成对应的 SDK 包。但我个人建议第一版先用原生 HTTP 请求来实现,因为这样你能清楚地看到请求体长什么样、响应结构是什么样,而不是被 SDK 封装好的方法名遮住了底层细节。等你把整个流程跑通之后,再换成 SDK 提升开发效率也不迟。
关于密钥管理,这里有个必须强调的安全习惯:永远不要把 API Key 硬编码在代码里,也不要提交到版本控制系统。正确做法是在项目根目录建一个.env文件,把密钥写进去,然后在.gitignore里把.env排除掉。代码里通过os.getenv读取。这个习惯看起来是小事,但一旦密钥泄露,后果可能很严重。
# .env 文件内容示例 LLM_API_KEY=你的密钥 LLM_BASE_URL=你的接口地址 LLM_MODEL=你使用的模型名称注意:
.env文件一定要加入.gitignore,这是最基本的安全底线。我见过有人把密钥直接推到公开仓库,几分钟内就被扫描到并产生了大量异常调用。
2.3 项目目录结构的规划
在动手写代码之前,先想清楚文件怎么组织。我的习惯是把不同职责的代码分到不同文件里,哪怕每个文件只有几十行。这样做的好处是,当你后面想替换某个模块(比如把 HTTP 请求换成 SDK、把内存记忆换成数据库存储)时,改动范围是可控的。
我用的结构是这样的:main.py作为入口,负责启动循环;llm_client.py封装和大模型的通信;tools.py定义所有可调用的工具;agent.py是核心的循环逻辑;memory.py管理对话历史。这个划分不是唯一的,但它对应了 Agent 的四个核心关注点,逻辑上很清晰。对于初学者来说,先按这个结构搭起来,后面理解起来会顺畅很多。
3. 大模型通信层的封装细节
3.1 请求体的构造逻辑
Agent 和普通聊天机器人最大的区别在于,它需要告诉大模型"你有哪些工具可以用"。这个信息是通过请求体里的工具描述字段传递的。以常见的对话补全接口为例,请求体大致包含三部分:模型名称、消息列表、工具定义列表。消息列表里既有用户的输入,也有模型之前的回复,还有工具执行的结果。工具定义列表则描述了每个工具的名字、用途和参数结构。
构造请求体时最容易出错的地方是消息角色的区分。通常有四种角色:system用来设定 Agent 的行为准则,user代表用户输入,assistant代表模型回复,tool代表工具执行结果。很多人第一次写的时候会把工具结果当成user消息发回去,导致模型理解混乱。正确的做法是,工具执行完之后,用tool角色把结果追加到消息列表,并且要带上对应的工具调用 ID,这样模型才能把结果和它之前的调用请求对应起来。
def build_request(messages, tools, model): return { "model": model, "messages": messages, "tools": tools, "tool_choice": "auto" }tool_choice这个参数值得说一下。设成auto表示让模型自己决定要不要调用工具;设成none表示强制不调用;也可以指定某个具体工具名强制调用。在 Agent 场景下一般用auto,因为我们希望模型根据当前情况自主判断。但有些时候,比如你明确知道这一步必须查数据,就可以临时改成强制调用,避免模型偷懒直接编答案。
3.2 响应解析与工具调用提取
模型返回的响应结构里,最关键的是判断它到底是想"直接回复"还是"调用工具"。这两种情况的处理路径完全不同。如果模型直接给了文本回复,那这一轮就可以结束了;如果它返回了工具调用请求,那就需要先执行工具,把结果喂回去,再让模型继续。
解析响应时,我习惯先检查有没有工具调用字段。如果有,就遍历每一个调用请求,提取出工具名和参数。参数通常是 JSON 格式的字符串,需要解析成字典才能用。这里有个坑:有些模型返回的参数 JSON 可能不完整或者格式有偏差,直接json.loads会抛异常。稳妥的做法是加一层 try-except,解析失败时把原始字符串作为错误信息返回给模型,让它重新组织参数。
def parse_response(response): choice = response["choices"][0] message = choice["message"] if message.get("tool_calls"): calls = [] for call in message["tool_calls"]: calls.append({ "id": call["id"], "name": call["function"]["name"], "arguments": call["function"]["arguments"] }) return {"type": "tool_call", "calls": calls, "raw": message} return {"type": "text", "content": message.get("content", "")}这个解析函数返回的结构很关键,它把"模型想干什么"这件事抽象成了一个明确的类型标记。主循环拿到这个结果之后,只需要根据type字段做分支处理就行,逻辑非常清晰。这种"先归一化再处理"的思路,在 Agent 开发里会反复用到。
3.3 错误处理与重试机制
和大模型 API 打交道,网络抖动、限流、超时这些问题几乎不可避免。如果不做处理,Agent 跑到一半突然崩掉,体验会非常差。我的做法是在通信层加一个简单的重试逻辑:捕获网络异常和特定的状态码,等待一小段时间后重试,最多重试三次。
重试的间隔我一般用指数退避,也就是第一次等 1 秒,第二次等 2 秒,第三次等 4 秒。这样既能给服务端喘息的时间,又不会让用户等太久。但要注意,不是所有错误都值得重试。比如参数错误(通常是 400 系列状态码)重试多少次都没用,应该直接抛出;而超时和限流(429、500、503 等)才适合重试。
import time def call_with_retry(func, max_retries=3): for attempt in range(max_retries): try: return func() except Exception as e: if attempt == max_retries - 1: raise wait = 2 ** attempt time.sleep(wait)这个重试函数虽然简单,但能挡掉大部分偶发故障。我在实际使用中发现,加上这层保护之后,Agent 因为网络问题中断的概率明显下降。当然,如果你的场景对延迟很敏感,可以把重试次数调低,或者只对特定错误重试。
4. 工具系统的设计与注册机制
4.1 工具的本质:给模型一双能干活的手
大模型本身只能生成文本,它没法直接读文件、查数据库、发请求。工具就是给它装上的"手",让它能通过这些手去操作外部世界。一个工具本质上就是一个函数,加上一段描述这个函数用途和参数的说明。模型根据这段说明来判断什么时候该用哪个工具。
设计工具时,最重要的原则是"描述要清楚,参数要简单"。我见过有人把工具描述写得非常含糊,结果模型要么不用,要么用错。比如一个查询天气的工具,描述写成"获取信息"就太模糊了,应该写成"根据城市名称查询该城市当前的天气状况,包括温度和天气现象"。参数也要尽量用基本类型,避免嵌套过深的结构,因为模型生成复杂 JSON 时出错概率会明显上升。
工具的定义通常包含三个部分:名称、描述、参数 schema。参数 schema 用的是 JSON Schema 格式,描述每个参数的类型、含义和是否必填。这个 schema 会直接进入请求体发给模型,所以它的质量直接影响模型的调用准确率。
weather_tool = { "type": "function", "function": { "name": "get_weather", "description": "根据城市名称查询当前天气状况", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,例如北京、上海" } }, "required": ["city"] } } }4.2 工具注册表与调度分发
当工具有多个时,需要一个注册表来管理它们。注册表的作用有两个:一是把所有工具的定义集中起来,方便一次性发给模型;二是提供一个从工具名到实际函数的映射,方便执行时查找。
我用一个字典来实现这个注册表,键是工具名,值是一个包含定义和实现函数的对象。注册的时候同时把定义和函数放进去,这样定义和实现就不会脱节。执行的时候,根据模型返回的工具名去字典里查,找到对应的函数并调用。
class ToolRegistry: def __init__(self): self._tools = {} def register(self, name, definition, func): self._tools[name] = {"definition": definition, "func": func} def get_definitions(self): return [t["definition"] for t in self._tools.values()] def execute(self, name, arguments): if name not in self._tools: return f"错误:未找到名为 {name} 的工具" try: return self._tools[name]["func"](**arguments) except Exception as e: return f"工具执行出错:{str(e)}"这个注册表的设计有个细节值得注意:execute方法捕获了所有异常,并把错误信息作为字符串返回,而不是直接抛出。这样做是因为工具执行失败不应该让整个 Agent 崩溃,而应该把错误信息反馈给模型,让模型决定是重试、换个工具,还是直接告诉用户出了问题。这种"错误也是信息"的思路,是 Agent 健壮性的重要保障。
4.3 参数校验与安全边界
模型生成的参数不能无条件信任。虽然大多数时候它是靠谱的,但偶尔会生成类型不对、缺少必填项、甚至包含恶意内容的参数。所以在执行工具之前,最好做一层校验。
校验分两个层次:第一层是结构校验,检查必填参数是否存在、类型是否匹配;第二层是业务校验,比如城市名称是否在支持列表里、数值是否在合理范围内。第一层可以通用化,写一个校验函数根据 schema 自动检查;第二层则需要针对每个工具单独写。
def validate_arguments(schema, arguments): required = schema.get("required", []) for key in required: if key not in arguments: return False, f"缺少必填参数:{key}" properties = schema.get("properties", {}) for key, value in arguments.items(): if key in properties: expected = properties[key].get("type") if expected == "string" and not isinstance(value, str): return False, f"参数 {key} 应为字符串" if expected == "number" and not isinstance(value, (int, float)): return False, f"参数 {key} 应为数字" return True, ""注意:如果你的工具涉及文件操作、命令执行等敏感行为,一定要在工具实现内部再加一层白名单或路径限制,不能只依赖模型生成的参数。模型可能被诱导生成越界的参数,安全边界必须由代码来守。
5. 记忆管理与上下文窗口控制
5.1 对话历史的组织方式
Agent 的"记忆"其实就是对话历史。每一轮用户输入、模型回复、工具调用和工具结果,都要按顺序记录下来,下次请求时一起发给模型。这样模型才能知道之前发生了什么,做出连贯的决策。
消息列表的结构是一个数组,每个元素是一条消息,包含角色和内容。工具调用消息比较特殊,它除了角色和内容,还要带上工具调用 ID。这个 ID 是模型生成调用请求时附带的,工具结果必须用同一个 ID 回复,模型才能对应上。
messages = [ {"role": "system", "content": "你是一个乐于助人的助手,可以调用工具来完成任务。"}, {"role": "user", "content": "帮我查一下北京的天气"}, {"role": "assistant", "content": None, "tool_calls": [...]}, {"role": "tool", "tool_call_id": "call_abc123", "content": "北京今天晴,25度"} ]这个结构看起来简单,但顺序和字段一个都不能错。我调试时遇到最多的问题就是工具结果消息忘了带tool_call_id,导致模型报错说找不到对应的调用。记住一个原则:模型发起的每个工具调用,都必须有一条对应的工具结果消息,不多不少。
5.2 上下文超长的截断策略
对话轮次一多,消息列表就会越来越长,最终会超出模型的上下文窗口限制。这时候必须做截断,否则请求会直接失败。截断的策略有几种,各有取舍。
最简单的是"保留最近 N 条",把最早的消息丢掉。这种策略实现简单,但会丢失早期的重要信息。稍微好一点的是"保留系统消息 + 最近 N 条",因为系统消息通常设定了 Agent 的行为准则,不能丢。再复杂一点的是"摘要压缩",把早期对话用模型总结成一段简短摘要,替代原始消息。这种方式信息保留得更好,但需要额外调用一次模型,成本和延迟都会增加。
我一般先用"保留系统消息 + 最近 N 条"这种简单策略,等实际发现信息丢失影响效果时,再考虑上摘要压缩。不要一上来就搞复杂方案,先用简单的跑通,遇到问题再优化。
def truncate_messages(messages, max_count=20): system_msgs = [m for m in messages if m["role"] == "system"] other_msgs = [m for m in messages if m["role"] != "system"] if len(other_msgs) <= max_count: return messages return system_msgs + other_msgs[-max_count:]5.3 工具结果的精简处理
工具返回的结果有时候会非常长,比如查询数据库返回了几百行记录,或者读取文件返回了整篇文档。这些内容如果原封不动塞进上下文,会迅速吃掉窗口空间。所以工具实现里最好做一层精简,只返回模型真正需要的信息。
精简的原则是"够用就好"。比如查询天气,返回温度和天气现象就够了,不需要把湿度、风速、气压、紫外线指数全带上。查询数据库,如果模型只是想知道有没有符合条件的记录,返回数量就行,不需要返回全部字段。这个判断需要结合具体业务场景,没有通用公式,但核心思路是:站在模型的角度想,它拿到这个结果之后要做什么决策,只给它做决策必需的信息。
我在实际项目里会给每个工具设一个返回长度上限,超过就截断并加上提示。这样即使某个工具返回了意外长的内容,也不会把上下文撑爆。
6. 主循环的驱动逻辑与终止条件
6.1 循环的骨架:观察、思考、行动
主循环是整个 Agent 的发动机。它的逻辑可以用一句话概括:不断重复"把当前消息发给模型 → 解析模型意图 → 如果是工具调用就执行并追加结果 → 如果是文本回复就结束"这个过程。
这个循环的终止条件有两个:一是模型返回了纯文本回复,说明它认为任务完成了;二是达到了预设的最大轮次,防止无限循环。第二个条件非常重要,因为模型有时候会陷入反复调用同一个工具的怪圈,没有上限的话会一直跑下去,既浪费资源又得不到结果。
def run_agent(user_input, max_turns=10): messages.append({"role": "user", "content": user_input}) for turn in range(max_turns): response = llm_client.chat(messages, tools) parsed = parse_response(response) if parsed["type"] == "text": messages.append({"role": "assistant", "content": parsed["content"]}) return parsed["content"] messages.append(parsed["raw"]) for call in parsed["calls"]: result = tool_registry.execute(call["name"], json.loads(call["arguments"])) messages.append({ "role": "tool", "tool_call_id": call["id"], "content": str(result) }) return "达到最大轮次限制,任务未完成"这段代码就是 Agent 的核心。看起来不长,但每一行都有讲究。比如messages.append(parsed["raw"])这一步,是把模型原始的回复消息(包含工具调用信息)追加到历史里,而不是只追加文本内容。如果漏了这一步,模型下一轮就不知道自己刚才调用过工具,会重复调用。
6.2 多工具并行调用的处理
有些模型支持一次返回多个工具调用,比如同时查天气和查汇率。这种情况下,循环里要遍历所有调用,逐个执行,然后把所有结果都追加到消息列表。注意每个结果都要带上对应的tool_call_id,不能搞混。
并行调用能提升效率,但也带来一个问题:如果其中一个工具执行失败,其他工具的结果还要不要?我的做法是照常执行所有工具,失败的返回错误信息,成功的返回结果,让模型自己判断怎么处理。这样比直接中断整个流程要灵活。
6.3 循环中的状态追踪与日志
调试 Agent 时,最痛苦的事情就是不知道它内部到底发生了什么。所以从第一版开始,就应该加上详细的日志。每一轮请求发了什么、模型返回了什么、执行了哪个工具、结果是什么,都打印出来。这些日志在排查问题时是无价之宝。
我习惯用 Python 的 logging 模块,把日志同时输出到控制台和文件。控制台方便实时观察,文件方便事后回溯。日志级别用 DEBUG,把请求体和响应体都记下来。虽然日志会比较多,但调试阶段这点存储成本完全可以接受。
import logging logging.basicConfig( level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("agent.log"), logging.StreamHandler() ] )有了这套日志,当 Agent 行为异常时,你可以直接翻日志看它在哪一步走偏了。是模型理解错了指令,还是工具返回了意外结果,一目了然。这个习惯我强烈建议从第一天就养成,后面会省下大量排查时间。
7. 实测中暴露的典型问题与应对
7.1 模型不调用工具而是直接编答案
这是最常见的问题之一。你明明给了工具,模型却不用,直接凭自己的知识回答。原因通常是工具描述不够有说服力,或者系统提示里没有强调"遇到需要实时信息的问题必须调用工具"。
解决办法有两个:一是在系统提示里明确要求,比如"当问题涉及实时数据、计算或外部信息时,必须调用相应工具,不要凭记忆回答";二是把工具描述写得更具体,让模型清楚知道这个工具能解决什么问题。实测下来,这两招结合使用效果最好。
7.2 工具参数格式错误导致执行失败
模型生成的参数偶尔会不符合预期,比如该传字符串的传了数字,该传数组的传了单个值。这种问题在参数结构复杂时尤其常见。应对方法是在工具执行前做校验,校验失败时把错误信息返回给模型,并在错误信息里说明正确的格式。模型看到错误提示后,通常下一轮就能改对。
如果某个工具频繁出现参数错误,那可能是工具定义本身有问题。检查一下参数描述是否清晰、是否给了示例。有时候加一个example字段,模型的表现会明显改善。
7.3 无限循环与重复调用
模型有时候会陷入"调用工具 → 得到结果 → 再调用同一个工具"的死循环。这通常是因为工具返回的结果没有满足模型的预期,它以为没查到,就反复查。应对方法除了设置最大轮次上限,还可以在工具结果里加上明确的提示,比如"这是查询结果,请基于此结果回答用户,不要重复查询"。
另一个技巧是记录每个工具被调用的次数,如果同一个工具在短时间内被调用超过阈值,就强制中断并返回提示。这种保护机制能有效防止资源浪费。
7.4 上下文膨胀导致响应变慢
随着对话轮次增加,消息列表越来越长,每次请求的 token 数也随之上升,响应时间会明显变长。除了前面说的截断策略,还可以考虑把不常用的工具定义从请求里移除,只保留当前场景可能用到的工具。工具定义本身也占不少 token,精简工具列表能省下可观的空间。
我在一个项目里把工具从二十个精简到五个常用的,请求 token 数直接降了三分之一,响应速度提升很明显。所以工具不是越多越好,够用就行。
8. 从能跑到好用:几个提升体验的细节
8.1 流式输出让等待不再焦虑
Agent 处理复杂任务时,可能要跑好几轮才出结果,用户干等着体验很差。如果模型接口支持流式输出,可以把模型的文本回复实时打印出来,让用户看到进度。虽然工具调用阶段没法流式,但至少最终回复能边生成边显示,感知上的等待时间会短很多。
实现流式输出需要把请求参数里的stream设为true,然后逐块读取响应。每一块是一个增量,需要拼接起来。这个改动不大,但对体验的提升很明显。
8.2 给工具加上超时控制
有些工具(比如网络请求、数据库查询)可能因为各种原因卡住,如果不设超时,整个 Agent 就会一直等下去。给每个工具的执行加上超时限制,超时后返回错误信息,让模型决定下一步。这样即使某个工具出问题,也不会拖垮整个流程。
超时时间根据工具类型来定,本地计算类可以短一些,网络请求类可以长一些。我一般设 10 到 30 秒,具体看场景。
8.3 把常用配置抽成参数
模型名称、最大轮次、超时时间这些配置,不要写死在代码里,抽成参数或配置文件。这样换模型、调参数时不用改代码,改配置就行。对于需要频繁试验不同配置的场景,这个习惯能省很多事。
我通常用一个config.py或者.env文件来管理这些配置,代码里通过读取配置来使用。这样同一套代码可以在不同环境跑不同的配置,灵活性高很多。
8.4 为工具编写独立的测试
工具是 Agent 和外部世界交互的接口,一旦工具有 bug,Agent 的行为就会变得不可预测。所以每个工具都应该有独立的单元测试,验证正常输入和异常输入下的行为。这样在集成到 Agent 之前,就能确保工具本身是可靠的。
测试工具时,重点覆盖边界情况:空输入、超长输入、类型错误的输入、以及工具内部可能抛异常的场景。把这些都测过,Agent 跑起来才踏实。
9. 我对 Agent 开发的一点个人体会
写到这里,一个能跑的小 Agent 基本就成型了。回头看整个过程,我觉得最有价值的不是某段具体代码,而是对"循环"和"边界"这两个概念的理解。Agent 的核心就是一个循环,而让这个循环稳定运行的关键,是给每个环节设好边界:工具执行的边界、上下文长度的边界、循环轮次的边界、错误处理的边界。边界设好了,Agent 就不会失控。
另外一点体会是,不要一开始就追求功能大而全。我见过不少人一上来就想做一个能处理各种任务的通用 Agent,结果工具越加越多,逻辑越来越复杂,最后连基本的对话都跑不稳。正确的做法是先做一个只能调用一两个工具的最小版本,把它跑通、跑稳,然后再逐步扩展。每加一个工具,都要重新验证整个流程,确保没有引入新的问题。
Agent 开发这个方向变化很快,新的模型、新的接口、新的模式层出不穷。但底层的这套循环逻辑和工程思路是相对稳定的。把这套基础打牢,后面不管上层怎么变,你都能快速跟上。希望这篇内容能帮你建立起对 Agent 内部机制的清晰认知,少走一些我当年走过的弯路。