做 LLM 应用开发这两年，我最大的感受是：模型层永远比上层逻辑变化得快。今天接一个闭源接口，明天换一个开源权重，后天又要兼容本地部署的量化版本，整套业务代码被 API 差异拖得越来越重。同时，Agent 的编码能力越强，越不满足于按我预设的工具列表去执行任务——它最好能自己写代码、自己注册工具、自己扩展边界。这就是我把内部项目整理成 Pi Agent Harness 并开源出来的原因。它做两件事：将多种 LLM API 统一成一个稳定入口，同时给 Agent 提供一套“自扩展编码工作”的运行时，让 Agent 能根据任务动态生成并注册新工具，而不是只能调用静态函数。这篇文章我会从设计动机、架构实现到落地踩坑，完整复盘这个开源实践。
1. 项目定位：为什么不是又一个 LLM 封装库
1.1 现有工具链的碎片化现状
我见过很多团队在接入大模型时，第一个反应就是“官方 SDK 挺好用”。确实，只接一家模型厂商，根本不需要什么统一层。但现实是，项目刚起步时用 A 厂商的模型，上线前要做效果对比，于是要同时接 B、C 厂商；后来为了控制成本，又把轻量任务切到本地量化模型；再后来团队里有人想试试开源模型微调后的效果。每一次新增模型提供商，都不是简单换一个参数，而是消息格式、token 计算方式、流式输出字段、工具调用协议全都不一样。
比方说，OpenAI 的工具调用用的是tools数组加tool_choice，而某家的函数调用可能叫functions，参数名称和返回结构完全不同。如果业务代码里到处直接调用厂商 SDK，排查问题时要在多个 SDK 的文档里来回跳，维护成本会指数级增长。我甚至见过一个项目里同时出现三套消息会话管理逻辑，只因为用了三家不同模型的 SDK。
更麻烦的是 Agent 场景。Agent 的每次推理可能要连续调用多轮工具，每一轮的对话上下文都要手工拼接。一旦遇到流式输出，不同厂商返回的delta结构也不一样。这些细节如果不抽象出来，Agent 的调试会非常痛苦。所以我才在 Pi Agent Harness 里先做了一层模型网关，让上层只依赖一个chat()接口，屏蔽掉后端差异。
1.2 统一 API 层与自扩展能力的边界
有人会问：市面上已经有 LiteLLM、OpenRouter 这类统一接入方案，为什么还要再造一个？LiteLLM 确实把大量模型协议翻译成了类似 OpenAI 风格的接口，但它解决的是“调用”问题，没有碰 Agent 运行时。OpenRouter 则更偏转发网关，缺少本地模型和自定义后端的灵活度。我的目标是做一个“harness”，英文里这个词常用于马具或安全背带，意思是你给 Agent 套上一套可控的执行框架，既能约束它，又能支撑它发挥能力。
因此 Pi Agent Harness 把边界画得很清楚：模型接入层只负责通信协议统一，Agent 运行时负责规划、工具注册和任务循环，沙箱负责代码执行，三者互相独立。这样用户可以根据自己的需要替换任意一层。比如你只想把多个模型 API 统一给内部业务用，可以只取model_gateway模块；如果你想给 Agent 加上写代码的能力，但不想自己实现会话管理，可以直接用agent_runtime加默认工具。
这套设计对三类人最有价值。第一类是后端工程师，想把杂乱的模型接入收敛成稳定的内部服务；第二类是 AI 应用开发者，在做编码助手、自动化脚本工具，希望 Agent 能自主完成“写代码、跑代码、改代码”的闭环；第三类是刚接触 Agent 开发的人，需要一个边界清晰的参考实现，研究工具注册和沙箱执行到底是怎么串起来的。
2. 核心思路：让 Agent 能创造自己的工具
2.1 从工具调用到工具创造
传统的大模型 Agent 工作流，核心是“工具调用”（tool calling）。开发者预先定义好函数签名和描述，模型根据用户的问题选择其中一个函数并传参，代码里再根据函数名分发执行。这套逻辑对“查天气、算数学、搜资料”这类固定场景很管用，但放到编程任务上就不够用了。编程任务的解决方案千变万化，你不可能预先把所有需要的函数都写好，更多时候需要模型针对当前问题“现写一个函数”。
“自扩展编码工作”的核心就是：允许 Agent 在执行任务过程中，生成代码片段，通过代码执行沙箱去运行，如果运行成功且函数签名有效，就把这个函数自动注册到工具列表里，后续的迭代可以直接调用。也就是说，Agent 的能力边界不是代码发布时固定的，而是随着任务推进不断生长的。
这个设计让我想起早期用 REPL 调试代码的经历。你在命令行里定义一个函数，接下里的小实验就可以反复调用它。Pi Agent Harness 把这种交互搬到了 Agent 内部：模型每一次“发挥”的成果，都会被保存为可复用的工具，而不是执行完就丢弃。这样处理的直接好处是后续每一轮推理的上下文可以更短，因为已经注册的工具只需要通过工具描述引用，不必把完整代码再贴一遍。
2.2 自扩展循环的停止条件与代价控制
自扩展听起来很美，但它也是“失控”的温床。如果模型不断生成新函数、不断执行新代码，却没有清晰的目标和停止条件，整个循环可能无限跑下去，token 费用和计算时间都不可控。所以我在 harness 里强制加入了几层控制。
第一层是迭代次数上限，默认 8 轮，防止 Agent 陷入“改代码—运行—报错—再改代码”的死循环。第二层是任务完成判定，Agent 需要自己总结当前成果，并明确标记“任务已完成”或“仍需要继续”，运行时才允许进入下一轮。第三层是执行预算，包括总 token 量、累计执行时间、生成的最大代码行数，任何一个触发都会强制终止当前任务。
这三层控制不是拍脑袋定的，而是从实际项目里踩坑踩出来的。之前我在另外一个原型里没有做完成判定，让模型自由发挥，它为了把一个测试用例跑绿，连续迭代了 23 轮，最后还把一些无关的调试文件注册成了工具。加上显式的完成标记和预算控制后，大部分编码任务都能在 3 到 5 轮内收敛，极少出现失控情况。
3. 架构设计与关键实现
3.1 模块组成与数据流
Pi Agent Harness 的项目结构很清晰，核心模块可以划分成这几块：
| 模块 | 职责 | 关键接口 |
|---|---|---|
| model_gateway | 统一 LLM API 接入，负责协议转换、重试、流式封装 | chat(messages, tools, config) |
| agent_runtime | 管理 Agent 主循环，维护会话上下文，调用规划器和执行器 | run(task, max_iterations) |
| tool_registry | 维护可用工具列表，负责新工具的签名解析与注册 | register(function_def, callable) |
| code_sandbox | 安全执行模型生成的代码，返回标准输出、错误信息和执行时长 | run_code(code, timeout) |
| planner | 根据任务目标，输出下一步动作（调用已有工具还是创建新工具） | plan(task, context) |
模块之间的数据流并不复杂。用户向agent_runtime提交一个任务，runtime 把当前任务描述和会话历史提交给模型网关，模型返回一个决策。如果决策是调用已有工具，runtime 就去 tool_registry 找到对应函数并执行；如果决策是生成新代码，runtime 会把代码交给 code_sandbox 执行，执行成功后再把新函数的签名和入口注册回 tool_registry。整个过程反复循环，直到主动终止或完成判定成立。
我特意把 code_sandbox 和 tool_registry 做成两个独立进程，而不是塞在 Agent 进程里。这样即使模型生成的代码出现段错误、OOM 或者死循环，也不会拖垮主服务。实际部署时，沙箱甚至可以跑在单独的容器或云函数里，进一步隔离资源风险。
3.2 统一 LLM API 适配层怎么写
统一 API 适配层的目标，是让上层只需要关心一个chat()方法。下面这段代码是我从项目里抽出来的简化版本，只保留了最核心的协议转换逻辑：
# model_gateway.py from typing import Optional class ModelGateway: def __init__(self, provider: str = "openai", **kwargs): self.provider = provider self.client = self._build_client(provider, kwargs) def _build_client(self, provider, kwargs): if provider == "openai": from openai import OpenAI return OpenAI( api_key=kwargs.get("api_key"), base_url=kwargs.get("base_url"), ) elif provider == "anthropic": from anthropic import Anthropic return Anthropic(api_key=kwargs.get("api_key")) elif provider == "ollama": # 本地模型场景，走 OpenAI 兼容协议 from openai import OpenAI return OpenAI( api_key="ollama", base_url=kwargs.get("base_url", "http://localhost:11434/v1"), ) else: raise ValueError(f"unsupported provider: {provider}") def chat(self, messages, tools=None, **params): if self.provider == "openai": response = self.client.chat.completions.create( model=params.get("model"), messages=messages, tools=tools, ) return response.choices[0].message elif self.provider == "anthropic": # 转换成 Anthropic 对应的工具调用格式 response = self.client.messages.create( model=params.get("model"), messages=messages, tools=tools, ) return _anthropic_message_to_openai_style(response) elif self.provider == "ollama": response = self.client.chat.completions.create( model=params.get("model"), messages=messages, tools=tools, ) return response.choices[0].message可以看出，统一 API 的重点不是封装一个类，而是把各家模型的“消息结构”翻译成统一的中间结构。我在项目里把中间消息结构定为openai风格，因为社区对这套结构的接受度最高，大部分开源模型服务也都实现了兼容接口。这样好处很明显：上面真正的 Agent 逻辑只认一种格式，模型厂商增加新能力时，只需要扩展适配层，不需要改动核心运行时。
3.3 自扩展工具注册机制
自扩展工具注册是整个 harness 最有意思的部分。模型生成的往往是一段包含函数定义的代码，我需要提取这个函数的纯代码、函数名、参数信息，然后把它包装成可调用的工具。简化后的注册逻辑大概是这个样子：
# self_extending_tool.py import ast import inspect import textwrap class SelfExtendingTool: def __init__(self, code: str): self.code = code self.function_defs = [] def parse(self): tree = ast.parse(self.code) for node in tree.body: if isinstance(node, ast.FunctionDef): self.function_defs.append(node.name) # 这里可以把 AST 里的参数、类型注解提取出来， # 生成 OpenAI 风格的 tools 结构 self._extract_signature(node) def register(self, registry): namespace = {} exec(compile(self.code, "<generated>", "exec"), namespace) for func_name in self.function_defs: func = namespace[func_name] registry.register_func(func_name, func)实际项目里，我不会让模型生成的代码任意执行，而是会把代码先扔进沙箱的临时命名空间，运行一次“空调用”确认函数可以被定义，再提取签名。这样至少能避免语法错误和部分运行时错误污染工具列表。
注册之后，工具并不能自动变成“能用的工具”。Agent 下一次要调用它，必须知道它有什么参数、返回什么类型。因此_extract_signature里我会解析函数的参数列表和 docstring，生成标准工具描述。注意，这里我反复强调 docstring，因为很多模型生成函数时不写注释，后面 Agent 检索工具时根本不知道这个函数是干什么的。所以在用户提示词里，我会引导模型在生成代码时写上清晰的功能描述。
4. 从零跑通一个自扩展编码任务
4.1 环境准备和安装
如果你也想把 Pi Agent Harness 跑起来，环境准备其实很简单。我建议使用 Python 3.10 以上版本，然后用uv做依赖管理，速度比 pip 快很多。
git clone https://github.com/yourname/pi-agent-harness.git cd pi-agent-harness uv venv source .venv/bin/activate uv pip install -e .安装完成后，先跑一下内置的完整性检查：
agent-harness --check这个命令会检查模型网关配置、沙箱环境、工具注册表是否都能正常工作。我强烈建议在首次运行前把沙箱跑通，因为后面很多问题都是沙箱环境没配对引起的。检查通过之后，再用agent-harness run --task "..."启动任务。
要特别说明的是，如果你想用本地模型，最好先准备好一个兼容 OpenAI 协议的本地推理服务，比如 Ollama 或其他工具，只需要保证访问地址能通即可。否则后面会出现连接失败，但你不一定马上意识到是本地模型服务没起来。
4.2 配置模型服务
模型网关支持通过环境变量或配置文件指定提供商。我习惯用.env文件，让每台机器上的密钥和地址不混进代码仓库。
# .env PAH_MODEL_PROVIDER=openai PAH_MODEL_NAME=gpt-4o-mini PAH_API_KEY=sk-... PAH_BASE_URL=https://api.openai.com/v1 # 如果要切本地模型 # PAH_MODEL_PROVIDER=ollama # PAH_MODEL_NAME=qwen2.5:7b # PAH_BASE_URL=http://localhost:11434/v1配置文件的解析逻辑很简单，所有PAH_前缀的变量会被读入配置对象，并传给模型网关。这样做的好处是，切换模型时不需要修改任何 Agent 代码，改环境变量重启服务就行。我经常在同一台机器上维护多份.env文件，比如.env.prod和.env.local，通过软链切换。
关于模型型号的选择，我会建议至少有 7B 或以上的参数规模，并且要支持函数调用，否则自扩展效果会打折扣。小参数量模型也能跑通基础对话，但在“生成代码并注册工具”的多步推理上，很容易漏掉参数或忘记 docstring。
4.3 执行任务与观察日志
配置完成后，就可以试一个最简单的自扩展编码任务。我会用一个经典的“写函数并测试”的例子作为入门任务：
agent-harness run --task "编写一个计算斐波那契数列的函数 fib(n)，然后调用它计算第 10 项，并把结果写回 final answer。完成以后把这个函数注册为工具，之后我们还会用到它。"你可以在日志里看到 Agent 的执行轨迹。它通常会先输出一个简要计划，然后生成代码，沙箱执行成功，再把函数注册到工具注册表，接着调用这个新函数，最后返回结果。整个过程看起来是：
- 模型生成
fib函数代码。 - 沙箱执行代码，确认语法有效。
- 工具注册表提取参数
n和 docstring，生成工具描述。 - Agent 下一轮选择调用这个新注册的
fib工具。 - 工具返回结果，Agent 总结并标记任务完成。
如果任务执行失败，日志里一般会显示模型“生成代码”和“注册工具”的中间内容。我建议开启 verbose 模式：
agent-harness run --task "..." --verboseverbose 模式会打印每次迭代的完整消息、工具调用参数、沙箱标准输出和错误信息。很多看似玄学的问题，打开 verbose 之后立刻就能定位。
5. 常见问题与排查实录
5.1 模型接入不兼容
最常见的坑，是模型服务虽然宣称“兼容 OpenAI 协议”，但实际返回的消息格式略有差异。比如部分本地服务不会返回tool_calls，或者把工具调用放在delta的某层结构里，导致统一适配层解析失败。
我排查这类问题的一般步骤是：先用 curl 或调试工具直接调用一次模型接口，看原始返回结构；再对比 Pi Agent Harness 适配层期望的结构；最后在适配层里加一个raw_response日志字段。如果只是标准 OpenAI 协议的微调，很多情况下是字段名大小写不一致，或者多了一层嵌套。这类问题不要在业务代码里打补丁，而是回到适配层去映射，否则越打越乱。
另外要提醒一点，如果你用了企业内部的模型网关，通常会有一套自己的鉴权头。统一 API 层需要添加自定义 header 的入口，不要把这部分写死在环境变量里。我在项目里支持了extra_headers，可以在配置里直接传鉴权字段。
5.2 沙箱执行环境缺依赖
模型生成的代码经常会用到第三方库，比如requests、pandas或者某个特定版本的numpy。如果沙箱环境缺少这些依赖，代码会抛ModuleNotFoundError，Agent 可能会连续尝试重装依赖，导致任务超时。
我的做法是在沙箱里预装一组常用的 Python 库，同时把“自动安装依赖”作为默认关闭的选项。因为让模型生成的代码去任意执行pip install是一件很危险的事情，尤其是在沙箱网络没有限制的情况下。如果你确实需要自动安装，必须限定在一个独立的虚拟环境中，并设置超时。
还有一个容易忽略的点：模型生成的代码经常使用相对路径存取文件，如果沙箱工作目录和 Agent 主进程不同，文件可能不会出现在预期位置。最好在沙箱初始化时固定工作目录，并把结果文件显式复制回主进程的数据目录。我在项目里还加入了output_file约定，模型需要把重要结果写到固定文件中，避免通过标准输出传输大段数据。
5.3 Agent 反复循环没有产出
这是一个很让人头疼的故障。现象是日志里每轮都有“继续处理”的决策，但工具注册表没有新增工具，最终答案也没有生成，直到迭代上限被强制结束。
从经验上看，这种情况的原因通常是三种。第一，模型没有理解“完成”的定义，任务描述里缺少可验证的完成标准。比如你只说“写一个计算斐波那契数列的函数”，不如改成“完成函数定义并成功执行 n=10 的测试，将结果写入 final answer”。第二，工具调用返回的错误信息不够具体，模型看到Tool execution failed并不知道该修哪里，最好返回完整 traceback。第三，上下文过长导致模型开始复读旧内容，需要手动截断早期信息或总结前几轮结论。
解决思路是在每一轮迭代结束后，强制模型生成一段“简短进展总结”，包括已完成、未完成、下一步动作。这样既能给模型提供结构化中间状态，也能让调试者一眼看出它在哪个环节卡住。我在 harness 里把这几个字段固化进提示词，效果很明显。
5.4 工具注册表冲突
自扩展工具多了以后，第二个典型问题是工具名冲突。模型生成一个parse_data，沙箱里可能早就注册过一个同名函数。如果直接覆盖，之前依赖旧版本工具的上下文全部会出问题。反过来，如果拒绝注册，模型又可能反复尝试用同一个函数名，浪费很多轮次。
我的处理方式是引入“工具命名空间 + 版本号”。新工具注册时，如果发现同名工具，自动加上后缀，例如parse_data_2，并且保留旧版本。在 Agent 调用工具时，可以由模型指定版本，也可以由运行时按“最近注册优先”策略匹配。这里有一点值得注意：工具注册表的描述必须足够详细，不然模型在选择工具时大概率凭名字乱猜，选错工具的连锁反应非常难排查。
另外，每当一轮任务结束，我会把临时注册的工具从全局工具表中清除，避免多个任务之间互相污染。只有标记了persistent=true的工具才会跨任务保留。
6. 开源实践中的经验与扩展方向
6.1 开源维护的几点建议
把 Pi Agent Harness 开源之后，我收获最大的不是 star 数，而是被迫把项目里“只可意会不可言传”的细节写成文档。这个项目最开始 README 写得非常简陋，第一批 issue 基本都是在问同一件事：怎么配置本地模型。后来我把环境变量说明做成表格，又补了几个可运行的 example，提问量立刻下来了。
开源仓库维护有一条很实用的原则：没有示例的配置项等于没有配置项。每新增一个环境变量，就必须配套一个示例。对于 Agent 类项目尤其如此，因为模型输出本身有随机性，用户如果照着文档跑通不了，第一反应是项目有 bug，而不是自己环境的问题。
我还引入了 CI 自动化，用 GitHub Actions 在每次提交时跑一遍 smoke test，包括最基础的“文本对话”和一个“自扩展注册”用例。虽然会比较耗时间和资源，但能防止改动适配层时不小心破坏核心流程。个人项目维护易受动力影响，CI 是成本最低的质量底线。
6.2 后续可以继续做的方向
这个 harness 目前已经能稳定跑通内部的一些自动化任务，但我认为还有几个方向特别值得扩展。
第一是多 Agent 协作。现在一个任务由一个 Agent 负责，复杂任务会显得力不从心。如果拆成“规划 Agent”、“编码 Agent”、“测试 Agent”，每个 Agent 共享同一个工具注册表，就能把一个庞大的编码任务拆到不同角色手里，减少上下文碎片化。第二是记忆持久化。目前工具注册表存在于内存中，重启后新注册的工具全部丢失。下一步可以把它持久化到 SQLite 或 Redis，跨会话复用工具，这样才能真正积累 Agent 的能力。第三是更完善的沙箱网络管控。模型生成的代码一旦联网，风险就会高很多，我计划给沙箱增加更细粒度的网络白名单和资源配额，让它适合企业内部多人共用。
最后再分享一点个人体会。做这类 Agent 框架，最容易犯的错误是一开始就设计得特别大，支持一堆模型厂商、挂十几个工具。等你真正跑起来会发现，一半的扩展点根本没用到，反而是核心循环的稳定性被拖垮了。我建议你拿到 Pi Agent Harness 之后，先只接一个本地模型，跑通一个最简单的“生成代码—注册工具—调用工具”闭环，再逐步加复杂任务。跑通一次稳定闭环，比叠加一堆功能有用得多。做开源项目也是一样，把最小核心做扎实，比画一张宏伟蓝图更能赢得社区信任。
