Function Call 与 Agent：从工具协议到错误恢复

0. 元信息

• 主题路径：docs/topics/llm-applications/subtopics/function-call-and-agent/
• 父主题：llm-applications
• 适合对象：已能调用模型 API、会写 Python 服务的学习者
• 建议周期：1～2 周
• 前置知识：无额外前置（继承父主题依赖）
• 最终目标：能实现受限 Agent loop，完成工具校验、状态转移、超时、重试、降级和审计

1. 学习路线

Function schema
  → 工具注册与参数校验
  → Agent loop 与状态机
  → 超时、重试、幂等与预算
  → 错误恢复、降级与可观测性

2. 阶段周数分配（1～2 周，每天 1.5～2 小时）

分阶段概述：

• 第 1 周（1 周方案）：完成 Function Call + Agent loop + 错误处理最小闭环；用 claude-haiku-4-5 跑通天气 / 订单两个工具的 demo。
• 第 1～2 周（2 周方案）：补安全审计 + 真实业务工具（订单状态、库存查询），并加上 trace 日志与脱敏；末尾做 1 次失败注入压测。

1 周方案每天 2 小时；2 周方案每天 1.5 小时，多留一天做安全审计与压测。Function Call 与 Agent 是 LLM 应用里最容易”看起来能用、实际上无限循环”的环节，先定状态机再写 loop。

3. 阶段表

4. 第一周（每天 1.5～2 小时）

环境约定：本子主题统一使用 Python 3.11+ 与 python -m venv .venv；anthropic SDK 通过 pip install -r requirements.txt 安装；API key 通过 export ANTHROPIC_API_KEY=... 共享 .env 模板：父主题 attachments/.env.example 含 Anthropic / OpenAI / Voyage / Qdrant / Ollama / OTel / Langfuse 全套变量；本地 cp ../../attachments/.env.example .env 后填值，.env 加入 .gitignore，加载用 python-dotenv。禁止把真实 key 提交到仓库或写入 trace / 日志。 注入；所有工具函数写在 tools/ 目录并以白名单方式注册到 registry.py。

第一周复盘要求

每日把工具 schema、Agent 状态图、超时 / 重试策略、失败注入结果写到 notes/agent/day{1..7}.md；Week 1 结束前用 30 分钟复盘：哪些工具最容易触发循环？哪些 schema 字段常被模型填错？

5. 阶段通用验收

1. 不看答案独立重写 ToolRegistry 与 Agent loop；
2. 用自己的话解释工具 schema、参数校验、状态转移与幂等的关系；
3. 画出 Agent 状态机图（含 Plan / Call / Validate / Observe / Recover / Done / Failed 7 个节点）；
4. 测试缺字段、类型错误、超时、网络错误、重复调用、敏感工具未确认 6 类失败路径；
5. 准备至少 3 组自定义工具并贴出实际 trace；
6. 记录 step 数、token 用量、cost 与延迟；
7. 能修改已有 Agent（加工具 / 改终止条件 / 加重试）并复现失败注入测试。

交付存放：第 3 项的状态机图、第 5 项的 trace、第 6 项的指标统一存到 notes/agent/ 或 README 对应章节，便于复盘与综合项目引用。

6. 最终验收（学完 1～2 周后）

• 完成 ≥ 6 个工具（天气 / 订单 / 库存 / 计算 / 退款 / 删除），每个工具附 JSON Schema、白名单配置与至少 3 条回归样例；
• 实现受限 Agent loop：最大步数 ≤ 6、重复动作检测、总 token 预算、单次 cost 上限全部可配；
• 完成 ≥ 30 条 trace 的失败注入压测，覆盖缺参数、类型错、超时、越权、重复动作、敏感工具 6 类；
• 建立审计日志：每个 tool_call 记录 request_id / tool_name / step / latency_ms / tokens / decision（allow / reject / retry）；
• 完成 1 个综合项目（见 §7），含 README、失败恢复与成本记录；
• 能用 15 分钟讲清楚 schema、状态机、超时重试、幂等、权限与审计之间的依赖。

API key 配置与回退方案

• 首选：Anthropic API（claude-haiku-4-5），配置方式：export ANTHROPIC_API_KEY=...；工具函数可全部本地 mock（先跑通协议再接真实业务）。
• 回退 1（无 API key）：本地 ollama run qwen2.5:7b 起 OpenAI-compatible 端点（默认 http://localhost:11434），把 SDK 的 base_url 切到本地；工具调用走本地模型 + 本地 mock 工具，仍可压测 Agent loop。
• 回退 2（纯离线）：去掉模型调用，把 Agent loop 改成”规则引擎 + 工具执行器”，验证状态机、超时、重试、审计逻辑；这一路径对 schema、终止条件、幂等的覆盖度足够，但不验证模型的工具选择能力。
• 记录：每次实验保存 {model, version, tools, max_steps, token_budget, cost_limit, trace_id} 到 notes/runs/<date>.json，便于复现。

7. 综合项目

首选：客服 / 订单 Agent（必做：知识库检索 + 至少 2 个业务工具 + 敏感工具人工确认 + 失败恢复 + 审计日志）。

备选：研究助手 Agent（必做：搜索工具 + 计算工具 + 引用与拒答 + 失败重试）。

任何综合项目都必须包含：

1. 需求说明：要解决的问题、用户故事、输入输出约定（含越权 / 超时 / 敏感操作分支）；
2. 数据流与工具选择理由：为什么选这套工具；为什么用这些 schema；记录权衡；
3. Prompt 与 Agent 算法说明：关键 prompt 模板、状态机、终止条件、预算策略；
4. 模块化源码：tools/ / registry.py / agent.py / audit.py 等职责单一；
5. 边界与安全测试：超时、网络错误、敏感工具未确认、重复动作、token 耗尽；
6. 运行说明：make run 或清晰 python -m ... 命令，注明环境变量与依赖；
7. README：项目介绍、运行步骤、目录结构、复盘（踩过的坑、可改进点）；
8. 评测与复盘记录：notes/retrospective.md（用时、难点、收获、下一步）。

notes/ 与 README 存放规范

所有”画图（Agent 状态机 / 失败恢复路径）“和”贴 trace”类交付物统一存放在项目根目录的 notes/ 子目录或 README 的对应章节；提交时一并带上，避免散落在聊天或临时文件里。综合项目的 notes/ 至少包含：

• notes/design.md：状态机图、工具 schema、终止条件与预算策略；
• notes/test.md：每组测试数据的输入、期望分支、实际 trace；
• notes/retrospective.md：复盘（用时、难点、收获、下一步）。

本主题贡献（Loop 6-D · llm-applications / function-call-and-agent）

本主题把”JSON schema 契约、Anthropic SDK 的 tool_use 流、Agent 状态机与终止条件”三条独立工程线拧成一条验收闭环，输出可被父主题营销页与官网首页直接复用的 agent 模板。

3 项核心职责

• tool_use 与 JSON schema 契约：用 Anthropic SDK（messages.create + tools 参数）写 tool 列表，每条 tool 用 JSON schema（input_schema + name + description）声明入参；schema 必须”够严但不 over-spec”，保证模型不会因模糊描述乱调工具。
• Agent 状态机 + 终止条件：用显式状态机（INIT → PLAN → ACT → OBSERVE → DONE / FAILED）管 agent，配预算策略（max_steps / max_tokens / timeout）；终止条件至少 3 个（DONE tool 被调 / max_steps 到达 / 错误率阈值），避免无限循环。
• 审计 + 安全：所有 tool 调用走 registry（注册中心）+ audit log（每次调用的入参 / 出参 / 耗时 / token 花费），敏感工具（写 DB / 发邮件 / 删文件）必须 human-in-the-loop 二次确认。

4 项交付物

1. tool 注册中心 + JSON schema 库：tools/ 目录每个 tool 一个文件（weather.py / search.py / db_query.py），schema 写在同文件，registry 自动加载并校验。
2. Agent 主循环源码：agent.py（基于 Anthropic SDK 的 messages 流）+ 状态机图（mermaid / ASCII）+ 终止条件 + 预算策略（max_steps / max_tokens / timeout）。
3. 审计日志 + trace：audit.py 记录每次 tool 调用的 input / output / latency / tokens，trace 按 session_id + step 聚合，能复盘到每一步。
4. 边界测试 + 必读交付：超时 / 网络错误 / 敏感工具未确认 / 重复动作 / token 耗尽 5 类各 1 例，配 notes/design.md（状态机图、schema、终止条件）+ notes/retrospective.md。

3 个验收指标

• JSON schema 校验：每个 tool 的输入严格通过 schema 校验，模糊入参被模型自动 retry 或拒绝，工具误调率 ≤ 1%。
• Agent 终止：5 类边界（超时 / 错误 / 重复 / token 耗尽 / 敏感工具）必须命中至少 1 个终止条件，无无限循环；max_steps 默认 ≤ 10。
• 审计：audit log 含 input / output / latency / tokens 四字段，按 session_id 可查完整 trace，敏感工具 100% 二次确认。

8. 推荐开源资料（按角色分工，避免堆链接）

许可证提示：复制或参考 LangGraph 等开源仓库代码前，先打开 LICENSE 确认。LangGraph 采用 MIT 许可证；OWASP / Anthropic 文档多为说明性材料，引用需保留出处。默认做法是读思路后自己重写，而不是复制粘贴；GPL 类代码用于商业 / 闭源项目前请逐条阅读。

API key 安全：示例代码统一使用 os.environ["ANTHROPIC_API_KEY"] 或等价方式；不要把 key 提交到仓库；本地用 .env（加入 .gitignore）+ python-dotenv 加载；CI 上用 secret 注入。工具调用 trace 中的敏感字段（手机号 / 卡号 / email）必须脱敏后再写日志。

默认使用顺序：先用 Anthropic Tool Use / OpenAI Function Calling 官方文档查协议 → 自己实现 ToolRegistry + Agent loop → 加超时 / 重试 / 幂等 / 预算 → 加权限白名单 + 审计日志 → 引入 LangGraph 对照状态图抽象 → 跑失败注入压测 → 复盘到 notes/retrospective.md。

9. 学习资料汇聚（v0.3 自包含）

9.1 背景与动机

Function Call 把自然语言决策转换为结构化工具请求，Agent 则把多轮决策组织成受约束的状态机。工程重点不是让模型“自主”，而是让每一步可验证、可停止、可恢复。

9.2 概念地图

stateDiagram-v2
  [*] --> Plan
  Plan --> CallTool
  CallTool --> Validate
  Validate --> Observe: 合法
  Validate --> Recover: 非法
  Observe --> Plan: 未完成
  Observe --> Done: 完成
  CallTool --> Recover: 超时/错误
  Recover --> Plan: 可恢复
  Recover --> Failed: 超预算或不可恢复

9.3 基础知识讲解

主推模型供应商官方工具调用文档和 JSON Schema 文档；备查 LangGraph 等状态图框架。先用 Python 显式循环实现，再比较框架；工具实现和日志代码使用前确认依赖许可证。

9.4 经典问题与经典案例

9.5 学习难点

• 概念难点：工具调用不是普通文本；必须区分模型意图、参数和执行结果。
• 思维难点：先定义状态、转移和终止条件，再写 loop。
• 工程难点：失败恢复必须幂等且有预算，不能用无限重试掩盖故障。

9.6 技术标准与接口

Entity

Tool schema、tool call、tool result、state、trace、deadline、retry policy、idempotency key。

Scope

Function Call 规定模型与应用的结构化交互，不授权工具本身，也不保证调用安全。

Structure

掌握 name、description、parameters、arguments、result、error、status、step count 和 token budget。

Ecosystem

供应商原生工具调用、OpenAI-compatible API、LangGraph 等框架各有消息格式和状态管理方式；以实际响应为准。

Depth Tiers

L0 知道工具调用存在；L1 看懂 schema；L2 能实现单工具调用；L3 能恢复错误并审计；L4 能设计权限和多 Agent 协作。本子主题要求 L3。

Source

官方工具调用、JSON Schema 和 LangGraph 文档；版本快照日期：2026-07-28。

9.7 关键代码

9.7.1 工具注册 + JSON Schema 校验

# 9.7.1 工具注册表 + 参数 schema 校验
from typing import Any, Callable

class ToolError(Exception):
    pass

class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, dict[str, Any]] = {}

    def register(self, name: str, description: str, parameters: dict, fn: Callable) -> None:
        if name in self._tools:
            raise ValueError(f"duplicate tool: {name}")
        self._tools[name] = {
            "name": name,
            "description": description,
            "parameters": parameters,
            "fn": fn,
        }

    def call(self, name: str, arguments: dict) -> Any:
        tool = self._tools.get(name)
        if tool is None:
            raise ToolError(f"unknown tool: {name}")
        ok, err = _validate(arguments, tool["parameters"])
        if not ok:
            raise ToolError(f"bad arguments: {err}")
        return tool["fn"](**arguments)

def _validate(args: dict, schema: dict) -> tuple[bool, str]:
    required = schema.get("required", [])
    props = schema.get("properties", {})
    for key in required:
        if key not in args:
            return False, f"missing {key}"
        v = args[key]
        t = props.get(key, {}).get("type")
        if t == "string" and not isinstance(v, str):
            return False, f"{key} not string"
        if t == "number" and not isinstance(v, (int, float)):
            return False, f"{key} not number"
    return True, ""

if __name__ == "__main__":
    reg = ToolRegistry()
    reg.register(
        "add",
        "返回两数之和",
        {"type": "object", "required": ["a", "b"],
         "properties": {"a": {"type": "number"}, "b": {"type": "number"}}},
        lambda a, b: a + b,
    )
    print(reg.call("add", {"a": 2, "b": 3}))  # 5

9.7.2 受限 Agent loop + 预算控制

# 9.7.2 显式 Agent loop：最大步数、重复动作检测、总 token 预算
from collections import deque

class AgentBudgetExceeded(Exception):
    pass

class AgentLoop:
    def __init__(self, max_steps: int = 6, max_repeat: int = 2, max_tokens: int = 2000) -> None:
        self.max_steps = max_steps
        self.max_repeat = max_repeat
        self.max_tokens = max_tokens
        self.tokens_used = 0
        self.history: deque[dict] = deque(maxlen=20)

    def step(self, plan_fn, tool_fn) -> dict:
        for i in range(self.max_steps):
            self.tokens_used += plan_fn.tokens
            if self.tokens_used > self.max_tokens:
                raise AgentBudgetExceeded("token budget exceeded")
            action = plan_fn(history=list(self.history))
            self.history.append({"role": "assistant", "action": action})
            if action.get("type") == "final":
                return action
            result = tool_fn(action)
            self.history.append({"role": "tool", "result": result})
            if self._is_repeating():
                raise AgentBudgetExceeded("repeat action detected")
        raise AgentBudgetExceeded("max steps reached")

    def _is_repeating(self) -> bool:
        recent = [h for h in self.history if h.get("role") == "assistant"]
        if len(recent) < self.max_repeat + 1:
            return False
        return all(h.get("action") == recent[-1].get("action") for h in recent[-self.max_repeat:])

9.7.3 真实可运行：Anthropic SDK + Function Call + 工具执行

# 9.7.3 用 Anthropic SDK 调一次带工具的 Function Call
# 运行：export ANTHROPIC_API_KEY=...; python fc_anthropic.py
import os
import json
import anthropic

client = anthropic.Anthropic()
TOOLS = [
    {
        "name": "get_weather",
        "description": "查询指定城市的当前天气",
        "input_schema": {
            "type": "object",
            "required": ["city"],
            "properties": {"city": {"type": "string"}},
        },
    }
]

def get_weather(city: str) -> dict:
    return {"city": city, "temp_c": 22, "condition": "sunny"}

resp = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=256,
    tools=TOOLS,
    tool_choice={"type": "auto"},
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
)

for block in resp.content:
    if block.type == "tool_use" and block.name == "get_weather":
        args = block.input
        result = get_weather(**args)
        print(json.dumps({"tool": block.name, "args": args, "result": result}, ensure_ascii=False))
        break

10. 常见误区

• 工具名用动词还是名词混用：模型选错时排查只能猜，统一成 verb_object 才便于白名单；
• 参数 schema 不写 required：模型填默认值时缺字段，业务侧才报错；
• 工具失败直接重试三次：写操作会产生副作用，必须用幂等键或拒绝重试；
• Agent 没有最大步数也没有 token 预算：跑飞一次就烧光额度，先定上限再上 loop；
• 把工具结果整段塞回 prompt：长度超限就被截断，先在工具侧做摘要或分页；
• 用”模型自己判断”代替白名单：敏感操作如退款、删库必须人工确认或独立 token；
• 工具返回错误时把 message 原样抛给模型：泄露内部栈，等于把内部信息交给 prompt 注入；
• 用”循环直到模型说完成”做终止条件：模型话痨，终止条件必须看 goal 状态或步数；
• 重试时不区分可重试与不可重试：400 / 鉴权失败重试 100 次也没用，先分类再重试；
• 多 Agent 协作时不做 trace 关联：复盘时找不到是哪一步把请求带偏的。

11. 所有知识点分类（统一规则）

1. 编程语言
2. 数据结构与算法
3. 计算机基础
4. 工程技术
5. Web 与后端
6. 前端与客户端
7. 数据与人工智能
8. 项目与职业能力

本计划归属：数据与人工智能 主 + 工程技术 辅。