企业官网建设流程全解析

1. 为什么偏偏拿 pi-agent 下手：一次"拆玩具"式的学习路径

我之前在好几个团队里都被问过同一个问题：想学 Agent 开发，到底该从 LangChain 看起还是直接啃论文？每次我都建议对方先去找一个结构足够小、但又五脏俱全的开源项目，把它当成乐高玩具一样拆开，拼回去，再拆一次。我自己就是这么过来的，而 pi-agent 是我拆过最顺手的一个参照物。

先说清楚我能提供的核心结论：Agent 开发不等于调用 LLM 接口。它真正的复杂度藏在"循环控制""工具编排""记忆管理"这三件事上。这三件事任何一个做得稀烂，模型再好都会表现成一个大号聊天机器人。参照 pi-agent 实现一个小 Agent 的好处是，你能在几千行代码范围内看到整套闭环是怎么流转的，而不是一头扎进几十万行的大型框架里。

网上关于"Agent 是什么"的热搜基本常年霸榜，但大家的困惑其实集中在几个具体问题上：Agent 和普通对话接口的区别到底在哪？Agent 的循环是谁在控制？工具调用怎么才能不崩？记忆怎么才能在有限上下文里塞下更多信息？pi-agent 这个项目恰好每道题都给了答案，只是答案需要你自己去源码里捞。

这篇文章不打算逐行解读 pi-agent 的代码，而是把我"剥开"它之后重写一个迷你版的过程完整讲一遍。项目正文为空的处境其实挺常见的——很多时候我们不是先有需求再有方案，而是先接触到一个让人手痒的项目，再琢磨怎么把它学透。所以本文的路线是：先讲 Agent 循环的本质，再给一个可运行的最小实现，然后逐个攻破工具系统和记忆设计，最后聊聊我跑完一轮 Demo 之后踩到的工程化坑。适合的人群是那种已经能熟练写 Python，但对 Agent 内部机制还是"用的时候会、看源码就懵"的朋友。

2. 剥到第一层：Agent 循环的本质是什么、哪些环节缺一不可

把 pi-agent 拆开第一眼，你会发现它根本没整什么玄学。核心就一个 while 循环：把当前消息列表交给模型，模型决定是直接回答还是调用某个工具，如果是调用工具，就执行工具、把结果塞回消息列表，再交给模型，直到模型给出最终答案。

这个"再交给模型"就是 Agent 和普通接口最大的分水岭。普通接口是单轮问答，模型给完答案就结束了；Agent 是多轮自驱，模型可以在一次任务里连续决策多次，每次决策都基于上一次行动的反馈。

2.1 Agent 循环的最小骨架

我把 pi-agent 里的循环剥出来，简化成下面这个骨架：

def run_agent(task: str, max_steps: int = 10): messages = [{"role": "user", "content": task}] step = 0 while step < max_steps: response = llm.chat(messages, tools=tool_schemas) step += 1 if response.tool_calls: for call in response.tool_calls: result = execute_tool(call.name, call.arguments) messages.append({ "role": "tool", "tool_call_id": call.id, "content": result }) else: # 模型不再要求调用工具，输出最终答案 return response.content return "达到最大步数，强制结束"

这段代码看着简单，但它揭示了 Agent 三个隐藏的核心机制：

• 轮次上限：max_steps不是可选项。模型在复杂任务里完全可能陷入死循环，如果没有步数限制，你的 token 费用会先把你劝退。
• 消息历史的累积：每一步的工具返回都作为新消息加入列表。这既是 Agent 能持续推理的原因，也是后面上下文爆炸问题的源头。
• 结构化工具调用：tool_calls是从模型返回里单独解析出来的结构，而不是让模型在文本里自己写"我要调用 xx 函数"。这一点很多新手会看漏，导致程序无法稳定判断模型意图。

2.2 模型感知与循环控制的解耦

pi-agent 还有个让我印象深刻的细节：它把"模型能不能感知到工具"和"模型要不要调用工具"拆成了两套配置。第一套是tools参数，它决定模型在回答时"知道有哪些工具可用"；第二套是循环里的终止条件，它决定模型什么时候"结束探索给出答案"。

这两件事混在一起写，是新手最容易踩的坑。比如有人会在 system prompt 里写"你可以调用以下工具"，却没有把工具的 JSON Schema 传给模型 API，结果模型一本正经地编造出根本不存在的函数名。反过来，传了工具定义但没写清"必须调用工具才能回答"这类约束，模型又会自作聪明地猜答案。

我参照 pi-agent 的实现后，把这两件事彻底分开管理：工具注册表管"有没有"，循环逻辑管"用不用"。工具注册表决定模型能看到哪些函数的定义，循环逻辑决定工具执行完之后怎么处理结果。边界清晰之后，排查问题会容易得多——模型不回工具调用，先查注册表；工具执行报错，再查循环里的异常处理。

3. 动手实现前的架构取舍：模型层直接裸调，还是用框架封装

在写第一行业务代码之前，必然要回答一个问题：Agent 核心逻辑到底应该基于某个现成框架，还是直接用模型 SDK 裸写。

我最初的冲动是直接用 LangChain，因为它的AgentExecutor看起来天然支持循环、记忆、工具。但参照 pi-agent 的做法之后我改了主意。pi-agent 的依赖极轻，核心逻辑里甚至没有强制要求你必选某个编排框架，更像是"模型 SDK + 你自己的循环控制"。

3.1 我最终敲定的技术选型

很多人会劝你"别重复造轮子"，但学习型项目恰恰应该自己造一遍轮子。你亲手写一次循环，再去看 LangChain 的AgentExecutor，会发现它的每一步你都知道在干什么；反过来，直接上手框架，你会被callback、intermediate_steps、memory这些抽象淹没。

3.2 为什么不建议第一版就上 Dify/CrewAI

Dify 和 CrewAI 都很优秀，但它们解决的问题和你现在要解决的问题不一样。Dify 更像工作流平台，帮你把 Agent 当作应用编排起来；CrewAI 则偏多角色协作，几个 Agent 互相配合完成任务。当你还在纠结"单个 Agent 的循环为什么跑得不对"时，这些框架的抽象层级反而成了障碍。

pi-agent 里的做法就朴素得多：一个进程里跑一个循环，工具就是普通函数，没有角色扮演、没有子 Agent 通信。这种朴素恰恰是学习的好土壤，先练好单 Agent 的肌肉记忆，再扩展多 Agent 协作，路径会顺很多。

3.3 关于 Rust 实现的一句话吐槽

热词列表里有一项是"基于 Rust 语言 AI Agent"，我懂这种追求极致性能的心情。但如果你连 Agent 循环都不熟，Rust 的所有权模型会把你按在地上摩擦。我自己用 Python 把循环逻辑和工具系统理清楚之后，才有底气去看 Rust 版的设计。语言只是载体，循环和记忆的抽象不分语言。

4. 从零实现核心循环：消息构造、工具调用返回、终止条件

这一章的代码就是你在第 2 节看到的骨架的完整版。我会补上所有细节，让它可以直接运行。

4.1 工具注册表

先定义工具注册表，让每个函数都能被模型发现并执行：

import inspect import json from typing import Any, Callable, Dict, List class ToolRegistry: def __init__(self): self._tools = {} def register(self, func: Callable): """通过解析函数签名生成 JSON Schema""" sig = inspect.signature(func) parameters = { "type": "object", "properties": {}, "required": [] } for name, param in sig.parameters.items(): if name in ("context", "llm"): continue parameters["properties"][name] = { "type": "string" if param.annotation is str else "number" } if param.default is inspect.Parameter.empty: parameters["required"].append(name) self._tools[func.__name__] = { "function": func, "schema": { "type": "function", "function": { "name": func.__name__, "description": func.__doc__ or "", "parameters": parameters } } } return func def get_schemas(self) -> List[Dict[str, Any]]: return [v["schema"] for v in self._tools.values()] def execute(self, name: str, arguments: str) -> str: tool = self._tools[name] # arguments 由模型传回，是 JSON 字符串 kwargs = json.loads(arguments) result = tool["function"](**kwargs) if not isinstance(result, str): result = json.dumps(result, ensure_ascii=False) return result

这段代码的亮点在于用inspect自动生成 Schema，不用手写 JSON 定义。对于学习型项目来说足够了。注意我特别跳过了名为context和llm的参数，这为后面给工具注入上下文对象留了接口。

4.2 Agent 类的循环

class MiniAgent: def __init__(self, registry: ToolRegistry, system_prompt: str = ""): self.registry = registry self.system_prompt = system_prompt or ( "你是一个可靠的助手。如果需要获取外部信息，必须调用提供的工具。" "工具返回结果后，根据结果继续推理，最终给出简洁答案。" ) def chat(self, user_message: str, max_steps: int = 10) -> str: messages = [{"role": "system", "content": self.system_prompt}] messages.append({"role": "user", "content": user_message}) for step in range(max_steps): response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=self.registry.get_schemas(), tool_choice="auto", ) message = response.choices[0].message messages.append(message) # 注意这里要保留模型的完整消息 if not message.tool_calls: return message.content # 执行所有并行工具调用 for tool_call in message.tool_calls: try: result = self.registry.execute( tool_call.function.name, tool_call.function.arguments ) except Exception as e: result = f"工具执行异常: {e}" messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) return "达到最大步数，提前终止。"

如果你跑过 OpenAI 的工具调用接口，应该能感受到messages.append(message)这行有多关键。模型返回的message对象里包含tool_calls结构，后续必须原样送回 API，否则模型会失去上下文关联。而工具结果必须挂在tool_call_id上，这是协议规定的关联方式，缺失任何一个字段请求都会报错。

4.3 一个带工具的完整示例

registry = ToolRegistry() @registry.register def get_weather(city: str) -> str: """查询指定城市的当前天气。""" # 这里替换成真实天气 API 请求 return f"{city}今天晴转多云，气温 22~30℃，东风 2 级。" @registry.register def calculate(expression: str) -> str: """计算数学表达式，如 '1+2*3'。""" return str(eval(expression)) agent = MiniAgent(registry) print(agent.chat("北京天气怎么样？"))

模型会先回调get_weather("北京")，拿到返回后组织成自然语言回答整个流程走完大概需要 3 到 5 次模型调用。第一次调用是模型判断需要工具，第二次是拿到工具结果后的总结，这中间你什么都没额外配置，Agent 的智能化完全是"循环 + 结构化调用"的自然涌现。

4.4 终止条件为什么要单独设计

我见过很多人在循环里只写一个while True，结果模型在某个任务上反复调用同一个只读工具，消耗大量真实 API 费用。参照 pi-agent 的做法，终止条件至少要考虑三个：

• max_steps硬限制，防止死循环；
• 如果工具调用连续 N 次没有产生新的有效信息（比如重复查同一个城市天气），要主动中断;
• 工具返回明显异常时，要给模型一次修正机会，但不能无限重试。

这三个条件的判断逻辑我会放在循环的最外层，因为它们属于"任务生命周期"控制，不应该混进工具本身。

5. 工具系统：让 Agent 拥有"手"的关键设计与打磨

交易所在第 4 章给出的注册表能跑通，但在实际使用中马上会遇到更深层的问题。这些问题我在跑多个任务之后才陆续发现。

5.1 工具描述才是模型调用准确率的胜负手

同样是get_weather函数，描述写"查询天气"和"查询指定城市的当前天气，返回气温、风力和降水概率"是不一样的效果。description字段直接影响模型何时选择这个工具以及如何填充参数。pi-agent 源码里对每个函数的描述写得非常细致，我当时不理解，后来做对比测试才明白，描述越具体，模型的参数猜测越准确。

建议把描述当成产品文案来写：说明功能边界（查天气 vs 查历史天气）、说明参数含义（city是城市中文名）、说明返回内容（文本便于直接回答或进一步加工）。

5.2 参数校验不能只靠模型的自觉

模型即使看到 JSON Schema，偶尔也会传错格式：把int传成字符串、漏掉必填字段、传入超出枚举范围的值。这时工具执行层要兜底：

def execute(self, name: str, arguments: str) -> str: tool = self._tools[name] kwargs = json.loads(arguments) # 对 kwargs 做类型和必填项校验 for req in tool["schema"]["function"]["parameters"].get("required", []): if req not in kwargs: return f"参数缺失: {req}，请检查调用" try: result = tool["function"](**kwargs) except TypeError as e: return f"参数类型错误: {e}" ...

关键点在于：校验失败时返回给模型的不是抛异常，而是描述性的错误文本。因为 Agent 的循环机制决定了一切结果最终都会送回模型，模型读取错误文本后可以自行修正参数。这比自己写一堆 if-else 去纠错要省事得多，也让 Agent 保持"能自我修复"的特性。

5.3 并行工具调用与副作用

新的 API 允许一次返回多个tool_calls，这个能力非常实用。比如"查询北京和上海两地天气并对比"，模型可能在同一轮里并行请求两个get_weather。

但并行带来一个副作用风险：如果这些工具里有一个是"下单支付"，而另一个是"查询余额"，并行执行顺序是不稳定的。pi-agent 对这个问题的处理很干脆——它给工具打上了串行/并行标注。我只在注册表里增加了一个简单字段：

def register(self, func: Callable, *, parallel: bool = True): ...

默认并行，但对于写操作、有状态操作，手动设为parallel=False，在执行循环里遇到非并行工具就临时退化成串行处理。这个细节让我避免了好几次"并发扣款"级别的灾难。

5.4 工具隔离与安全边界

Agent 的工具本质上是把网络请求、文件读写、命令执行能力交给模型调度。如果工具是execute_shell(cmd: str)这种，你要清楚这意味着什么——模型完全可能因为 prompt 注入或幻觉执行出危险命令。我的习惯是第一版不接入任何 shell 类工具，只暴露无副作用的查询函数。等到需要写文件时，也严格限制路径在特定沙箱目录内。

6. 记忆与上下文窗口：怎么让 Agent 记住更多又不爆 token

所有人跑通 Agent 第一件事都会遇到同一个问题：连续对话几轮之后，消息列表越来越长，成本直线上升，而且模型开始忽略早期信息。这就是 Agent 记忆设计的核心矛盾。

6.1 消息历史的三种基本策略

参照 pi-agent 的进化路径，我首先实现的是摘要压缩。核心做法是：当消息数超过阈值（比如 20 条），用一次模型调用把历史消息压缩成摘要，然后清空历史，把摘要作为一条 system 消息塞进列表。

def compact_messages(messages, summary=""): # 触发条件：消息超过阈值 if len(messages) < 20: return messages, summary # 把所有历史消息合并成一段文本 history_text = "\n".join( f"{m['role']}: {m['content']}" for m in messages ) prompt = ( "将以下对话历史压缩为摘要，保留关键事实、用户意图、" "已经执行过的工具和结果。不要遗漏重要约束。\n\n" f"历史对话:\n{history_text}" ) new_summary = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}] ).choices[0].message.content return [ {"role": "system", "content": f"历史摘要: {new_summary}"} ], new_summary

这个方案在真实场景里立竿见影——上下文从动不动多出的几千 token 降到稳定几百。

6.2 长任务的跨会话记忆

热搜词里的"agent记忆"大多指向跨会话持久化。用户上一轮问了一个问题，今天重新打开应用还想接着聊，这需要把摘要持久化到数据库。我在项目里用了一个轻量的 SQLite 表：

CREATE TABLE agent_memory ( id INTEGER PRIMARY KEY, session_id TEXT NOT NULL, summary TEXT NOT NULL, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP );

每次摘要压缩后更新 session 对应的 summary 字段。新会话启动时，把这个摘要作为初始 system 消息注入。这样 Agent 就有了"你上次聊到哪了"的基本记忆，成本只需一次数据库查询。

6.3 token 用量换算的实际感觉

很多刚接触 Agent 的人会低估 token 消耗。一次包含工具调用的完整对话，仅仅是把全部消息送去模型就已经消耗了几千 token，再加上模型输出的工具调用参数、工具返回，一次"思考"成本远超普通聊天。做多轮 Agent 时，我把摘要压缩阈值调低到 12 条消息，并且每轮对话之后打点记录 token 增量，看到数字就会对成本有真实体感。

7. 并发与稳定性：网上都在问的"过 Demo 阶段"要面对什么

热搜里有一项是"AI Agent 怎么扛并发"，这个问题并非要把设计复杂度全部拉满，但要清楚 Agent 服务和普通 HTTP 服务在并发模型上的差异，并有针对性地处理。

7.1 Agent 服务并发的本质瓶颈

普通 Web 接口的瓶颈在数据库和业务逻辑；Agent 服务的瓶颈则复杂得多，模型 API 的响应时间本来就有波动，一个复杂任务可能要调 3-5 次模型 API，中间还穿插工具执行，单次请求的总耗时可能是秒级到分钟级。

如果按照传统 Flask 线程模型不加限制地接收请求，很快会爆发两个问题：

• 模型 API 的并发限流被打穿，大量请求 429；
• 每个请求的内存和上下文状态撑爆后端。

我的做法是：Agent 执行体自身通过队列调度，不直接暴露给用户任意创建任务。实现方式是给每个用户会话创建一个"任务队列"，Agent 串行消费队列里的消息，不同用户之间可以并发，同一个用户的请求宁愿排队也不并行处理。

7.2 超时、重试与幂等

工具调用天然有失败概率：天气 API 可能超时、文件可能读取失败。循环里的容错办法我前面已经提过，但工程化的稳定还需要额外两层。

第一层是模型调用本身的超时重试。OpenAI SDK 支持timeout参数，建议设置timeout=60并且对网络错误做指数退避重试，最多 3 次。第二层是工具侧的幂等设计：写操作要带唯一请求 ID，重复执行不会造成重复影响。 第三层是任务超时熔断。一个 Agent 任务如果 5 分钟还没跑完，大概率卡死或者上下文爆炸了，应该直接被标记为失败并释放资源。

7.3 结构化日志是 Agent 排障救命稻草

Agent 的每次循环都涉及模型调用、工具结果、消息变化，排错难度远超普通接口。我给循环的每个关键节点加结构化日志：

logging.info({ "step": step, "model_input_tokens": usage.prompt_tokens, "model_output_tokens": usage.completion_tokens, "tool_calls": [c.function.name for c in message.tool_calls] if message.tool_calls else None, "final_answer": message.content if not message.tool_calls else None })

有了这个日志，线上出问题时可以在几秒内定位——是模型进入循环了？是某个工具迟迟没返回？是 token 超限导致的截断？顺着时间线把每步日志拉出来，很多诡异问题都会现形。

8. 参照 pi-agent 的进阶玩法：Harness 与框架对比思考

热搜词里同时出现了"agent harness和agent区别"以及"agent框架如langchain、dify、crewai等哪个好"。在学习型项目跑通之后，这些概念其实可以串起来看。

8.1 当你开始写 harness，你才真正理解了 Agent 周期

很多资料把 harness 翻译成"装配"或者"框架"，我理解它就是"控制 Agent 循环、记忆、工具调用的那一整套外壳"。你照着 pi-agent 实现了自己的 MiniAgent，其实就是在造一个微型 harness。

当你开始思考"我这个 harness 能不能让子 Agent 复用父 Agent 的工具？""能不能在循环中间插入一个人工确认环节？""能不能把某一步的回溯改成分支探索？"——这些问题的答案都写在 harness 的接口设计上。相比盲目对比 LangChain 好不好用，先把 harness 拆明白再去看框架，每一家设计的好与鸡肋你都能看懂个七七八八。

LangChain 强在抽象生态全，但学习成本高；Dify 强在可视化编排和团队协作；CrewAI 强在角色协作，但单 Agent 的深度控制并不突出。如果你已经能徒手实现一个 Agent 循环，再去选择框架就有明确标准：它能让我少写哪些代码？它有没有限制我的循环控制自由度？它能和我现有的工具系统无障碍打通吗？这几个问题比我列一堆框架特性对比表更有实际价值。

8.2 Skill 机制的启发

热词里有一项是"agent skill"，我看到 pi-agent 里真的把每个工具打包成了一个"技能"——函数、描述、示例参数全部绑定。这种设计比裸函数更接近真实 Agent 应用：技能可以被学习和复用，也可以挂到特定角色下发布。如果你想做更高级的 Agent，提前把工具设计升级成技能包，会让后续的扩展优雅不少。

9. 跑通之后的下一步：我的体会与小建议

整个项目玩下来，我最深的体会是：Agent 开发入门的关键不是模型选得多好，而是你能否把"循环、工具、记忆"这三者的边界在代码里控制清楚。很多人被 LangChain 的 Agent 搞得很挫败，根本原因不是 LangChain 差，而是他们没有亲手建立过这三者的心智模型。参照 pi-agent 自己实现一遍，这个心智模型就内化成了肌肉记忆。

最后分享几个实际经验：

• 第一版尽量用 Python，不要一边学概念一边折腾类型系统。理解循环之后再考虑 Rust 等更高性能的移植。
• 每实现一个功能，立刻写一段测试让它真的调用一次工具。不要只测"模型能回答"，要测"模型能在需要时正确选择工具并处理异常结果"，这两件事覆盖了 Agent 90% 的稳定性问题。
• 日志和监控千万别省，Agent 的不可预测性决定了你一定会需要一个"重播"机制，把线上失败的完整消息轨迹拉出来复现。
• 预算控制尽早做，我在摘要压缩和步数上限都踩过坑，第一次跑长任务的时候 token 消耗直接超了预期两倍，后来才通过实时计数器和阈值控制找回平衡。

这个迷你版 Agent 项目我至今还保留着，后续我又给它加了多轮对话总结、技能包加载和新工具类型。如果正处在"听说过 Agent、想动手但不知从哪下手"的阶段，建议你今天就写一个 while 循环的雏形出来——跑了第一个工具调用之后，整个 Agent 世界的大门才算真正打开。