Agent 程序的最小结构:六个部分
一个最基础的 Agent 程序本质上是一个循环调度系统:让模型调用外部工具、获取执行结果、并基于结果继续推理,直到任务完成。 这一页以 OpenAI 兼容格式为唯一主线,把最小 Agent 逐部分拆开——每部分都给出"功能 / 为什么需要 / 对应 API 字段 / 包含内容"四问四答,最后落到一段逐行修正过的主循环伪代码与数据流图。 它是 agent-harness-loop.html(工程心智模型)的姊妹规格页:那页回答"Agent 是什么",本页回答"最小可用的代码由哪几块构成"。
概览
| 项目 | 说明 |
|---|---|
| 本页定位 | 最小 Agent 的结构规格页 · 以 OpenAI 兼容格式(Chat Completions)为唯一主线,不混述 Anthropic 格式 |
| 前置知识 | 知道"API 是发 JSON 进、收 JSON 出"即可(不了解可先读 agent-harness-loop.html 第 1 讲的 30 秒入门) |
| 读完会什么 | 能说出六部分各自的职责与对应 API 字段;能对照伪代码讲清"一轮循环里发生了什么";知道 3 个新手必踩的坑 |
| 与姊妹页分工 | agent-harness-loop.html 给工程视角六组件(含 Error Handler / Logger)+ 本仓 S1-S7 映射;本页给最小结构的规格级拆解,两套拆法在下方映射表互证 |
第一篇 · 声明与状态(第一至第三部分)
工具清单(Tools Registry):声明模型可调用的外部操作
功能:定义模型可调用的外部操作列表。为什么需要:模型本身无法访问文件系统、执行命令或调用 API,必须通过预定义的接口才能实现这些操作。对应 API 概念:请求体中的 tools 字段。关键边界:发给 API 的只是"声明"(名称 + 描述 + 参数 Schema),真正干活的执行函数留在本地——模型发出的是结构化调用请求,执行永远发生在你的代码里,因此权限与沙箱永远是客户端的责任。
- 工具名称(如 read_file)
- 工具描述(模型据此判断何时该调用)
- 参数定义(参数名、类型、是否必填、描述,用 JSON Schema 表达)
- 本地执行函数(实际干活的代码,不发给 API)
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取指定路径的文件内容",
"parameters": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "文件路径" }
},
"required": ["path"]
}
}
}
系统提示词(System Prompt):约束模型的行为边界
功能:设定模型的角色、行为规则和安全边界。为什么需要:模型的默认行为是通用对话,必须通过系统提示词将其约束为特定场景下的 Agent 行为。对应 API 概念:messages 数组中 role: "system" 的第一条消息。同一模型 + 同一工具集,换一套系统提示词就是另一个 Agent——系统提示词才是 Agent 真正的规格说明。
- 身份定义(如"你是一个 Coding Agent")
- 行为规则(如"修改文件前必须先读取")
- 安全约束(如"禁止访问项目目录外的路径")
- 输出要求(如"完成任务时直接输出总结")
对话历史(Message Store):维护多轮上下文状态
功能:维护完整的多轮对话记录。为什么需要:模型 API 是无状态的——每次请求不会自动记住之前的内容,必须把完整历史重新传入。对应 API 概念:请求体中的 messages 数组。messages 是模型唯一能"看见"的世界:循环里发生的每一步,都必须显式写回这个数组,否则模型下一轮完全不知道。
| role | 含义 | 谁产生的 |
|---|---|---|
| system | 系统指令 | 开发者预设 |
| user | 用户输入 | 用户 |
| assistant | 模型回复 | 模型 |
| tool | 工具执行结果 | 本地代码 |
- assistant 消息发起工具调用时,content 字段通常为 null,实际内容在 tool_calls 字段中(部分兼容端点会同时返回文本,所以用"通常"而非"总是")
- tool 消息必须携带 tool_call_id,用于关联"哪次调用的结果"——一次 assistant 消息可能包含多个调用,靠 id 一一配对
第二篇 · 通信与驱动(第四至第六部分)
模型调用器(LLM Client):封装与模型 API 的通信
功能:封装与模型 API 的通信。为什么需要:处理网络请求、身份认证、超时重试等底层细节,让上层代码只关心业务逻辑。对应 API 概念:client.chat.completions.create() 方法调用。
- 管理 API 地址和密钥
- 组装请求参数(model、temperature、max_tokens 等)
- 将 messages + tools 打包发送
- 处理网络超时、限流、格式异常等错误
响应解析与工具调度(Dispatcher):分类处理模型输出
功能:判断模型返回的类型,执行对应操作。为什么需要:模型的响应只有两种情况需要区分处理——路由只看 tool_calls 是否非空(即便响应同时带有文本,那段文本也只是"附言",不影响路由)。
| 响应类型 | 判断依据 | 含义 | 处理方式 |
|---|---|---|---|
| 纯文本 | tool_calls 为空 | 模型认为任务完成(或需要用户介入) | 退出循环,输出结果 |
| 工具调用 | tool_calls 非空 | 模型需要执行某个操作 | 调用本地函数,把结果写回消息列表 |
- 从 tool_calls 中提取工具名称和参数——参数是 JSON 字符串,不是对象,必须先 json.loads(手写 Agent 第一大坑)
- 根据名称找到对应的本地执行函数
- 调用函数,获取结果
- 将结果封装为 role: "tool" 的消息,附带 tool_call_id
主循环(Agent Loop):串联全局,控制何时停下来
功能:串联以上所有模块,驱动任务从开始到结束。为什么需要:Agent 任务需要多轮迭代(推理 → 执行 → 观察 → 再推理),必须用循环驱动。注意伪代码里的 5 处修正标注:它们是这段"看似简单"的代码从能跑变成能用的距离。
import json
def run_agent(user_task):
# 初始化:系统提示词 + 用户任务入列
messages = [system_prompt, {"role": "user", "content": user_task}]
for turn in range(MAX_TURNS): # 修正③ 安全上限,防无限循环
# 此处 response 指代 message 对象;真实 SDK 完整路径是
# response.choices[0].message(修正④ 声明抽象层级,照抄不踩空)
response = llm_client(messages, tools)
# 修正② 用 not 同时覆盖 None 与空列表 [](部分兼容端点返回 [])
if not response.tool_calls:
return response.content # 纯文本 → 任务完成,退出循环
messages.append(response) # assistant 消息先入历史(含 tool_calls)
for call in response.tool_calls: # 多个调用逐个回填 = 并行工具调用
try:
# 修正① arguments 是 JSON 字符串,必须先解析再传参
result = execute(call.name, json.loads(call.arguments))
except Exception as e:
result = f"Error: {e}" # 修正⑤ 错误也作为观察回传,让模型自行调整
messages.append({
"role": "tool", # 结果以 tool 消息回填
"tool_call_id": call.id, # 关联"哪次调用的结果"
"content": result,
})
# 回到循环顶部,带着新历史发起下一轮请求
return "达到最大轮数,任务未完成" # 异常终止路径(不只是"纯文本"一条出路)
- 最大轮数限制(防止无限循环)
- 单次工具执行超时
- 异常捕获:工具报错时将错误信息回传给模型,让它自行调整——错误也是观察(observation)的一部分,这正是 ReAct 范式的核心:模型把失败结果当作下一轮推理的输入
- (通往生产级的钩子)危险操作前暂停等待人工确认——权限门控是生产级 Coding Agent 与玩具 demo 的分水岭
六部分的数据流
messages = [system, user]"] INIT --> LOOP{"⑥ for turn in
range(MAX_TURNS)"} LOOP -->|"每轮"| CALL["④ 模型调用器
发送 messages + tools"] CALL --> LLM["🟧 LLM API"] LLM --> PARSE{"⑤ 解析响应
tool_calls 为空?"} PARSE -->|"是(纯文本)"| OUT["✅ 退出循环
输出 content"] PARSE -->|"否(工具调用)"| EXEC["⑤ 工具调度
json.loads 解析参数"] EXEC --> TOOL["🟩 本地执行函数
(错误也作为结果回传)"] TOOL --> APPEND["③ 对话历史追加
role:tool + tool_call_id"] APPEND --> CALL LOOP -.->|"轮数耗尽"| STOP["⛔ 返回:达到最大轮数"]
| 部分 | 一句话职责 |
|---|---|
| ① 工具清单 | 声明"能做什么" |
| ② 系统提示词 | 约束"该怎么做" |
| ③ 对话历史 | 记住"做过什么" |
| ④ 模型调用器 | 负责"怎么通信" |
| ⑤ 响应解析与调度 | 决定"下一步做什么" |
| ⑥ 主循环 | 控制"何时停下来" |
两套拆法互证:六部分 ↔ harness 六组件
同一件事的两种切法:规格视角 vs 工程视角
本页的"六部分"(最小结构规格)与姊妹页 agent-harness-loop.html 的"harness 六组件"(工程视角)不是两套互相竞争的理论,而是同一循环的两种切法。对照之后能看到两件事:大部分组件一一对应;工程视角多出的 Error Handler 与 Logger,正是"最小结构"与"生产可用"之间的距离。
| 本页六部分 | harness 六组件 | 关系说明 |
|---|---|---|
| ① 工具清单 | ② Tool Registry | 同一件事:声明 + 执行函数 |
| ② 系统提示词 | (并入 ① State Manager) | 最小结构独立成部分;工程视角视为 messages[0],不单独拆 |
| ③ 对话历史 | ① State Manager | 同一件事:messages 数组的维护 |
| ④ 模型调用器 | (并入 ⑤ Error Handler 的重试层) | 工程视角把"通信"与"通信出错怎么办"拆开讲 |
| ⑤ 响应解析与调度 | ③ Output Parser | 最小结构把"解析"与"路由执行"合并为一个部分 |
| ⑥ 主循环 | ④ Loop Controller | 同一件事:循环 + 三层守门(轮数 / 墙钟 / token) |
| ——(最小结构未含) | ⑤ Error Handler / ⑥ Logger | 生产必备件:错误恢复策略与轨迹记录(见下方"关于扩展") |
常见坑与关于扩展
- arguments 是 JSON 字符串不是对象——直接 execute(name, call.arguments) 会把整个字符串当位置参数传进去,必须 json.loads
- tool 消息漏带 tool_call_id 或配错对——API 直接报 400,且一次多个调用时必须逐个对应
- 判定只写 is None——部分 OpenAI 兼容端点无工具调用时返回空列表 [] 而非 None,用 not response.tool_calls 两种都覆盖
照着伪代码写的第一个坑:tool_call 是三层结构
不少教程把工具调用写成 call.name / call.arguments,照抄会直接报错。 OpenAI 的 tool_call 是三层:tool_call → function → name / arguments。
{
"id": "call_abc123",
"type": "function",
"function": {
"name": "read_file",
"arguments": "{\"path\": \"src/main.py\"}"
}
}
| 易错点 | 常见写法 | 正确写法 |
|---|---|---|
| 取工具名 | call.name | call.function.name |
| 取参数 | call.arguments 直接传 | json.loads(call.function.arguments) |
| 判空 | if ... is None | if not msg.tool_calls(None 与 [] 都覆盖) |
| 写回结果 | content=result | content=str(result)(必须是字符串),且要截断 |
补齐两块:从"能跑的玩具"到"敢用的 Agent"
六部分是最小结构,但生产环境还必须补两块:上下文预算(超窗是崩,不是变慢)与 权限网关(提示词是软约束,网关才是硬门禁)。补完后是 8 个部件。
| 部件 | 一句话职责 | 漏掉会怎样 |
|---|---|---|
| System Prompt | 约束"该怎么做" | 退化成通用聊天,不会主动用工具 |
| Tool Registry | 声明"能做什么" | 只能说说而已,动不了任何东西 |
| Message Store | 记住"做过什么" | 每轮都是失忆状态 |
| Context Budget(补) | 守住"还能装多少" | 长任务跑到中途崩溃,且报错在 API 侧极难排查 |
| LLM Client | 负责"怎么通信" | 一次网络抖动就整个任务失败 |
| Output Parser | 决定"下一步做什么" | 分不清"在思考"和"要调工具" |
| Permission Gate(补) | 决定"允许不允许" | 提示词里写着禁止删库,代码里照删不误 |
| Guard and Logger | 兜底"出错了怎么办" + 记录"怎么做的" | 出错即死;事后无法回答"为什么这次没成功" |
def run_agent(user_task):
messages = [SYSTEM_PROMPT, {"role": "user", "content": user_task}]
for turn in range(MAX_TURNS): # 守门终止
# 补 1 · 上下文预算:超窗前压缩,超窗即崩(不是变慢)
messages = compact(messages, TOKEN_BUDGET)
msg = llm_client(messages, TOOLS).choices[0].message
if not msg.tool_calls: # 用 not,不要用 is None
return msg.content
messages.append(msg)
for call in msg.tool_calls:
name = call.function.name # 三层结构,见上表
args = json.loads(call.function.arguments)
# 补 2 · 权限网关:硬门禁,拒绝也要回传给模型
if not permission.check(name, args):
result = "DENIED: 该操作不在允许范围内"
else:
try:
result = execute(name, args)
except Exception as e: # 错误回填,别吞异常
result = f"ERROR: {type(e).__name__}: {e}"
messages.append({
"role": "tool",
"tool_call_id": call.id, # 必须与调用 id 对齐
"content": truncate(str(result)), # 结果转字符串 + 截断
})
logger.write({"turn": turn, "messages": messages[-3:]})
return "达到最大轮数,任务未完成"
- 先把上下文预算与权限网关的代码全部删掉,跑一个 30 轮长任务,观察它在哪里失败、报错信息是什么
- 再逐条加回,每加一条跑一次,记录"加了这条之后多撑了几轮"
- 把 permission.check 改成恒返回 True,用提示词注入让它访问项目外路径,观察是否拦得住
关于扩展:最小结构是底座,不是天花板
这六个部分描述的是一个典型的最小完整结构。实际生产中的 Agent 会在此基础上叠加更多能力——上下文压缩(长任务 messages 必然超窗,几乎是生产必备件而非可选件)、多工具并行调用(主循环伪代码中的 for 循环已天然支持,不是新增机制)、MCP 协议(本质是把①工具清单外置为网络协议,架构不变)、多智能体协作(会出现循环嵌套与多份独立状态,最小结构不变但层级加深)、安全护栏(权限门控、沙箱)。
大部分扩展可以在此框架内理解为某一层的增强,而非架构的根本改变——这是"最小结构"作为教学底座的价值:先认清不变量,再看变量。