AI Agent执行框架：工具调用、超时重试与上下文管理实战

围绕新一代模型代号 Astra 的讨论，最近在技术社区里热度很高，但绝大多数内容集中在参数规模、榜单分数、上下文长度这些普通开发者既无法验证、也无法直接改变的地方。真正决定你线上系统效果的，其实是另一个被严重低估的环节：执行框架。

同样一个模型，接在两套不同的执行框架上，任务成功率可能相差一倍以上。这不是夸张，而是工具调用循环、超时策略、上下文裁剪、错误重试这些"工程细节"叠加出来的结果。下面这篇文章不讨论任何未公开的模型规格，只聊一件可以动手验证的事：执行框架是怎么一步步决定 AI 系统性能上限的，以及怎么用 Python 搭出一套自己能跑、能测、能上线的框架骨架。

本文适合已经调用过模型接口、准备把单次问答升级成多步任务执行的开发者阅读。读完之后你应该能明白：任务失败的根因到底在模型侧还是框架侧，怎么通过 trace 定位到具体某一层，以及一套最小可用的执行框架需要具备哪些能力。

1. 为什么同一个模型，换个框架效果差一截

1.1 执行框架到底包含什么

先把概念说清楚。执行框架（Execution Harness）指的是包裹在模型之外、负责"把一次模型输出变成一串可执行动作"的全部代码。它至少包含四件事：请求怎么构造、模型返回的工具调用怎么解析、工具怎么真正执行、执行结果怎么回填给下一轮。

很多同学会把它和"提示词工程"混为一谈。其实提示词只是框架里很小的一块输入。举个具体的例子：模型返回了一个工具调用，参数里少了一个必填字段。提示词写得再漂亮也没用，真正决定这次请求是失败重试还是优雅降级的是框架里的参数校验逻辑。

再举一个例子。你的工具调用平均耗时 800 毫秒，一轮任务平均要 5 次工具调用，用户端超时设置是 10 秒。这时候能不能跑通，跟模型聪不聪明关系不大，取决于你有没有做并发、有没有做超时隔离、有没有在第四次调用失败时提前收敛。这些都是框架层面的事。

所以当我们说"这个模型的性能上限"时，准确的说法应该是"这个模型在这套框架下能达到的系统上限"。模型能力是天花板，框架决定了你实际能摸到天花板的百分之几。

1.2 模型上限与系统上限是两回事

把这件事拆成两个指标会更清楚：

"系统有效上限"这个词是我自己在项目里用的，意思是：在给定的延迟预算、成本预算和错误率要求下，这套系统能稳定完成的任务复杂度上限。

它有三个硬约束。第一是步数预算，多轮工具调用会线性放大延迟和成本，超过一定轮次之后边际收益迅速衰减。第二是上下文预算，每一轮的工具返回都要塞回上下文，很快就会被无关信息挤满。第三是错误预算，任何一次工具失败都可能导致整条链路中断，如果单次成功率是 95%，五步串联后的成功率只剩 77%。

框架的价值就在于：在不改变模型的前提下，尽量把这三个预算用足。比如把工具返回做摘要压缩，就是在省上下文预算；把可以并行的工具调用并发执行，就是在省延迟预算；把非幂等操作标记为不可重试，就是在省错误预算。

1.3 常见误判：把框架问题当成模型问题

我在排查线上问题时遇到过太多这种情况：业务方反馈"模型答非所问"，日志一拉，发现是工具返回了一个 500 错误页面的 HTML，被原样塞回了上下文，模型只能基于这堆乱码继续推理。

还有一类更隐蔽的：框架把历史消息无限追加，到第七八轮时上下文已经塞满了前面所有工具返回，模型开始"忘记"最初的任务目标，输出变得发散。这种表现看起来特别像"模型能力不足"，实际上是上下文管理没做好。

第三个高频误判是重试策略。有些团队对工具调用做了无差别重试，包括那些有副作用的写操作。结果一次网络抖动导致订单被创建了三次。这时候表现出的是"数据错乱"，但根因还是在框架。

判断方法其实很简单：把出问题的那次运行的完整 trace 拉出来，看模型每一轮收到的输入是什么。如果输入本身就是脏的、缺的、超长的，那问题一定在框架侧，换模型解决不了。

2. 执行框架的四层结构与性能损耗点

2.1 接入层：请求构造与协议适配

接入层负责把业务请求翻译成模型能理解的输入。听起来简单，但这里有几个容易踩的点。

第一是系统提示和用户输入的边界。系统提示应该放稳定不变的角色约束和输出格式要求，用户输入放具体任务。如果把两者混在一起拼接，后面做上下文裁剪时会很难处理，因为你没法安全地丢弃中间某一段。

第二是工具描述的注入方式。工具列表不是越多越好。工具数量从 5 个涨到 30 个，模型选错工具的概率会明显上升，同时每次请求的输入长度也翻了几倍。实践中更推荐按场景分组，一次只注入当前任务真正可能用到的 5 到 8 个工具。

第三是协议适配。不同供应商的接口在消息结构、工具调用格式、流式返回上都有差异。接入层的职责是把这些差异收敛成一个内部统一的中间结构，这样上层的编排逻辑就不需要关心底层是哪家。这一层做好了，后面换模型供应商的成本会低很多。

2.2 编排层：循环、终止与预算

编排层是框架的心脏，核心就是一个循环：调模型、看是不是要调工具、调工具、把结果塞回去、再调模型，直到模型给出最终答案或者触发终止条件。

这个循环必须有三类终止条件，缺一不可。

第一类是正常终止，模型返回了最终答案。

第二类是步数终止，循环次数达到上限。这个上限不能拍脑袋定，要结合业务实际。比如查询类任务通常 3 到 5 步足够，复杂的数据分析可能要到 10 步以上。上限设得太小会导致任务被打断，设得太大则会在模型"绕圈"时白白烧钱。

第三类是预算终止，包括总耗时超限、累计 token 超限、工具调用次数超限。预算终止比步数终止更贴近真实约束，因为不同步骤的耗时差异可能非常大——一次本地计算可能只要 1 毫秒，一次外部 API 调用可能要 3 秒。

除了终止条件，编排层还要处理一个容易被忽略的问题：模型返回的工具调用参数不合法时怎么办。直接抛异常会让整个任务失败，更好的做法是把校验错误作为工具结果回填给模型，让它自己纠正一次。这个"自我纠正"的机会通常能挽回相当一部分失败。

2.3 工具层：schema、副作用与超时

工具层是整个框架里风险最集中的地方，因为它直接接触外部系统。

Schema 定义要做细。不能只写"这是一个查询工具"，要写清楚参数类型、是否必填、取值范围。参数校验要放在真正执行之前，这样能挡住大部分无效调用。我在项目里见过因为没做校验，模型传入了一个 limit=-1 的参数，直接把数据库查询拖垮的案例。

副作用标记要显式。每个工具都应该声明自己是只读的还是可写的。只读工具可以放心重试和并发，可写工具必须做幂等处理或者干脆禁止自动重试。这个标记建议直接写在工具注册信息里，而不是靠命名约定。

超时必须独立设置。工具超时和模型超时是两个完全不同的概念。模型超时通常是几十秒级别，工具超时应该按业务宽容度单独设置，几秒到十几秒不等。所有工具调用都应该跑在带超时的执行器里，避免一个卡死的下游把整个任务线程占住。

2.4 观测层：没有 trace 就没有优化

这一层经常被放到最后做，然后就一直没做。但从排查效率来看，它的优先级应该排在前三。

一套够用的 trace 至少记录这几个字段：每一步的类型（模型决策还是工具执行）、耗时、输入摘要、输出摘要、是否重试。有了这些，你就能回答"这次任务为什么慢"和"这次任务为什么错"这两个问题。

更进一步，建议给每次任务一个全局 trace id，并把每步的关键指标打成结构化日志。这样后面做统计的时候，可以直接算出 P50 和 P95 延迟、平均步数、工具失败率这些指标，优化才有依据。

3. 环境准备与项目骨架

3.1 运行环境与依赖

本文的示例代码只依赖 Python 标准库，不需要安装任何第三方包，这样你可以直接复制运行，不用先折腾环境。

bash复制# 推荐 Python 3.9 及以上
python --version

# 本文示例不依赖第三方库，如需运行环境可以直接用系统自带 Python

如果你需要接入真实模型，把示例里的 MockLLM 换成对应供应商的 SDK 即可，编排逻辑完全不用改。这也是为什么前面强调接入层要做协议适配——换供应商的时候，改动应该被限制在一个文件内。

版本需要根据你的项目实际情况调整，本文示例以常见环境为例，重点演示框架的设计思路和关键机制，代码结构本身是通用的。

3.2 目录结构

text复制agent-harness/
├── demo.py                 # 演示入口
├── framework/
│   ├── __init__.py
│   ├── types.py            # 数据结构定义
│   ├── registry.py         # 工具注册表
│   ├── tools.py            # 具体工具实现
│   ├── runner.py           # 编排执行器
│   └── mock_llm.py         # 模拟模型，便于验证框架本身
└── workspace/
    └── notes.txt           # 供文件读取工具使用的示例文件

这个结构里最关键的一点是把"模型"抽象成了一个只有单一方法的对象。框架不关心它是真的模型还是假的模型，只关心它能不能根据消息列表返回一个决策。这种设计让框架本身可以被单元测试覆盖，不需要每次都真的调接口。

3.3 基础数据结构

python复制# 文件路径：framework/types.py
from dataclasses import dataclass, field


@dataclass
class Step:
    """记录执行过程中的一步，用于 trace 和问题定位。"""
    index: int
    kind: str           # "llm" / "tool" / "final"
    detail: str = ""
    elapsed_ms: float = 0.0
    retried: int = 0


@dataclass
class RunResult:
    """一次完整任务的执行结果。"""
    status: str                         # ok / max_steps / deadline / tool_budget
    answer: str = ""
    steps: list = field(default_factory=list)
    tokens_used: int = 0
    elapsed_ms: float = 0.0

    def dump(self) -> str:
        lines = [f"status={self.status} tokens={self.tokens_used} elapsed={self.elapsed_ms:.1f}ms"]
        for s in self.steps:
            lines.append(
                f"  [{s.index}] {s.kind:<5} {s.elapsed_ms:>8.1f}ms retry={s.retried} {s.detail}"
            )
        return "\n".join(lines)

Step 里有一个 retried 字段，这个字段在实际排查中很有价值。如果某次任务的工具重试次数特别高，往往意味着下游不稳定或者超时阈值设得太紧，这时候即使任务成功了也应该去看看原因。

RunResult.dump() 是专门为排查设计的，输出格式尽量紧凑，方便直接贴到工单里。很多团队的实践是：任何非 ok 状态的任务，都自动把 dump 结果落盘，后面出问题时有据可查。

4. 核心机制拆解：从工具注册到执行循环

4.1 工具注册表与参数校验

注册表要解决三个问题：工具怎么描述、参数怎么校验、异常怎么兜底。

python复制# 文件路径：framework/registry.py
class ToolRegistry:
    def __init__(self):
        self._tools = {}

    def register(self, name, description, schema):
        """装饰器式注册，schema 用于参数校验和向模型描述参数结构。"""
        def deco(fn):
            self._tools[name] = {
                "name": name,
                "description": description,
                "schema": schema,
                "fn": fn,
            }
            return fn
        return deco

    def spec(self):
        """导出给模型看的工具描述列表。"""
        return [
            {
                "name": t["name"],
                "description": t["description"],
                "parameters": t["schema"],
            }
            for t in self._tools.values()
        ]

    def call(self, name, arguments):
        tool = self._tools.get(name)
        if tool is None:
            return False, f"未知工具: {name}"

        required = tool["schema"].get("required", [])
        missing = [k for k in required if k not in arguments]
        if missing:
            return False, f"缺少必填参数: {missing}"

        try:
            return True, str(tool["fn"](**arguments))
        except TypeError as e:
            return False, f"参数与函数签名不匹配: {e}"
        except PermissionError as e:
            return False, f"权限拒绝: {e}"
        except Exception as e:
            return False, f"工具执行异常: {type(e).__name__}: {e}"

这里有一个设计取舍值得说明：call 方法不抛异常，而是统一返回 (ok, content) 二元组。原因是编排层需要把失败信息回填给模型，让模型有机会换个方式重试。如果这里直接抛异常，编排层就得写一堆 try-except，代码会很难看。

把 PermissionError 单独拎出来处理也是刻意为之。越权访问属于安全事件，应该在日志里单独标记并告警，而不是和普通业务异常混在一起。

4.2 多轮执行循环与终止条件

python复制# 文件路径：framework/runner.py（核心循环片段）
def run(self, user_input: str) -> RunResult:
    start = time.time()
    steps = []
    tokens = 0
    tool_calls = 0
    messages = [{"role": "user", "content": user_input}]

    for i in range(1, self.max_steps + 1):
        if time.time() - start > self.total_timeout:
            return RunResult("deadline", "", steps, tokens, (time.time() - start) * 1000)

        t0 = time.time()
        decision = self.llm.complete(messages)
        tokens += decision.get("tokens", 0)
        steps.append(Step(i, "llm", decision.get("type", "?"), (time.time() - t0) * 1000))

        if decision["type"] == "final":
            steps[-1].kind = "final"
            steps[-1].detail = decision["content"][:60]
            return RunResult("ok", decision["content"], steps, tokens,
                             (time.time() - start) * 1000)

        tool_calls += 1
        if tool_calls > self.max_tool_calls:
            return RunResult("tool_budget", "", steps, tokens,
                             (time.time() - start) * 1000)

        name = decision.get("name", "")
        args = decision.get("arguments", {})
        t1 = time.time()
        ok, content, retried = self._run_tool(name, args)
        steps.append(Step(i, "tool", f"{name} ok={ok} {content[:40]}",
                          (time.time() - t1) * 1000, retried))

        messages.append({"role": "assistant", "content": f"调用工具 {name}"})
        messages.append({"role": "tool", "content": content if ok else f"[失败] {content}"})
        messages = self._trim(messages)

    return RunResult("max_steps", "", steps, tokens, (time.time() - start) * 1000)

这段代码里有几个细节需要展开。

失败信息前面加了 [失败] 前缀，这是为了让模型能明确区分"工具执行成功但结果为空"和"工具执行失败"这两种情况。很多框架把失败信息直接塞回去，模型会误以为那就是工具的输出内容，然后基于错误信息继续编造答案。

_trim 在每一轮之后都被调用，而不是等上下文塞满了才处理。这是有意的，因为等到塞满再处理，前面几轮可能已经因为上下文过长导致模型表现下降了。

tool_calls 计数和 max_steps 是两个独立的上限。某些任务里模型可能一步之内发起多个工具调用，这时候步数上限挡不住工具调用次数的爆炸增长，必须单独计。

4.3 超时与重试：只重试该重试的

python复制# 文件路径：framework/runner.py（工具执行片段）
def _run_tool(self, name, args):
    attempt = 0
    last_err = ""
    while attempt <= self.max_retry:
        attempt += 1
        with ThreadPoolExecutor(max_workers=1) as pool:
            future = pool.submit(self.registry.call, name, args)
            try:
                ok, content = future.result(timeout=self.tool_timeout)
            except FutureTimeout:
                ok, content = False, f"工具 {name} 执行超时({self.tool_timeout}s)"

        if ok:
            return True, content, attempt - 1
        last_err = content
        if not self._retryable(content):
            break
        time.sleep(0.2 * attempt)

    return False, last_err, attempt - 1

@staticmethod
def _retryable(msg: str) -> bool:
    """只有超时类错误才值得重试，参数错误重试一百次也一样失败。"""
    return "超时" in msg or "timeout" in msg.lower()

这里最值得注意的是 _retryable 这个判断。很多团队的重试逻辑是"失败就重试三次"，看起来很健壮，实际上是在浪费时间和配额。参数错误、权限错误、工具不存在，这些都属于确定性失败，重试只会让延迟翻三倍，结果完全一样。

真正该重试的是瞬时性故障：网络抖动、下游超时、连接池耗尽。这类错误重试一两次往往就能成功。

另外必须说明一点：ThreadPoolExecutor 的超时只能让上层不再等待，并不能真正终止已经提交的任务。如果一个工具会写文件或者改数据库，超时之后它可能还在后台继续跑。所以对于可写工具，一定要在工具内部自己也设超时，不能只依赖外层。

退避策略用的是简单的线性退避（0.2 秒、0.4 秒）。生产环境更推荐指数退避加随机抖动，避免多个任务同时重试造成雪崩。

4.4 上下文裁剪与预算控制

python复制# 文件路径：framework/runner.py（上下文裁剪片段）
def _trim(self, messages, keep_recent=6, max_chars=1200):
    system = [m for m in messages if m["role"] == "system"]
    body = [m for m in messages if m["role"] != "system"][-keep_recent:]

    out = list(system)
    for m in body:
        content = m.get("content", "")
        if len(content) > max_chars:
            content = content[:max_chars] + "…(内容已截断)"
        out.append({**m, "content": content})
    return out

这个裁剪策略非常简单，但解决了一个很实际的问题：工具返回的内容长度是不可控的。一次文件读取可能返回几万字，一次接口调用可能返回一个巨大的 JSON。如果不做截断，两三轮之后上下文就爆了。

keep_recent 控制保留多少轮历史。这里有个权衡：保留太少，模型会丢失任务上下文；保留太多，上下文会膨胀。实践中 6 到 10 轮是个比较常用的起点。

单纯截断有个明显的副作用：如果关键信息正好在被截断的后半段，模型就看不到了。更好的做法是让工具自己负责返回摘要，比如文件读取工具只返回匹配到的行，而不是整个文件。这个原则可以概括为一句话：让每个工具返回它自己的摘要，而不是让框架去截断。

5. 完整实战：一个可运行的执行框架

5.1 编写工具实现

python复制# 文件路径：framework/tools.py
import ast
import operator
import os
import time

WORKSPACE = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "workspace"))

_BIN_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
}


def safe_calc(expr: str) -> float:
    """只允许四则运算，禁止任何函数调用和属性访问。"""
    def _eval(node):
        if isinstance(node, ast.Expression):
            return _eval(node.body)
        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        if isinstance(node, ast.BinOp) and type(node.op) in _BIN_OPS:
            return _BIN_OPS[type(node.op)](_eval(node.left), _eval(node.right))
        if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub):
            return -_eval(node.operand)
        raise ValueError(f"不支持的表达式节点: {type(node).__name__}")

    return float(_eval(ast.parse(expr, mode="eval")))


def read_file(path: str, max_chars: int = 800) -> str:
    """读取 workspace 目录下的文件，禁止路径穿越。"""
    full = os.path.abspath(os.path.join(WORKSPACE, path))
    if not full.startswith(WORKSPACE + os.sep):
        raise PermissionError(f"路径越界，拒绝访问: {path}")
    with open(full, "r", encoding="utf-8") as f:
        return f.read(max_chars)


_DOCS = {
    "限流": "执行框架应在工具层做并发限流，避免瞬时打满下游。",
    "重试": "只有幂等操作才允许自动重试，重试次数建议不超过 3 次并带退避。",
    "超时": "超时必须设置，且要区分模型超时与工具超时两个不同的概念。",
}


def search_docs(keyword: str) -> str:
    hits = [v for k, v in _DOCS.items() if keyword in k or keyword in v]
    return "\n".join(hits) if hits else "未检索到相关条目"


def slow_task(seconds: float = 3.0) -> str:
    """故意很慢的工具，用来演示超时处理。"""
    time.sleep(seconds)
    return "slow_task finished"

safe_calc 单独实现而不用 eval，这一点必须强调。eval 在接收模型生成的字符串时是极高风险的操作，一个精心构造的表达式就能读取环境变量甚至执行任意代码。用 ast 白名单遍历只允许四则运算，是这类场景的标准做法。

read_file 里的路径校验同样关键。os.path.join 遇到 ../ 是会把路径往上跳的，所以必须先 abspath 再判断前缀。这个校验看似简单，但漏掉的团队非常多。

5.2 编写 Runner

python复制# 文件路径：framework/runner.py
import time
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import TimeoutError as FutureTimeout

from .types import RunResult, Step


class AgentRunner:
    def __init__(self, llm, registry, max_steps=8, total_timeout=30.0,
                 max_tool_calls=10, tool_timeout=5.0, max_retry=1):
        self.llm = llm
        self.registry = registry
        self.max_steps = max_steps
        self.total_timeout = total_timeout
        self.max_tool_calls = max_tool_calls
        self.tool_timeout = tool_timeout
        self.max_retry = max_retry

    def run(self, user_input: str) -> RunResult:
        start = time.time()
        steps, tokens, tool_calls = [], 0, 0
        messages = [{"role": "user", "content": user_input}]

        for i in range(1, self.max_steps + 1):
            if time.time() - start > self.total_timeout:
                return RunResult("deadline", "", steps, tokens,
                                 (time.time() - start) * 1000)

            t0 = time.time()
            decision = self.llm.complete(messages)
            tokens += decision.get("tokens", 0)
            steps.append(Step(i, "llm", decision.get("type", "?"),
                              (time.time() - t0) * 1000))

            if decision["type"] == "final":
                steps[-1].kind = "final"
                steps[-1].detail = decision["content"][:60]
                return RunResult("ok", decision["content"], steps, tokens,
                                 (time.time() - start) * 1000)

            tool_calls += 1
            if tool_calls > self.max_tool_calls:
                return RunResult("tool_budget", "", steps, tokens,
                                 (time.time() - start) * 1000)

            name = decision.get("name", "")
            args = decision.get("arguments", {})
            t1 = time.time()
            ok, content, retried = self._run_tool(name, args)
            steps.append(Step(i, "tool", f"{name} ok={ok} {content[:40]}",
                              (time.time() - t1) * 1000, retried))

            messages.append({"role": "assistant", "content": f"调用工具 {name}"})
            messages.append({
                "role": "tool",
                "content": content if ok else f"[失败] {content}",
            })
            messages = self._trim(messages)

        return RunResult("max_steps", "", steps, tokens,
                         (time.time() - start) * 1000)

    def _run_tool(self, name, args):
        attempt, last_err = 0, ""
        while attempt <= self.max_retry:
            attempt += 1
            with ThreadPoolExecutor(max_workers=1) as pool:
                future = pool.submit(self.registry.call, name, args)
                try:
                    ok, content = future.result(timeout=self.tool_timeout)
                except FutureTimeout:
                    ok, content = False, f"工具 {name} 执行超时({self.tool_timeout}s)"
            if ok:
                return True, content, attempt - 1
            last_err = content
            if not self._retryable(content):
                break
            time.sleep(0.2 * attempt)
        return False, last_err, attempt - 1

    @staticmethod
    def _retryable(msg: str) -> bool:
        return "超时" in msg or "timeout" in msg.lower()

    def _trim(self, messages, keep_recent=6, max_chars=1200):
        system = [m for m in messages if m["role"] == "system"]
        body = [m for m in messages if m["role"] != "system"][-keep_recent:]
        out = list(system)
        for m in body:
            content = m.get("content", "")
            if len(content) > max_chars:
                content = content[:max_chars] + "…(内容已截断)"
            out.append({**m, "content": content})
        return out

5.3 编写演示入口并运行

python复制# 文件路径：framework/mock_llm.py
class MockLLM:
    """模拟模型：按预设脚本返回决策，用于验证框架本身是否正确。"""

    def __init__(self, script):
        self._script = list(script)
        self._cursor = 0

    def complete(self, messages):
        if self._cursor >= len(self._script):
            return {"type": "final", "content": "已完成。", "tokens": 0}
        decision = dict(self._script[self._cursor])
        self._cursor += 1
        return decision

python复制# 文件路径：demo.py
import os

from framework.mock_llm import MockLLM
from framework.registry import ToolRegistry
from framework.runner import AgentRunner
from framework.tools import read_file, safe_calc, search_docs, slow_task, WORKSPACE

os.makedirs(WORKSPACE, exist_ok=True)
with open(os.path.join(WORKSPACE, "notes.txt"), "w", encoding="utf-8") as f:
    f.write("排查记录：上周任务失败率升高的原因是工具超时阈值设得过紧。\n")

registry = ToolRegistry()

registry.register(
    name="calculator",
    description="计算四则运算表达式，仅支持 + - * / 与括号",
    schema={
        "type": "object",
        "properties": {"expr": {"type": "string"}},
        "required": ["expr"],
    },
)(safe_calc)

registry.register(
    name="read_file",
    description="读取 workspace 目录下的文本文件",
    schema={
        "type": "object",
        "properties": {"path": {"type": "string"}},
        "required": ["path"],
    },
)(read_file)

registry.register(
    name="search_docs",
    description="在内部规范文档中检索关键词",
    schema={
        "type": "object",
        "properties": {"keyword": {"type": "string"}},
        "required": ["keyword"],
    },
)(search_docs)

registry.register(
    name="slow_task",
    description="模拟耗时较长的下游调用",
    schema={
        "type": "object",
        "properties": {"seconds": {"type": "number"}},
        "required": [],
    },
)(slow_task)

script = [
    {"type": "tool_call", "name": "calculator",
     "arguments": {"expr": "12*7+3"}, "tokens": 58},
    {"type": "tool_call", "name": "search_docs",
     "arguments": {"keyword": "重试"}, "tokens": 46},
    {"type": "final",
     "content": "12*7+3 = 87。另外，只有幂等操作才允许自动重试，"
                "重试次数建议不超过 3 次并带退避。",
     "tokens": 72},
]

result = AgentRunner(MockLLM(script), registry).run(
    "帮我算一下 12*7+3，并说明重试要注意什么"
)
print(result.dump())
print("最终答案:", result.answer)

bash复制python demo.py

5.4 输出解读

正常运行的输出大致如下：

text复制status=ok tokens=176 elapsed=xxx.xms
  [1] llm       0.0ms retry=0 tool_call
  [1] tool     xx.xms retry=0 calculator ok=True 87.0
  [2] llm       0.0ms retry=0 tool_call
  [2] tool     xx.xms retry=0 search_docs ok=True 只有幂等操作才允许自动重试…
  [3] final    0.0ms retry=0 12*7+3 = 87。另外，只有幂等操作才允许…
最终答案: 12*7+3 = 87。另外，只有幂等操作才允许自动重试，重试次数建议不超过 3 次并带退避。

因为你用的是 MockLLM，模型侧的耗时是 0，这正好方便观察框架本身的损耗。工具耗时那一列才是真正需要关注的部分。

接下来把脚本换成故意出错的版本，验证兜底逻辑：

python复制# 文件路径：demo_timeout.py
from framework.mock_llm import MockLLM
from framework.runner import AgentRunner
from demo import registry

bad_script = [
    {"type": "tool_call", "name": "slow_task",
     "arguments": {"seconds": 8}, "tokens": 40},
    {"type": "final", "content": "下游较慢，已降级返回缓存结果。", "tokens": 30},
]

result = AgentRunner(MockLLM(bad_script), registry,
                     tool_timeout=1.0, max_retry=2).run("调用一下慢接口")
print(result.dump())

预期输出里会看到工具耗时接近 1 秒、retry=2，说明框架正确识别了超时并且只对超时类错误做了重试。如果这时候把 slow_task 换成一个参数错误类的失败，你会看到 retry=0，因为 _retryable 判断不通过，直接短路返回。这两个对比就是重试策略是否正确的直接证据。

再试一个更贴近真实故障的例子：让模型调用一个不存在的工具。

python复制bad_script = [
    {"type": "tool_call", "name": "query_database",
     "arguments": {"sql": "select 1"}, "tokens": 35},
    {"type": "final", "content": "工具不可用，改为基于已有信息回答。", "tokens": 28},
]

框架会返回 未知工具: query_database，这条信息带着 [失败] 前缀回填给模型，模型在下一轮就有机会改用其他工具或者直接给出降级答案。整个任务不会抛异常，也不会卡死。这就是编排层兜底能力的实际价值。

6. 故障注入与排查清单

6.1 三类高频故障

第一类是工具不可用。表现是任务在某个固定节点反复失败。排查时先看 trace 里的 ok=False 出现在哪一步，再看失败信息是"未知工具"还是"执行异常"。前者说明模型对工具的认知和实际注册不匹配，通常是工具描述写得含糊；后者要看下游服务的实际状态。

第二类是上下文膨胀。表现是任务前几步正常，到后面开始跑题或者重复。根因往往是某个工具返回了超长内容。排查方法是看每一步的输入长度变化，突然暴涨的那一步就是元凶。

第三类是循环打转。表现是模型反复调用同一个工具，参数几乎一样。这种情况通常是工具返回的信息没有让模型得到它想要的答案，比如返回了"未找到"但没有告诉它该换个关键词。解决思路是在工具返回里给出引导性提示，或者干脆在框架层加一个"相同工具相同参数连续出现两次就强制终止"的保护。

6.2 排查顺序

遇到任务失败，建议按这个顺序走：

1. 先看 status 字段，确认是正常结束还是被终止条件打断。
2. 再看 steps 里最后一步的类型，是 llm 还是 tool。
3. 如果卡在 tool，看工具返回内容和重试次数。
4. 如果卡在 llm，看这一步模型实际收到的上下文长度。
5. 最后看整体耗时分布，找出最慢的那一步。

这个顺序的原则是"先定位层，再定位点"。不要一上来就怀疑模型，大部分问题都在工具层和上下文层。

6.3 常见问题对照表

这张表里的最后一行值得单独说。并发执行的收益在工具调用次数多的时候非常明显。三个各耗时 1 秒的独立查询，串行是 3 秒，并发是 1 秒出头。但并发的前提是这几个工具之间确实没有依赖关系，而且下游能扛住翻倍的瞬时压力。没有限流的并发，本质上是在把风险转嫁给下游。

7. 工程化建议与性能优化

7.1 并发与限流

并发和限流必须成对出现。只做并发不做限流，早晚会出问题。

python复制# 文件路径：framework/limiter.py
import threading


class SemaphoreLimiter:
    """给工具调用加并发上限，避免瞬时打满下游。"""

    def __init__(self, limit: int = 8, timeout: float = 2.0):
        self._sem = threading.Semaphore(limit)
        self._timeout = timeout

    def __enter__(self):
        acquired = self._sem.acquire(timeout=self._timeout)
        if not acquired:
            raise TimeoutError("获取并发令牌超时，请稍后重试")
        return self

    def __exit__(self, exc_type, exc, tb):
        self._sem.release()
        return False

把这个限流器套在 registry.call 外面即可。timeout 参数的意义是：拿不到令牌时不要无限等待，直接快速失败，让编排层有机会降级。这一点比限流本身更重要——很多系统不是被下游拖垮的，而是被无限等待的线程堆满的。

限流值怎么定？一个实用的起点是下游服务能承受并发数的 60% 左右，留出余量给其他调用方。然后压测验证，逐步调整。

7.2 安全边界

执行框架天然是一个"把模型输出变成真实操作"的组件，安全边界必须提前设计好。

工具白名单是底线。不要在运行时根据模型给的字符串动态导入模块或执行命令。所有可调用的工具都必须在注册表里显式声明。

文件操作要限定根目录。前面 read_file 里的路径前缀校验是最低要求，更严谨的做法是把工作目录本身也设为只读挂载。

数据库操作要区分读写。查询工具可以放开，写操作工具必须走独立的授权链路，最好加上人工确认环节。删除和更新语句必须强制带 WHERE 条件，这一点应该在工具实现里做硬校验，而不是靠提示词约束。

敏感信息不能进上下文。工具返回内容里如果有密钥、手机号、身份证号，要在返回之前做脱敏。因为上下文会被完整记录到日志里，一旦泄漏就是大范围问题。

7.3 可观测性与灰度

日志至少要包含 trace id、步骤序号、工具名、耗时、成功与否、重试次数。这几项组合起来，基本能还原出一次任务的完整路径。

指标方面建议至少埋四个：任务成功率、平均步数、P95 延迟、工具失败率。这四个指标能覆盖绝大多数异常。任务成功率下降通常意味着工具侧有变动；平均步数上升意味着模型决策质量变差或者上下文管理出了问题；P95 延迟飙升往往是某个下游变慢；工具失败率是最直接的报警源。

上线节奏上，建议新工具先只读灰度，跑一段时间确认稳定后再开放写操作。框架层的改动同样要走灰度，因为一个细微的重试策略调整就可能让下游压力翻倍。

7.4 成本与延迟的取舍

框架层能做的优化，本质上都是在成本和延迟之间找平衡点。

减少步数是降本最直接的手段。可以给模型加一个"如果你已经有足够信息回答，请直接给出最终答案"的约束，并把它放在系统提示的靠后位置，这样模型在最后一轮更容易注意到。

上下文压缩是第二个抓手。工具返回只保留关键字段，历史轮次只保留摘要。有团队做过统计，把工具返回从全文改成结构化摘要之后，token 消耗能降一半左右，而任务成功率基本不变。

第三个是缓存。对于参数相同、结果稳定的查询类工具，可以在框架层做短期缓存。但要注意，缓存只适用于只读工具，写操作绝对不能缓存。

8. 学习路线与落地建议

8.1 三个阶段的演进路径

第一阶段，先把单轮工具调用跑通。也就是模型返回一个工具调用，框架执行，然后把结果给模型生成最终答案。这个阶段的目标是验证工具注册、参数校验、异常兜底这三件事。

第二阶段，加上多轮循环和终止条件。这一步的关键是别急着优化效果，先让框架在各种异常下都不崩。故障注入测试在这个阶段价值最大，前面演示的未知工具、超时、参数缺失，都应该有对应的测试用例。

第三阶段，做观测和调优。有了 trace 之后，你会发现很多原本以为是模型能力问题的现象，其实是框架细节没做好。这时候的优化才是有依据的。

这三个阶段不建议跳过。我见过不少项目直接从第三阶段开始，先搭一套复杂的编排引擎，结果基础的参数校验都没做，一出问题就完全没法定位。

8.2 上线前必须回答的问题

在把框架推上生产之前，建议逐条确认下面这些问题：

• 每个工具都声明了是否幂等吗？非幂等工具的重试被禁掉了吗？
• 每个工具都有独立的超时吗？超时之后后台任务会继续跑吗？
• 工具返回内容有长度上限吗？超长内容会被截断还是会被摘要？
• 上下文的裁剪规则是什么？关键信息会不会被裁掉？
• 总耗时、总步数、总工具调用次数这三个预算都设置了吗？
• 失败信息回填给模型时，能明确区分"失败"和"空结果"吗？
• 有没有并发限流？拿不到令牌时是快速失败还是无限等待？
• trace 里能不能还原出每一步模型实际收到的输入？

这八个问题里，任何一个答不上来，都建议先补上再上线。它们对应的都是真实生产环境里出现过的问题，而不是理论上的风险。

如果你的框架现在只能做到单轮问答，那也没关系，从工具注册和参数校验开始改就行，改动量比想象中小。真正需要时间的是建立起"用 trace 说话"的习惯——看到失败先拉日志，而不是先换模型。这个习惯建立起来之后，你会发现系统的改进速度会快很多。

代码和排查思路都可以直接拿去改，建议先在自己的测试环境里把超时和重试这两块跑一遍，这两个机制的调整往往能带来最直接的收益。