本文目录14 个章节
先完成一次工具调用
brief-agent 已经有 TextModel、Goal、State 和 Action，但模型还接触不到外部资料。加入 search_local_sources 后，模型可以在缺少资料时提出调用请求。
范围刻意保持很窄：只有一个工具、一次调用、一个固定资料库。多轮循环、重试和权限留到后面。先把一次往返看清楚，工具系统才不会变成黑箱。
准备一个可重复的本地资料库
资料库用字典表示，避免网络结果变化影响练习。先把它保存在 src/brief_agent/tools.py，第 7 篇会用可检索资料库替换字典实现。
DOCUMENTS = {
"agent-loop": "Agent loops repeat observation, decision, action, and evaluation.",
"workflow": "Workflows follow code-defined paths and finite branches.",
}
def search_local_sources(query: str) -> dict:
words = set(query.lower().split())
matches = []
for source_id, text in DOCUMENTS.items():
score = len(words & set(text.lower().split()))
if score:
matches.append({"source_id": source_id, "text": text, "score": score})
matches.sort(key=lambda item: (-item["score"], item["source_id"]))
return {"matches": matches[:3]}这个函数与模型没有关系。它接收普通参数，返回可序列化 JSON。先单独运行 search_local_sources("agent loops")，确认输入输出稳定，再把它暴露给模型。
用 JSON Schema 描述工具
模型看到的是名称、描述和参数 Schema。strict 模式要求对象关闭额外字段，并列出必填属性。
TOOLS = [{
"type": "function",
"name": "search_local_sources",
"description": "Search approved local documents and return at most three matches.",
"strict": True,
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Terms to search for"}
},
"required": ["query"],
"additionalProperties": False,
},
}]工具描述要说明动作、资料范围和结果上限。模型用描述判断何时调用，应用用 Schema 检查参数形状。资源是否存在、用户是否有权限，仍要在工具函数内部判断。
一次工具调用如何往返
一次完整往返包含两次模型请求。
┌────────┐ ①问题 ┌────────┐
│ 用户 │ ───────▶ │ 模型 │
└────────┘ └───┬────┘
│ ② function_call
▼
┌──────────┐
│ 应用执行器 │
└────┬─────┘
│ ③ function_call_output
▼
┌────────┐
│ 模型 │
└───┬────┘
│ ④ 最终回答
▼
┌────────┐
│ 用户 │
└────────┘模型不会执行 Python 函数。它只返回 function_call，应用解析参数、查找白名单工具、执行函数，再用相同 call_id 回传结果。
把两次请求写进程序
import json
import os
from openai import OpenAI
client = OpenAI()
inputs: list = [{"role": "user", "content": "Find a source that explains agent loops."}]
first = client.responses.create(
model=os.environ["OPENAI_MODEL"],
tools=TOOLS,
input=inputs,
)
inputs += first.output
for item in first.output:
if item.type != "function_call":
continue
if item.name != "search_local_sources":
result = {"ok": False, "error": "unknown_tool"}
else:
arguments = json.loads(item.arguments)
result = {"ok": True, "data": search_local_sources(**arguments)}
inputs.append({
"type": "function_call_output",
"call_id": item.call_id,
"output": json.dumps(result),
})
final = client.responses.create(
model=os.environ["OPENAI_MODEL"],
tools=TOOLS,
input=inputs,
)
print(final.output_text)inputs += first.output 会保留模型产生的全部输出项。推理模型可能返回 reasoning item，丢掉它会破坏下一次请求的上下文。
读一遍运行日志
日志至少记录响应 ID、调用 ID、工具名和成功状态，不记录 API Key 或未经脱敏的私有资料。
response.created response_id=resp_01
tool.requested call_id=call_01 name=search_local_sources
tool.completed call_id=call_01 ok=true matches=1
response.completed response_id=resp_02如果模型直接回答，first.output 中不会出现 function_call，程序应直接使用 first.output_text。如果出现未知工具，应用返回结构化错误，不能用 eval 或动态导入执行模型给出的名称。
给这次往返加三条小测试
- 工具函数对相同输入返回相同排序。
- 合法
function_call能找到白名单函数并回传同一call_id。 - 未知工具返回
unknown_tool，执行器没有调用任何函数。
一次工具往返跑通后，模型已经能改变程序的下一步。接下来把这段代码收进循环，使模型拿到搜索结果后还能换关键词补查，并为循环加入三个明确出口。
一次工具调用为什么不够
技术调研很少在一次搜索后结束。模型可能先搜索主题，再读取命中的文档；发现证据不足后，还会换关键词补查。前面的单次 Function Calling 把“请求工具—回传结果”写死为一次，第三个动作没有位置可放。
下面把固定的两段请求收进 src/brief_agent/runtime.py。每一轮都执行同一组动作：调用模型、检查输出、执行工具或结束。
┌──────────────────────┐
│ 调用模型，读取输出 │
└──────────┬───────────┘
▼
┌──────────────────┐
│ 有 function_call? │
└──────┬─────┬─────┘
有 没有
│ │
▼ ▼
执行白名单工具 返回最终文本
│
▼
追加工具结果
│
└──────▶ 下一轮循环状态只保留必要数据
最小版本用 inputs 保存要送回模型的历史，再记录独立事件日志。两者用途不同：inputs 服务下一次推理，事件日志服务调试和评测。
循环还需要两个计数器：模型轮数和工具调用数。工具出错后仍可能进入下一轮，所以两个预算不能混成一个数字。
写出通用工具分发器
import json
from collections.abc import Callable
Tool = Callable[..., dict]
def dispatch_tool(name: str, arguments_json: str, tools: dict[str, Tool]) -> dict:
tool = tools.get(name)
if tool is None:
return {"ok": False, "error": "unknown_tool"}
try:
arguments = json.loads(arguments_json)
return {"ok": True, "data": tool(**arguments)}
except json.JSONDecodeError:
return {"ok": False, "error": "invalid_json"}
except TypeError:
return {"ok": False, "error": "invalid_arguments"}
except Exception:
return {"ok": False, "error": "tool_failure"}分发器只接受注册表中的函数。错误转换成短而稳定的代码，详细堆栈留在受保护日志中。模型可以看到 invalid_arguments 并修正参数，不需要接触服务器内部异常。
Agent Loop 的完整实现
import json
def run_agent(client, model: str, question: str, tools_schema: list,
tools: dict[str, Tool], max_turns: int = 6) -> dict:
inputs: list = [{"role": "user", "content": question}]
events: list[dict] = []
for turn in range(1, max_turns + 1):
response = client.responses.create(
model=model,
tools=tools_schema,
input=inputs,
)
inputs += response.output
calls = [item for item in response.output if item.type == "function_call"]
events.append({"turn": turn, "tool_calls": len(calls)})
if not calls:
return {
"status": "succeeded",
"output": response.output_text,
"events": events,
}
for call in calls:
result = dispatch_tool(call.name, call.arguments, tools)
events.append({"tool": call.name, "ok": result["ok"]})
inputs.append({
"type": "function_call_output",
"call_id": call.call_id,
"output": json.dumps(result),
})
return {"status": "budget_exhausted", "events": events}循环在调用模型前检查轮次。第六轮产生工具请求时，工具仍会执行，随后因为没有第七轮而返回预算耗尽。生产版可以在工具调用数达到上限时更早阻止执行，此处先保持控制流清楚。
三个出口必须分别测试
成功出口发生在模型返回最终文本且没有工具请求。应用保存文本和事件，状态为 succeeded。
失败出口用于无法继续的应用错误，例如任务无效、权限拒绝或状态损坏。工具的普通失败不一定结束循环，模型可以根据错误改换行动。
预算出口防止模型持续搜索或重复同一调用。达到上限时保留已有事件，状态为 budget_exhausted，不能把半成品伪装成成功答案。
running ──最终文本──▶ succeeded
│
├──不可恢复错误──▶ failed
│
└──轮数耗尽─────▶ budget_exhausted先用脚本模型调试循环
真实模型具有随机性。此处先手工构造三段响应：第一次搜索、补充搜索、最终回答；第 5 篇再把这组脚本封装为 Fake Model 和离线测试。事件顺序应为：
turn=1 → search_local_sources(query="agent loops") → ok
turn=2 → search_local_sources(query="stopping conditions") → ok
turn=3 → final → succeeded再构造一个连续六轮请求搜索的脚本，确认循环返回 budget_exhausted。如果测试进程卡住，循环仍有隐式路径没有受预算控制。
第一个 Agent 里程碑
此时系统已经满足 Agent 的最小定义：模型根据观察选择工具，工具结果进入下一轮，应用用明确条件结束运行。代码还不能说明输出是否符合业务要求，也难以在没有 API 的环境稳定测试。
第 5 篇先暂停增加能力，补上输入约束、Fake Model 和验收测试。RAG、记忆和框架迁移都将使用这条可比较的基线。
REFERENCES
参考链接
所属系列
AI AGENT 入门教程