并行工具调用与冲突处理

"查北京和上海两个城市的天气"这类问题，模型可能一次返回两条 tool_calls。默认串行执行会白白浪费一半时间。本章让工具调用并发执行：ThreadPoolExecutor 跑批量、结果按原顺序回填，并处理三个实战难点——同资源写冲突、重复调用幂等、部分失败重试。

模型一次返回多个 tool_calls

带 tools 调用时（第 8 章 run_agent 用的就是这套协议），只要服务商支持，模型可在一个回复里给出多条 tool_calls（OpenAI 兼容参数 parallel_tool_calls 默认开启）：

from llm_client import LLMClient, get_message   # 沿用第 2 章
from tools import TOOL_SCHEMAS                  # 沿用第 5 章

llm = LLMClient()
resp = llm.chat(messages, tools=TOOL_SCHEMAS,
                parallel_tool_calls=True)   # 允许并行（默认开启，可显式写出）
msg = get_message(resp)
for tc in msg.tool_calls or []:
    print(tc.id, tc.function.name, tc.function.arguments)
# 输出（示例）：
# call_bj get_weather {"city": "北京"}
# call_sh get_weather {"city": "上海"}

个别服务商不识别 parallel_tool_calls 参数时删掉即可（不传默认也允许并行）。每个 tool_call 有三个字段：id（回填时一一对应）、function.name、function.arguments（JSON 字符串）。

并发执行 + 顺序回填

执行层用 ThreadPoolExecutor 并发跑，pool.map 保证结果顺序与输入一致；单条失败自动重试，不影响其他条：

# agent_demo/parallel_tools.py —— 第 15 章
import json
import time
from concurrent.futures import ThreadPoolExecutor

def execute_batch(tool_calls, tools, retry_times: int = 3) -> list:
    """并发执行一批 tool_calls。
    tools: {函数名: 函数}；返回 [(call_id, content)]，与输入顺序一致。"""
    def run(call):
        fn = tools[call.function.name]
        args = json.loads(call.function.arguments or "{}")
        for attempt in range(retry_times):
            try:
                return call.id, fn(**args)
            except Exception as e:                 # 失败自动退避重试
                if attempt == retry_times - 1:
                    return call.id, f"执行失败（重试 {retry_times} 次）：{e}"
                time.sleep(0.3 * (attempt + 1))
    with ThreadPoolExecutor(max_workers=8) as pool:
        return list(pool.map(run, tool_calls))     # map 保持输入顺序

def fill_back(messages, results) -> list:
    """把执行结果按原顺序回填成 tool 消息（id 与工具调用一一对应）。"""
    for call_id, content in results:
        messages.append({"role": "tool", "tool_call_id": call_id,
                         "content": str(content)})
    return messages

离线试跑：并发比串行快一倍

模型响应拿不到也能验证执行层：用 SimpleNamespace 伪造 tool_calls（字段与 SDK 对象同构），配两个各睡 1.5 秒的假接口：

from types import SimpleNamespace

def slow_job(name: str) -> str:      # 模拟耗时接口，离线可跑
    time.sleep(1.5)
    return f"{name} 完成"

calls = [SimpleNamespace(id="c1", function=SimpleNamespace(
             name="slow_job", arguments='{"name": "任务A"}')),
         SimpleNamespace(id="c2", function=SimpleNamespace(
             name="slow_job", arguments='{"name": "任务B"}'))]
t0 = time.time()
results = execute_batch(calls, {"slow_job": slow_job})
print(results)
# 输出：[('c1', '任务A 完成'), ('c2', '任务B 完成')]
print(f"总耗时 {time.time() - t0:.1f}s（串行约 3s）")
# 输出：总耗时 1.6s（串行约 3s）

冲突处理一：同资源写冲突要加锁

两个工具并发写同一个文件/表会交错或互相覆盖。文件类操作在函数内部加 threading.Lock，把"读-改-写"变成原子段：

import threading

_append_lock = threading.Lock()      # 模块级：保护同一资源的所有写方

def append_note(name: str, line: str) -> str:
    """给会议纪要追加一行；加锁保证并发下不交错。"""
    with _append_lock:
        with open(f"notes/{name}", "a", encoding="utf-8") as f:
            f.write(line + "\n")
    return f"已写入 {name}：{line}"

并发调用两次后文件里两行都完整存在，不会出现"两行各写一半"的脏数据。数据库同理：sqlite 短连接 + 事务本身串行化写，第 10 章的 _exec 已天然安全。

冲突处理二：重复调用要幂等

模型偶发重复调用同一个工具。分两类处理：只读类（如查天气）加结果缓存，同参数直接返回上次结果：

_cache = {}     # tools 为外层维护的 {函数名: 函数} 字典

def cached_execute(name: str, args: dict) -> str:
    key = (name, json.dumps(args, sort_keys=True))
    if key not in _cache:
        _cache[key] = tools[name](**args)          # 只读工具：重复调用无副作用
    return _cache[key]

写类（如转账、下单）绝不能用结果缓存糊弄，要用"业务唯一键 + 数据库约束"保证只生效一次：

def transfer(biz_id: str, amount: float) -> str:
    """同 biz_id 重复提交只生效一次。"""
    conn = sqlite3.connect("ledger.db")
    try:
        conn.execute("CREATE TABLE IF NOT EXISTS tx("
                     "biz_id TEXT PRIMARY KEY, amount REAL)")
        conn.execute("INSERT INTO tx(biz_id, amount) VALUES(?, ?)",
                     (biz_id, amount))
        conn.commit()
        return f"转账 {amount} 元成功（流水号 {biz_id}）"
    except sqlite3.IntegrityError:                 # 主键冲突 = 重复请求
        return f"流水号 {biz_id} 已处理过，重复请求已忽略"
    finally:
        conn.close()

冲突处理三：部分失败自动重试

execute_batch 里已内置退避重试。注意两点：重试只对"暂时性"错误有意义（网络抖动、超时），参数写错、业务规则拒绝这类错误重试只会浪费 token；重试仍失败时把错误当普通工具结果回填，让模型自己决定换参数还是放弃：

attempts = {"n": 0}

def flaky() -> str:                  # 前两次抛错，第三次成功
    attempts["n"] += 1
    if attempts["n"] < 3:
        raise TimeoutError("模拟网络抖动")
    return "第 3 次调用成功"

print(execute_batch([SimpleNamespace(id="r1", function=SimpleNamespace(
        name="flaky", arguments="{}"))], {"flaky": flaky}))
# 输出：[('r1', '第 3 次调用成功')]

接入第 8 章 run_agent

SDK 返回的对象与上面的 SimpleNamespace 字段同构。把第 8 章 agent.py 的 run_agent 里"逐条执行 tool_calls"的 for 段替换成批量执行即可（tools_map 直接用第 5 章的 TOOL_REGISTRY——它正是"名字 → 函数"的字典）：

# 替换 run_agent 内部处理 tool_calls 的那一段
if msg.tool_calls:                                # msg = get_message(resp)
    results = execute_batch(msg.tool_calls, TOOL_REGISTRY)   # ① 并发执行
    for call_id, content in results:              # ② 按原顺序回填 tool 消息
        conv.append({"role": "tool", "tool_call_id": call_id,
                     "content": content})
    continue                                      # ③ 继续下一轮，让模型总结/再调

注意两点：回填顺序必须与 tool_calls 一致（execute_batch 用 pool.map 保证，模型靠"第几条结果对第几个 id"来对应）；run_agent 原本的"死循环计数"逻辑在并行版里对每个 call 独立计数即可，其余保险原样保留。

小结：并行工具调用 = parallel_tool_calls 让模型一次给多条 + ThreadPoolExecutor 并发执行 + 按原顺序回填 tool 消息；冲突处理三件套——写冲突加锁、重复调用靠缓存或业务唯一键幂等、暂时性失败退避重试并把最终错误当结果交还模型判断。