Skip to content

03 工具调用与执行循环 ​

第二课的程序总是先检索再生成,流程是固定的。本课把同一个 search() 包装成工具,让模型提出检索请求,应用执行后再把结果交回模型。重点是看清控制循环和执行边界,而不是增加工具数量。

源码是 examples/doc_assistant/agent.py。沿用同一份资料、同一套答案结构和引用校验。

1. 一轮工具调用有五个角色 ​

对象职责本例
用户问题提出目标如何构建文档
工具定义告诉模型工具名称、用途和参数search_docs(query)
模型响应请求调用某个工具function_call
应用执行器验证参数并执行允许的函数调用本地 search()
工具结果把执行结果交回对应的调用function_call_output

模型输出一个函数名并不会自动运行你的 Python 代码。工具真正执行的地方,是应用中的明确分支。官方的函数调用流程也区分了模型提出调用与应用执行工具这两个阶段。函数调用说明

2. 读一遍完整数据流 ​

text
问题
  → 模型请求 1
  → function_call(name=search_docs, call_id=...)
  → 检查工具名、参数字段、问题长度
  → 本地检索
  → function_call_output(call_id=同一个 ID)
  → 模型请求 2
  → 结构化答案
  → 引用校验

调用 ID 用于说明“这份结果属于哪次调用”,不能拿另一轮的 ID 代替。发送下一轮时,代码保留上轮的完整 output,而不仅仅是函数调用条目;推理模型的输出还可能包含需要继续传递的 reasoning 条目。会话状态管理

本例使用 store: false 和显式历史,不依赖 previous_response_id。不需要解析或展示加密的推理内容,只需要保持相关输出条目完整。

3. 实际实现 ​

py
"""第三课:单个只读工具与有次数上限的调用循环。"""
import json
from common import options, payload, live_request, parse_answer, show, validate_question
from retrieval import search


TOOLS = [{
    "type": "function", "name": "search_docs", "description": "搜索教学文档中的证据,不执行命令。",
    "strict": True,
    "parameters": {
        "type": "object", "properties": {"query": {"type": "string"}},
        "required": ["query"], "additionalProperties": False,
    },
}]


def run_agent(question, send=live_request, max_rounds=3):
    history = [{"role": "user", "content": validate_question(question)}]
    allowed_ids = set()
    for _ in range(max_rounds):
        request = payload(history.copy())
        request.update(tools=TOOLS, parallel_tool_calls=False)
        response = send(request)
        if response.get("status") != "completed":
            raise ValueError("响应未完成,停止工具循环")
        output = response.get("output", [])
        calls = [item for item in output if item.get("type") == "function_call"]
        if not calls:
            return parse_answer(response, allowed_ids)
        # 完整保留 output,其中可能包含后续调用需要的 reasoning 条目。
        history.extend(output)
        if len(calls) != 1:
            raise ValueError("本教程每轮只允许调用一个工具")
        call = calls[0]
        if call.get("name") != "search_docs":
            raise ValueError("工具不在允许列表中")
        arguments = json.loads(call["arguments"])
        if not isinstance(arguments, dict) or set(arguments) != {"query"}:
            raise ValueError("工具参数必须且只能包含 query")
        evidence = search(arguments["query"])
        allowed_ids.update(doc["id"] for doc in evidence)
        history.append({"type": "function_call_output", "call_id": call["call_id"],
                        "output": json.dumps(evidence, ensure_ascii=False)})
    raise RuntimeError("已达到模型请求次数上限,停止本轮任务")


def scripted_sender(request):
    """离线模拟两轮协议,不模拟语言模型的推理能力。"""
    results = [item for item in request["input"] if item.get("type") == "function_call_output"]
    if not results:
        print("离线第 1 轮:脚本生成 search_docs 调用")
        return {"status": "completed", "output": [{"type": "function_call",
            "name": "search_docs", "call_id": "demo-call-1",
            "arguments": json.dumps({"query": request["input"][0]["content"]}, ensure_ascii=False)}]}
    evidence = json.loads(results[-1]["output"])
    print("离线第 2 轮:脚本直接摘录检索结果")
    answer = ({"status": "answered", "answer": evidence[0]["text"], "citations": [evidence[0]["id"]]}
              if evidence else {"status": "insufficient_evidence", "answer": "样例文档没有答案。", "citations": []})
    return {"status": "completed", "output": [{"type": "message", "content": [
        {"type": "output_text", "text": json.dumps(answer, ensure_ascii=False)}]}]}


if __name__ == "__main__":
    args = options("演示工具调用协议;默认使用离线脚本响应")
    show({"mode": "live" if args.live else "offline-scripted",
          "result": run_agent(args.question, send=live_request if args.live else scripted_sender)})

这里有几处有意保持简单的设计:

  • 只注册 search_docs,不会根据模型给出的名字动态导入函数。
  • query 是唯一允许的参数,额外字段直接报错。
  • 工具查询复用第二课的长度与空值检查。
  • 每轮只允许一个工具调用,最多发起三次模型请求。
  • 答案引用只能来自本次工具实际返回过的文档。
  • 工具只能读取四条教学资料,没有写文件、执行 Shell 或访问任意 URL 的能力。

工具参数的 Schema 帮助模型按格式生成请求;Python 校验和允许列表负责实际执行边界。两者不能相互替代。

4. 先观察离线协议 ​

bash
python -X utf8 agent.py "如何构建文档"

先看到两条提示:脚本生成工具调用、脚本摘录检索结果。最终输出类似:

json
{
  "mode": "offline-scripted",
  "result": {
    "status": "answered",
    "answer": "构建文档运行 npm run docs:build。构建产物保存到 docs/.vitepress/dist,构建成功不代表已部署。",
    "citations": ["build"]
  }
}

这条输出是本地资料的确定性摘录,不是模型效果展示。scripted_sender() 只负责制造两轮协议消息,真正的检索、参数检查、结果回传和引用校验仍执行正式代码。

真实调用使用:

bash
python -X utf8 agent.py "如何构建文档" --live

真实模型可能改写检索词、提出额外检索或报告资料不足,不能预先承诺它总是两轮完成。三轮预算包含最终回答的那次模型请求;如果模型一直请求工具,程序最终会停止并报告达到上限。

5. 用反例理解停止条件 ​

模型提出的动作程序应该怎样处理
search_docs({"query":"构建文档"})检索并返回对应调用的结果
run_shell({"command":"..."})工具名不在允许列表,停止
search_docs({"query":"...","path":"..."})出现额外字段,停止
search_docs({"query":""})查询无效,停止
一直重复搜索达到请求次数上限后停止
未查到任何资料却引用 build引用不在已提供证据中,拒绝答案

报错让失败明确可见。本例没有自动换模型、自动补参数或重试到成功,方便你定位模型行为与程序规则之间的冲突。

6. 固定工作流和 Agent 如何选择 ​

如果任务始终是“检索 → 回答”,第二课更容易测试,通常也少一次让模型决定检索的调用。本课适用于学习需要条件判断、多步查询或工具选择的流程。

增加第二个工具之前,先回答:模型需要根据什么信息选择?两个工具的参数和返回结果有什么不同?选择错误时应用如何发现?如果这些问题还说不清,先保留固定流程。

生产系统还需要限制总时长、累计用量和工具结果大小。只有轮数上限无法阻止单次请求很慢,也无法代替身份授权;即使只读检索也必须限制用户能读哪些资料。

7. 练习与完成标准 ​

运行 python -X utf8 -m unittest -v,找到工具名、额外参数和轮数上限对应的测试。尝试把最大轮数临时设为 1,观察为什么一次工具调用之后还没有机会生成最终答案,然后恢复原值。

完成标准:能画出两轮消息的顺序,解释 call_id 的作用;知道模型不能绕过工具允许列表;能区分离线协议演示与真实模型执行;能说明固定工作流什么时候比 Agent 更合适。

下一课:评测与迭代。扩展阅读:工具设计、核心循环、工具权限与执行安全。接口资料核对日期:2026-09-30。