构建并复用 JIT 能力#

一次对话可能需要超出固定工具集的工具。 JITHarnessExtension 让受信任的模型编写可复用的 Python 工具或 Markdown 工作流、组合已有的工具、增强自身能力,并卸载不再需要的工具。你的应用可以保存这个工具箱,并基于它启动新会话。

JIT 指的是在任务需要时才构建工具环境,而不是 Python 即时编译器。该功能在 Responses 和 Chat Completions 上都使用普通的函数调用。它不需要原生的 tool_search,不过 Responses 端点可以延迟加载 控制模型看到什么 中描述的检查工具。

警告

生成的 Python 以你的应用的权限运行。源码一被激活,导入语句和模块级语句就会执行,而且发生在任何工具调用之前。这不是沙箱。只有在执行受信任代码是合适场景时才启用它,并在激活来自其他地方的源码之前先审查快照。

启用会话本地的工具箱#

提供一个模型,以及你的应用自身拥有的基础工具:

from zett_agent.agent import AgentRunConfig
from zett_agent.client import create_agent
from zett_agent.extensions.memory import InMemoryMessageAccumulator
from zett_agent.jit_harness import JITHarnessExtension

jit = JITHarnessExtension()
client = await create_agent(
    model,
    config=AgentRunConfig(session_id="builder"),
    tools=[add],
    extensions=[InMemoryMessageAccumulator(), jit],
)
reply = await client.run("Build and use a reusable tool that doubles an integer.")

请在异步函数中运行这段代码。 model 是一个已配置好的、支持函数调用的模型提供商, add 则是你定义的类型化基础工具。该扩展会提供当前的工具 guidelines,因此不需要再添加 ToolGuidelinesExtension。显式给出的扩展列表会替换掉可选默认项;请像上面那样包含历史记录,或使用 会话持久化。

由模型决定是否创建工具。创建过程不会测试工具的主体逻辑:注册成功只表示 Python 模块已加载、工具定义通过了校验,并不表示其实现能解决任务。

从高成本的工作流中学习#

当一项任务涉及大量探索性步骤,或者耗费可观的时间和 token 时,harness 的 guidelines 会请模型考虑把行之有效的做法保留下来。如果步骤稳定、可以自动化,就选择 Python;如果这条路径仍需要判断,例如挑选相关证据或解读错误,就选择文本。两种形式都不应变成对话的堆砌。请保留其中有用的路径、适用条件、依赖、已知的坑以及结果检查方式。

文本工具是一段可复用的 Markdown 流程,类似于 SKILL:

from zett_agent.jit_harness import JITHarness, ToolKind

harness = JITHarness()
harness.create_tool(
    kind=ToolKind.TEXT,
    name="investigate_test_failure",
    description="Use when a test fails and the cause is not yet known.",
    source="""# Investigate a failing test

1. Reproduce the smallest failing test with the available test runner.
2. Read the failing assertion and the relevant source, not the whole repository.
3. Compare the expected behavior with the observed inputs and result.
4. Verify the diagnosis by rerunning that test and related regression tests.

Record a confirmed cause, not a guess. Do not copy credentials into reports.
""",
)
page = await harness.call_tool("investigate_test_failure")
print(page["content"])

模型使用同一个 create_tool,并传入 kind="text"。name 和简短的 description 是必填项;entrypoint 必须省略。默认仍然是创建 Python 工具,因此现有的 create_tool(source, entrypoint=...) 调用依然可用。

已加载的文本工具会暴露一个读取器,默认参数为 offset=0 和 max_chars=2000。模型调用它来读取相关步骤,然后配合可用的工具去执行。读取文本不会执行 Python,也不会自动运行工作流。完整正文不会注入每一次模型请求:只提供该工具的选择提示、读取器模式和 guidelines。较长的工作流与读取源码时一样,按同一套精确且受预算约束的分页读取。

update_tool(name, markdown, description=...) 会替换文本工作流,而不改变它的类型或名称;省略 description 会保留当前的提示。编辑之前,可以用 read_tool_source 查看完整路径。卸载会隐藏读取器,同时保留其内容;加载则恢复它。Python 工具和文本工具都会保存到应用的快照中,供后续会话复用。在应用保存之前,学习成果只保留在内存中。

请用有代表性的任务验证所提出的优化,并把创建和测试它的成本计算在内。guidelines 鼓励这种行为,但并不保证模型一定能学会,也不保证保存下来的路径全局最优。

选择是否管理基础工具#

manage_base_tools=True(默认值)允许 harness 卸载/重新加载并分析应用自有的本地工具,也包括由扩展提供的工具。在初始化过程中稍后注册的工具,或被某个策略钩子替换的工具,都会在模型/工具分发之前被接管。同步操作不会恢复被策略移除的工具。

如果想保护基础工具箱,并选择不对它做性能分析:

jit = JITHarnessExtension(manage_base_tools=False)
# The standalone JITHarness constructor accepts the same option.

在这种模式下,基础工具保留它们正常的请求处理函数。 unload_tool 和 load_tool 会拒绝基础工具的名称;不会收集基础工具的调用测量数据或近期错误。动态 Python 工具仍然可以组合当前已注册的基础工具,参数校验、中间件和工具钩子都保持有效。动态调用和管理调用仍保留各自的性能分析。父工具记录的耗时仍然包含花在基础依赖上的时间。

两种模式下,基础实现都无法更新,也无法序列化为 Python 源码。快照会保留基础工具的名称和卸载状态,应用必须在每个请求中重新提供实现。该选项不会写入快照:由接收方应用自行选择。在基础工具管理被禁用的情况下加载快照时,快照中已有的基础工具卸载标记会被保留下来,但不会强制执行;动态工具的卸载标记仍会强制执行。之后在启用管理的情况下加载该快照时,被保留的基础工具卸载标记会重新生效。

性能分析只观察经由 Agent/harness 路由的调用。直接调用应用原有的 AgentTool 会绕过 harness,因此既无法被统计,也无法通过卸载请求中的注册来阻止。

模型的管理工具#

工具

行为

create_tool(source, ...)

使用 entrypoint 创建 Python 工具,或使用 kind="text"、name 和 description 创建文本工具。files 用于添加工具包模块。

update_tool(name, ...)

替换或添加动态工具的文件、源码或入口; remove 用于删除某个文件。一次发送的所有内容会编译成一个版本。

unload_tool(name)

隐藏可调用的注册,同时保留源码及其名称。

load_tool(name)

恢复已保留的工具,而无需重新生成代码。

list_tools(offset=0, limit=20)

读取一页很小的目录:只有名称、状态和描述预览。

inspect_tool(name)

读取定义、参数和工具包文件大小,不包含源码、性能分析或历史记录。

read_tool_source(name, path=None, offset=0, max_chars=2000)

读取某个工具包文件中一段精确的片段,默认为入口模块。

read_tool_schema(name, offset=0, max_chars=2000)

当定义中报告 parameters_omitted 时,读取 JSON Schema 片段。

inspect_tool_stats(name)

只读取该工具整个生命周期的聚合测量数据。

inspect_tool_calls(name, offset=0, limit=5)

读取精简的单次调用测量数据,不含源码或聚合数据。

inspect_tool_errors(name=None, offset=0, limit=3, max_chars=512)

读取近期错误消息的预览和出现次数,可按工具筛选。

管理工具的名称是保留的。模型无法覆盖或卸载这些工具,也无法替换应用提供的基础工具。所有管理操作都使用运行时的串行调度模式。如果一次响应中还包含普通的并行调用,它们会在串行的管理调用之前完成。

最后五个工具用于读取保留的细节,属于 inspection tier:read_tool_source、read_tool_schema、inspect_tool_stats、inspect_tool_calls 和 inspect_tool_errors。其余工具是负责构建、修改和列出能力的控制循环。控制模型看到什么 说明了如何暴露 inspection tier。

会改变状态的工具 —— create_tool、update_tool、unload_tool 和 load_tool —— 只返回 name 和 status,例如 {"name": "twice", "status": "loaded"}。它们不返回性能分析、元数据、源码或调用历史。性能分析仍会照常记录,需要时可以通过 inspect_tool_stats、inspect_tool_calls 或 Python 的 profile() API 查询。

每个查询只有一个用途。发现类查询不会获取性能分析;参数检查不会获取源码或历史记录。面向模型的查询响应限制在 8,000 个字符的序列化 JSON 预算之内,其中计入转义和标点。 limit 为 1–100, max_chars 为 1–4,000。两者都是上限:预算可能截短某一页,因此始终要按 next_offset 继续读取。

目录中的描述最多预览 200 个字符;定义中的描述最多预览 500 个。源码和模式片段是精确内容,而不是摘要:请按顺序拼接,并在读取期间工具发生变化时,比对各页之间的 content_id。过大的模式会明确从 inspect_tool 中省略,同时给出 parameters_omitted 和 parameters_chars;可以用 read_tool_schema 恢复完整且合法的 JSON。基础工具没有保留的 Python 源码,因此 read_tool_source 会拒绝它们。

应用侧的方法仍然保持完整:harness.list_tools() 返回完整目录,definition(name) 返回完整的元数据和模式,source(name) 返回完整的动态源码,inspect_tool(name) 返回模式、guidelines 和源码。测量数据仍由各自的方法提供:profile_stats(name) 返回聚合数据,profile(name) 返回单次调用的记录。这些方法不受模型响应预算的限制。

控制模型看到什么#

每一次请求都会发送全部定义和指令。inspection tier 是大多数会话根本用不到的部分,因此由 JITHarnessExtension(inspection_tools=...) 决定它如何到达模型:

模式

行为

InspectionToolExposure.DEFERRED(默认)

把检查相关的定义标记为延迟加载。支持延迟加载的模型提供商(OpenAI Responses)会通过其托管的 tool_search 端点按需获取某一个定义。忽略该标志的协议 —— Chat Completions、Anthropic、Google 和 Ollama —— 仍会收到全部定义,因此在这些协议上不会丢失任何能力:模型始终可以读取源码、模式、测量数据和错误历史。

InspectionToolExposure.ALWAYS

在第一次请求时就发送全部管理定义,效果等同于该层级并不存在。

InspectionToolExposure.HIDDEN

完全不暴露检查工具。应用仍然保留 profile、profile_stats、tool_errors、source 和 inspect_tool,但模型无法自行检查代码或测量数据。

这一选择也会影响提示词:被隐藏的检查工具不会被提及名称,而可能延迟加载它们的端点会被告知可能需要搜索。其他行为都不变;工具的名称、串行调度和输出预算都保持原样,管理工具的名称在任何模式下都保持保留状态。

在默认工具箱上按 cl100k_base 分词统计,一次请求的开销大致为:

配置

Responses

Chat Completions

ALWAYS

3,230 tokens

3,230 tokens

DEFERRED(默认)

2,370 tokens

3,250 tokens

HIDDEN

2,080 tokens

2,080 tokens

每个数字都涵盖控制类工具的定义加上指令块,不含工具 guidelines 和你自己的工具。延迟加载可以省下检查相关的定义(约 880 tokens);隐藏则会移除这些定义及其指令(再多约 290 tokens)。当模型需要能自行诊断自己的工具时,保持默认值;不需要时则隐藏。在 Chat Completions 上,真正起作用的手段也是隐藏,因为延迟加载在那里没有效果 —— 该协议总是接收全部定义。

延迟加载依赖模型提供商的一项能力,因此它永远不会改变 Chat Completions 的行为。隐藏则纯粹是应用侧的决策,而在任何模式下 harness API 都保持完整。

这些控制类定义比引入工具包文件和程序化 guidelines 之前更大。其中大部分体积来自 JSON Schema 本身的呈现:每个可选参数都写成 anyOf 联合类型,每个属性都带有一个生成出来的 title,这十一项工具加起来约占上面数字中的 280 tokens。

编写并组合工具#

Python 模块会收到 tool、ToolExecutionMode 和 harness。源码中可以包含导入语句、辅助函数和类。请传入某个被装饰的入口的变量名,例如 entrypoint="twice":

@tool(guidelines="Use to add an integer to itself.")
async def twice(value: int) -> int:
    """Add the same integer twice."""
    return await harness.call_tool("add", left=value, right=value)

工具契约与 定义并注册工具 相同:带类型注解的参数、一段描述,以及非空的 guidelines。JIT 工具是普通函数,不是延迟加载的工具,也不是客户端搜索工具。

在异步工具中使用 await harness.call_tool(name, **arguments)。对于普通的 def 函数,请使用阻塞式的组合入口:

@tool(guidelines="Use to add an integer to itself.")
def twice(value: int) -> int:
    """Add the same integer twice."""
    return harness.call_tool_sync("add", left=value, right=value)

两条路径都会检查当前的注册,并执行参数校验、中间件和必要的工具钩子。依赖返回的是 Python 值,而不是 JSON 字符串,除非某个结果钩子有意对它做了转换。除非你的 Python 代码捕获了异常,否则异常会向上传播到外层工具。只有外层调用的结果会记录到模型对话中;不会产生嵌套的执行轨迹。请求用户输入或审批的依赖仍然会执行那次交互。

同步代码会在事件循环之外运行。不要在异步代码中调用 call_tool_sync,也不要在没有正在运行的同步工具时调用它。组合深度受 max_call_depth 限制(默认 32)。取消无法强制停止任意的同步 Python 代码,也无法撤销副作用;请为阻塞式 I/O 单独设置超时,并避免游离在外的后台任务。

程序化工具调用(PTC)#

工具内部的组合就是程序化工具调用(PTC):模型编写会调用工具的代码,而不是每个模型轮次只请求一次工具调用。这样一次外层调用就可以代替几十个轮次。

from .batch import summarize

@tool(guidelines="Use to summarize every requested file in one call.")
async def summarize_files(paths: list[str]) -> str:
    """Read many files in code and return one compact summary."""
    return await summarize(harness, paths)

其中循环体存放在 batch.py 里:

async def summarize(harness, paths):
    lines = []
    for path in paths:
        text = await harness.call_tool("read_file", path=path)
        lines.append(f"{path}: {len(text)} characters")
    return "\n".join(lines)

这样做的收益:

  • 更少的模型轮次。 循环、分支和重试都在 Python 中进行,因此对话只增长一次,而不是每次工具调用都增长一次。

  • 更小的上下文。 嵌套结果留在 Python 中。只有外层工具的返回值会成为 ToolMessage;中间载荷永远不会发回模型,也不会追加执行轨迹。

  • 可观测性保持不变。 每次嵌套调用都会由性能分析记录测量数据,并连同它的来源一起发出 jit_tool_started / jit_tool_completed,因此应用仍然能看到实际运行了什么。

设计时需要留意的限制:

  • 依赖必须处于已加载状态,深度受 max_call_depth 限制。

  • 嵌套调用仍会执行工具钩子和中间件,因此请求用户输入或审批的依赖仍然会中断。

  • 参数校验会作用于嵌套调用,并且除非你的 Python 代码捕获了异常,否则异常会向外传播。

  • 让返回值便于 JSON 处理:它就是模型读到的内容。

把工具拆分到多个模块#

大型工具可以做成一个小型工具包,而不是一个冗长的模块。入口充当工具包根目录的角色,其他文件通过相对导入访问:

jit.get_harness("builder").create_tool(
    source=entry_source,
    entrypoint="summarize_files",
    files={
        "batch.py": batch_source,
        "parsers/__init__.py": parser_source,
    },
)

嵌套目录会变成子包,因此 from .parsers import parse 和 from .parsers.lines import split 都能正常工作。路径各段必须是可导入的 Python 名称,根目录下的 __init__.py 就是入口本身,而一个工具包最多包含 32 个文件,每个文件最多 200,000 个字符。

读取和编辑都保持按文件进行:

  • inspect_tool(name) 会列出工具包中的文件及其字符数,入口以 __init__.py 列出。

  • read_tool_source(name, path="batch.py") 按页读取单个文件;默认是入口。

  • update_tool(name, files={...}, remove=[...]) 会添加或替换你发送的文件,并删除你指定的文件。未包含在内的文件会保留原有内容,因此小改动只带来一次小更新。

  • 一次调用中的所有内容都会作为一个新版本统一校验并编译。因此跨文件重构会一次生效,而导入损坏之类的失败则会保留上一个版本。

有两点后果值得注意。第一,工具注册时就会执行导入语句和模块主体,就像一次普通的导入那样,因此辅助函数中的 SyntaxError 或 ImportError 会在创建或更新时暴露出来。第二,之后该工具包不会保留在 sys.modules 中,所以应在模块层级导入工具所需的内容,而不是在函数内部延迟导入。

工具包在 unload_tool 和快照中都能保留下来。只要快照中包含至少一个多文件工具,就会写入 format_version: 2;单模块工具则继续写入版本 1,因此现有文件保持不变,仍可正常读取。

卸载而不丢失源码#

unload_tool 会立即把某个工具从当前可调用定义中移除,也从同一会话后续的请求中移除。它不会抹去更早的对话消息,不会收回 Python 权限,也不会终止已经开始的调用。依赖它的工具仍保持注册;调用已卸载的依赖会抛出明确的错误。加载会恢复该定义。

unload_tool 返回 status="unloaded"; load_tool 返回 status="loaded"。快照把被隐藏的工具名存放在 unloaded_tools 中。

更新已卸载的工具会保留它的卸载状态。更新不能重命名工具。已在进行的调用会以它们启动时的实现完成;后续调用则使用通过校验的替换版本。源码加载失败会让旧的注册保持原样,但无法回滚文件写入之类的模块级副作用。

借助性能分析构建有用的工具#

每个 harness 都会为基础工具、动态工具和管理工具记录会话本地的测量数据。模型通过 inspect_tool_stats 读取聚合计数器,通过 inspect_tool_calls 读取单次调用的测量数据。guidelines 要求它在保证正确性、完整性、安全性和回答质量的前提下,以尽可能少的总 token 用量和端到端时间完成任务。它鼓励在有帮助时进行 JIT 优化,并把检查、工具创建和校验的成本一并纳入权衡。当现有工具或直接方案已经能高效完成任务时,仍然优先选择它们。相关的测量数据能帮助模型决定改进什么,并验证结果。

要留意那些可以由一个专注的工具来组合的、反复出现的调用方/依赖关系,那些在到达模型之前就能缩减的大体积输出,以及那些缓慢或容易失败的实现。性能分析反映的是关于执行的证据,而不是质量评分:很少被调用的工具可能依然不可或缺,字符更少本身也不代表回答更好。

应用代码可以检查同一份数据:

harness = jit.get_harness("builder")
report = harness.profile("twice", offset=0, limit=100)
stats = report.stats[0]
print(stats.call_count, stats.average_duration_ms)
print(stats.average_input_chars, stats.average_output_chars)
for call in report.calls:
    print(call.call_id, call.caller_tool_name, call.status, call.duration_ms)

聚合统计按结果统计已完成的调用次数,并给出平均耗时以及平均输入和输出字符数。每条已完成的记录都带有它的调用 ID、工具名、来源、结果、耗时、测量到的长度,以及嵌套调用时的组合方调用者,失败时还带有错误类型。生命周期累计值不会因更新或卸载而重置。

每条记录还带有一个 ToolOrigin,用来指明是哪个注册表处理了这次调用:

  • base —— 来自 tools=[...] 的应用自有工具,或由某个提供工具的扩展提供。在启用基础工具管理时,它可以被卸载/重新加载并接受性能分析。它的实现没有保留源码,既不能更新也不能序列化;快照只保留它的名称和卸载状态。

  • python —— 模型用 create_tool 创建、或用 update_tool 替换的动态工具。已卸载的动态工具仍然保持 python;它的调用状态会变成 rejected。

  • text —— 已保存的 Markdown 工作流。调用测量的是读取工作流某一页,而不是执行它的步骤,也不衡量任务最终的质量。

  • management —— 保留的控制函数之一,例如 create_tool 或 inspect_tool。

  • unknown —— 一种防御性兜底值,用于记录该次尝试时不在任何注册表中的名称。正常分发不会产生它。

list_tools 和 inspect_tool 对每个工具的目录 origin 使用相同的取值;在那里只会出现 base 和 python。枚举成员与它们的字符串值比较时相等,因此 origin == "python" 依然可用,JSON 导出中会包含上面所示的纯字符串 "base"、"python"、"management" 和 "unknown"。

status 字段是一个 ToolCallStatus,描述该次尝试是如何结束的:

  • completed —— 处理函数正常返回。之后出现的序列化或生命周期钩子失败不会改变该记录,因此 completed 表示工具本身已经运行过。

  • failed —— 工具已启动并抛出异常:可能是参数校验、处理函数代码,或未捕获的嵌套依赖错误。

  • cancelled —— 等待的调用方被取消。同步工作线程可能仍在运行,因为 Python 无法强制停止它。

  • rejected —— harness 在任何处理函数主体运行之前就拒绝了该调用:工具已卸载、缺少它的基础工具绑定、在当前请求中已不再注册,或者嵌套调用超过了 max_call_depth。

近期错误记录复用同一个枚举,但只会保存 failed 或 rejected;用成功或取消状态来构造这类记录会被拒绝。

解读结果时请遵循这些测量规则:

  • 输入长度按紧凑的 Unicode JSON 计算,包含属性名和标点。文本输出使用普通字符串长度;结构化输出使用紧凑 JSON。多模态输出只统计文本块,不含图像字节和 URL。这些数字是字符数,而不是字节数、token 数,也不是模型提供商载荷的精确大小。

  • 缺失的测量值是 None,而不是零。字符平均值只使用有测量数据的样本。错误输出统计的是错误消息的长度;取消没有输出测量值。

  • 测量在 8,000 个字符处停止,这与面向模型的响应使用同一预算,因此报告为 8,000 意味着“至少这么多”。超过这个上限后,精确测量只会增加分发耗时,却不会改变任何决策。

  • 耗时是分发过程的实际用时,其中包括校验、同步工作线程分发和嵌套调用。外层的中间件、生命周期钩子、用户审批等待以及输入/输出测量都不在这个边界之内。父调用包含它的子调用,因此把两者相加会重复计算嵌套工作。被取消的同步分发会在其 await 被取消时停止计时;Python 无法强制停止工作线程,它可能会稍后才结束。

  • 中间件的短路不会被计入,因为工具从未执行。未知名称、失效的请求引用,以及被拒绝的模块加载调用也不会产生记录。

性能分析在内存中只保留一个有界的已完成尝试窗口,而不保留它们的输入/输出内容或异常消息:最新的 call_capacity 条记录(默认 2,000)会保留下来,最先被分发的先淘汰,同时生命周期聚合数据会继续统计每一次尝试。当长时间会话需要更小的窗口时,可以给 JITHarness 或 JITHarnessExtension 传入另一个 call_capacity。这不会改变正常的对话历史,也不会额外添加嵌套执行轨迹事件。各个会话的性能分析彼此隔离。它们能在同一个内存中的 harness 里跨后续请求保留,但不会包含在可执行快照中:激活快照会重新开始测量。正在进行的调用不会被发布;它们的记录会在该次尝试结束时出现。

模型的 inspect_tool_calls 返回 calls、total_calls 和 next_offset。它默认返回五条记录,并省略工具名、空字段和请求元数据。调用方关系和错误类型只在适用时出现。它不返回源码、模式或聚合统计。

应用侧的 harness.profile 使用 offset、limit(默认 100)和 next_offset;调用记录在这里分页,而不是通过 inspect_tool。聚合数据始终覆盖已观测到的完整生命周期,而不仅是当前这一页。如果你在读取全部记录时需要稳定的分页,请在请求之后导出:

json_page = report.model_dump_json()
# Store pages using your application's storage.
from zett_agent.jit_harness import JITHarnessProfile

restored_page = JITHarnessProfile.model_validate_json(json_page)

对于长时间会话,记录保留占用的内存与调用次数成正比。请导出你需要的内容,并用 forget_session 释放空闲的 harness。性能分析导出不会激活代码,也不会为新 harness 的测量数据提供初始值。

诊断近期错误#

inspect_tool_errors 把近期的错误消息交给模型,用来决定下一步该怎么做:用修正后的参数重试、加载某个依赖、修复动态 Python 源码,或者解释某个应用侧的失败。它是一个只读工具;查询错误不会重试调用、不会更新源码,也不会保存快照。

每个会话都有一个独立的、用于存放不同错误的 LRU 缓存。默认保留 100 条条目,每条消息最多 4,000 个字符:

jit = JITHarnessExtension(error_capacity=100, error_message_chars=4000)
# After a request:
report = jit.get_harness("builder").tool_errors("twice", limit=20)
for error in report.errors:
    print(error.error_type, error.message, error.occurrences)
    print(error.last_call_id, error.caller_tool_name)

这两个配置值都必须是正整数。容量对整个 harness 生效,而不是为每个工具单独分配。错误按工具名、异常类型和完整的错误消息去重。重复出现的错误会增加 occurrences,并把该条目移到最近的位置;缓存满时,最近最少出现的条目会被淘汰。读取不会刷新新旧顺序,成功的调用也不会清除旧的证据。

记录中包含错误类型、被限长的消息、最后出现时间、出现次数,以及最近一次调用的调用方和调用 ID。 message_truncated 表示是否只保留了前缀。截断后前缀相同的不同消息仍然分别保留。如果某个异常无法格式化,则保留它的类型和上下文,并设置 message=None。

occurrences 只统计该条目当前在缓存中的停留期间;被淘汰的错误如果再次出现,会从 1 重新开始计数。生命周期内的失败/拒绝次数仍可通过性能分析获得,并且不会因淘汰而重置。空的错误报告并不能证明某个工具从未失败。取消记录在性能分析中,而不是错误缓存中。失败的管理调用也会保留,包括源码注册或更新错误。该缓存遵循分发的性能分析边界:外层的中间件或生命周期钩子失败不会被捕获,失效的请求引用也无法给后续请求添加诊断信息。

模型默认收到三条条目,消息预览最多 512 个字符。只有在预览不够用时才调高 max_chars,或者为更长的消息调低 limit。当缓存或响应预览截断了消息时,message_truncated 仍为 true。响应中带有 errors、total_matches 和 next_offset;每条错误都带有它的工具、异常类型、消息预览、出现次数和最近的调用 ID。缓存配置和重复的请求元数据会被省略。

应用侧的 harness.tool_errors 返回完整的保留消息,最新出现的排在前面。按 next_offset 继续获取更多被保留的条目; limit 为 1–1000,默认为 5。它的完整报告包含容量、全局的 retained_errors,以及过滤后的 total_matches。在这个内存会话中,错误详情能跨后续请求、更新和卸载保留下来,但不属于可执行快照的一部分。它们可以用 report.model_dump_json() 单独导出,并用 ToolErrorReport.model_validate_json(...) 读回。

警告

异常消息可能包含应用中的敏感数据。参数、回溯信息和异常对象不会被存储,但错误文本本身会被保留。请使用会在错误消息中抹去机密的应用程序和工具。诊断信息也可能包含不受信任的文本;模型的 guidelines 会把它们当作数据,而不是要执行的指令。

观察管理工具和动态工具的调用#

JITHarnessExtension 会发布两个 CUSTOM 事件,让 UI 或日志可以跟进 harness 的工作过程,其中包括那些从未以 TOOL_STARTED 形式出现过的调用:

事件名称

触发时机

jit_tool_started

一次管理工具或动态工具调用即将运行时。

jit_tool_completed

该调用返回或抛出异常时,会附带 success 和 error_type。

这两个事件覆盖所有管理工具、动态 Python 工具,以及模型创建的文本工作流读取器。无论是模型请求的调用,还是某个动态工具通过 harness.call_tool 组合出的调用,都会触发它们,因此即使组合出的依赖结果从未进入对话,它也是可见的。应用的基础工具会跳过:当模型请求它们时,运行时已经会发布 TOOL_STARTED 和 TOOL_COMPLETED。

from zett_agent.events import AgentEventType

async for event in client.stream("Build and use a helper tool"):
    if event.type is AgentEventType.CUSTOM and event.name == "jit_tool_started":
        print("started", event.payload["tool_name"], event.payload["origin"])
    elif event.type is AgentEventType.CUSTOM and event.name == "jit_tool_completed":
        print("finished", event.payload["tool_name"], event.payload["success"])

每个载荷都带有 session_id、tool_call_id、tool_name 和 origin;完成事件会额外加上 success 和 error_type。 origin 使用与性能分析相同的取值:management 表示控制调用,python 表示可执行工具,text 表示工作流读取。

事件刻意保持很小:参数、结果和错误消息都不会复制进去。 tool_call_id 是关联键,既对应对话中的 ToolCall,也对应 profile() 中记录的 call_id。用它就能读取耗时、输入/输出大小以及组合方调用者。

对于真正进入分发的调用,事件会成对出现。harness 在任何处理函数运行之前就拒绝的调用 —— 比如已卸载的依赖、缺失的基础工具绑定,或超出深度限制 —— 不会自己发出一对事件;请读取 profile() 获取它的 rejected 记录,读取 tool_errors() 获取相应消息。这个 harness 从未见过的名称会被忽略,而不会作为 JIT 调用上报;另一个扩展中的生命周期钩子若在处理函数之前抛出异常,就可能留下一个只有开始、没有完成的事件。

保存并在新会话中使用#

保存由应用控制,绝不会自动发生:

from zett_agent.jit_harness import JITHarnessSnapshot

jit.get_harness("builder").save("harness.json")
snapshot = JITHarnessSnapshot.load("harness.json")
reused = JITHarnessExtension(snapshot)
other = await create_agent(
    model,
    config=AgentRunConfig(session_id="next-session"),
    tools=[add],
    extensions=[InMemoryMessageAccumulator(), reused],
)
reply = await other.run("Use the saved toolbox.")

每个会话都会收到该模板的一份独立副本。同一会话中后续的请求会保留它的演进;修改一个会话不会影响另一个会话,也不会覆盖该文件。保存目标的父目录必须存在。替换是原子操作,因此写入中断不会暴露出写了一半的 JSON 文件。

快照包含带格式版本的 Python 源码、入口、工具元数据、Markdown 工作流、基础工具名称和卸载状态。它们不包含对话历史、可调用对象、凭据、连接或可变的 Python 全局变量。重新加载会依据源码重建模块全局变量。Python 依赖必须安装在目标应用的环境中。Python 记录保存在 tools 中;Markdown 记录使用 text_tools。较旧的、仅含 Python 的格式 1 快照加载时,text_tools 集合为空。

JITHarnessSnapshot.load 和 from_json 只校验数据,不会执行源码。 JITHarness(snapshot, tools=[...]) 会直接激活代码,或者由扩展在新会话的第一次请求时激活。已卸载工具的模块也会被激活;卸载控制的是能否调用,而不是加载产生的副作用。缺失的基础工具不会阻止激活,但调用其中之一会明确失败。请通过 tools=[...] 或提供工具的扩展来绑定基础工具。请求作用域的绑定会在每次请求后释放,因此该请求之外的调用无法复用已关闭的应用资源。

JITHarnessExtension 通过 harness 自身的生命周期 API 来驱动这种绑定:attach(context, management_tools) 绑定一个请求并挂载它的工具,detach(context) 释放它,而 active 属性报告当前是否有请求处于绑定状态。扩展会成对调用它们;应用通常最多只读取 active。重复附加会抛出异常,而分离一个更早的请求会被忽略,这样较晚的清理就不会拆掉当前正在运行的请求。

请用 snapshot.to_json() 和 JITHarnessSnapshot.from_json(...) 接入你自己的存储。对于长时间运行的服务器, jit.forget_session(session_id) 会释放空闲的内存工具箱,而不会删除快照或对话历史。

试用完整的离线示例#

这个预设脚本的模型会创建 twice、调用它,并在一个独立会话中复用它保存下来的源码。程序还会检查同步 → 异步 → 基础工具的流水线、空输入、失败的更新、依赖卸载以及快照隔离。它使用临时目录,不发起任何网络请求。它还会保存、卸载、恢复并读取一个可复用的文本工作流。该预设脚本的模型还会在一次工具调用之后检查性能分析,程序则校验聚合数据和嵌套调用的测量数据:

async def main() -> None:
    jit = JITHarnessExtension()
    client = await create_agent(
        BuilderModel(),
        config=AgentRunConfig(session_id="builder"),
        tools=[add],
        extensions=[InMemoryMessageAccumulator(), jit],
    )
    assert (await client.run("Build and use twice.")).content == "42"
    measurements = jit.get_harness("builder").profile()
    assert next(stat for stat in measurements.stats if stat.tool_name == "twice").completed_count == 1
    assert next(call for call in measurements.calls if call.tool_name == "add").caller_tool_name == "twice"
    with TemporaryDirectory(prefix="zett-jit-example-") as directory:
        path = Path(directory) / "harness.json"
        jit.get_harness("builder").save(path)
        reused = JITHarnessExtension(JITHarnessSnapshot.load(path))
        other = await create_agent(
            BuilderModel(),
            config=AgentRunConfig(session_id="reused"),
            tools=[add],
            extensions=[InMemoryMessageAccumulator(), reused],
        )
        assert (await other.run("Use the saved twice tool.")).content == "42"
        await exercise_lifecycle(Path(directory) / "pipeline.json")
    print("JIT tool created and reused: 42")
    print("JIT invariants checked: composition, failed update, unload, snapshot isolation")
    print("JIT profiling checked: per-tool totals and nested call records")
    print("JIT diagnostics checked: recent dependency error retained")
uv run python docs/_examples/jit_harness.py

预期输出:

JIT tool created and reused: 42
JIT invariants checked: composition, failed update, unload, snapshot isolation
JIT profiling checked: per-tool totals and nested call records
JIT diagnostics checked: recent dependency error retained

下载 完整程序。若需精确的契约,请参阅 JITHarness、JITHarnessSnapshot 和 JITHarnessExtension。