工具运行时上下文注入¶

工具天然是「无状态」的执行单元：LLM 决定调什么、Drive 解析参数、_run 执行业务。但有些工具不满足于此，它们需要看到 Chain 的运行时全貌 —— 当前的执行轨迹（intermediate_trace）、可用的对话历史（conversation）、神经系统的工具召回通道（neural）、本次调用自身的身份（current_tool_call_id）。本页讲清楚这套「Chain → Tool」的运行时上下文是怎么注入进来的，以及工具开发者要做什么才能用上它。

设计理念¶

为什么不用 contextvars 或全局单例？

工具的执行路径同时跨越同步与异步、本地 MiniDrive 与远程 Neural、还可能在 pytest-xdist 多 worker 下并发。任何隐式的上下文传递都会让数据流难以追踪。TFRobot 选择了显式参数注入：上下文以 kwargs 形态从 Chain 一路透传到工具的 _run / _async_run，谁需要谁取，签名即文档。

而「需不需要」由工具自己声明：通过一个布尔字段 merge_context 决定 —— 默认 False（工具是纯函数，签名干净）；显式 True 表示「我要参与 Chain 上下文」，框架才会把 kwargs 合并到 _run 的入参里。这种「先声明、后注入」的契约保证了：

纯函数工具的签名不会被无用 kwargs 污染
关心上下文的工具能在 _run 内一句 kwargs.get("intermediate") 拿到
不接受 **kwargs 的工具直接被 Pydantic / Python 在签名层挡掉，错误暴露在第一行

核心概念¶

注入物清单¶

当 merge_context=True 且 _run(self, *args, **kwargs) 签名接受可变 kwargs 时，框架在调用前会把以下三个 key 写入 kwargs：

Key	类型	含义	注入位置
`intermediate`	`IntermediateResult`	Chain 当前轮的运行时上下文。包含 `context.current_input`、`context.conversation`、`current_tool_output`、`think_times` 等	Chain 调 `mini_drive.run(..., intermediate=...)` 时透传
`neural`	`Neural \\| None`	工具所属 Robot 的神经系统。可用于反向召回其他工具或广播信号；未连接时为 `None`	`BaseTool.run / async_run` 内由 `self._neural` 自动注入
`current_tool_call_id`	`str`	本次工具调用的 `tool_call.call_id`，用于错误归因与消息关联	`Drive.run / async_run` 内由 `tool_call.call_id` 注入

注意：intermediate 由上层 Chain 调度决定是否传入，本地裸 tool.run() 直接调用时该字段就不在 kwargs 里 —— 工具应优雅降级（参考 §运行时行为 → ExpandContextTool 范式）。

merge_context 字段语义¶

merge_context: bool 是 BaseTool 上的 Pydantic 字段（默认 False）。它只控制一件事：在调用 _run(*tool_args, **tool_kwargs) 之前，是否执行 tool_kwargs.update(kwargs) 把上下文 kwargs 合进去。

LLM 给的 tool_params → 按 params_schema 解析 → (tool_args, tool_kwargs)

if merge_context:
    tool_kwargs.update(kwargs)   # 合并 intermediate / neural / current_tool_call_id

self._run(*tool_args, **tool_kwargs)

因此 merge_context=True 与 _run 签名带 **kwargs 必须捆绑出现：

开了 merge_context 而 _run 不收 **kwargs → 运行时 TypeError: got unexpected keyword argument 'intermediate'
关了 merge_context 而 _run 期待 intermediate → 取到 None，必须降级

注入调用链¶

Chain.before_do
  └── mini_drive.run(tool_call=..., intermediate=event_data.intermediate)
        └── Drive.run(tool_call, intermediate=...)
              ├── kwargs["current_tool_call_id"] = tool_call.call_id
              └── tool.run(tool_params=..., **kwargs)
                    ├── kwargs["neural"] = self._neural          # BaseTool.run 行 449
                    ├── tool_args, tool_kwargs = parse(tool_params, params_schema)
                    ├── if self.merge_context:
                    │       tool_kwargs.update(kwargs)            # 行 478-480
                    └── self._run(*tool_args, **tool_kwargs)      # 业务点

异步路径完全对称（Drive.async_run → BaseTool.async_run → _async_run）。

配置参数¶

参数	类型	说明	默认值
`merge_context`	`bool`	是否在调 `_run` 前把上下文 kwargs 合并到工具入参	`False`

运行时行为¶

工具开发者范式：ExpandContextTool¶

仓库内的 ExpandContextTool 是「读+改 intermediate」类工具的标准范例（tfrobot/drive/tool/expand_context_tool.py）：

from typing import Any, ClassVar
from pydantic import TypeAdapter

from tfrobot.drive.tool.base import BaseTool
from tfrobot.schema.brain.intermediate import IntermediateResult
from tfrobot.schema.meta.tf_field import TFField


class ExpandContextDemoTool(BaseTool):
    name: ClassVar[str] = "expand_context_demo"
    description: ClassVar[str] = "Demo: read & mutate Chain intermediate via merge_context."
    params_schema: ClassVar[TypeAdapter | None] = None

    merge_context: bool = TFField(
        True,
        title="合并上下文",
        description="本工具需读取 IntermediateResult 才能定位当前消息的 TraceView。",
    )

    def _run(self, *args: Any, **kwargs: Any) -> str:
        intermediate: IntermediateResult | None = kwargs.get("intermediate")
        if intermediate is None:
            return "[Demo] 错误：当前调用栈未提供 intermediate（可能是被裸调用）。"

        current = intermediate.context.current_input
        return f"[Demo] 当前 think_times={intermediate.think_times}, current_input={type(current).__name__}"

    async def _async_run(self, *args: Any, **kwargs: Any) -> str:
        return self._run(*args, **kwargs)

关键点：

签名带 **kwargs —— merge_context=True 的硬性约束。
优雅降级 —— kwargs.get("intermediate") 而非 kwargs["intermediate"]，本地裸调用时不至于抛 KeyError。
业务参数走 params_schema —— Chain 上下文走 kwargs，二者分离不混淆；如果声明了 params_schema，LLM 提供的参数会先按 schema 校验、解析为 tool_args / tool_kwargs，再叠加 intermediate 等系统 kwargs。

调用 neural 反向召回¶

neural 字段用于工具主动向同一 Robot 内的其他工具或外部信号源发起请求。典型用例：BaseRetrieveTool 通过 neural.signal.send 召回向量索引服务。

from typing import Any, ClassVar

from tfrobot.drive.tool.base import BaseTool
from tfrobot.neural.base import Neural
from tfrobot.schema.meta.tf_field import TFField


class NeuralBroadcastDemoTool(BaseTool):
    name: ClassVar[str] = "neural_broadcast_demo"
    description: ClassVar[str] = "Demo: send a signal via the connected Neural."
    params_schema: ClassVar[None] = None
    merge_context: bool = TFField(True, title="合并上下文", description="需访问 neural")

    def _run(self, *args: Any, **kwargs: Any) -> str:
        neural: Neural | None = kwargs.get("neural")
        if neural is None:
            return "[Demo] 当前工具未连接 Neural（独立调用模式）。"
        return f"[Demo] connected to neural={neural!r}"

    async def _async_run(self, *args: Any, **kwargs: Any) -> str:
        return self._run(*args, **kwargs)

self._neural 在工具被加入 Drive / Robot 时由 BaseTool.connect_to_neural() 自动注入；通过 kwargs 拿到的 neural 与 self._neural 等价，优先用 kwargs（语义更显式、单元测试时可被 mock 替换）。

利用 current_tool_call_id¶

current_tool_call_id 是 LLM 本次工具调用的全局唯一标识，主要用于：

工具内部错误上报时携带该 ID，便于在 intermediate.current_tool_output 中关联返回值
同一轮 LLM 触发多个工具并发时，按 call_id 区分各工具输出

from typing import Any, ClassVar

from tfrobot.drive.tool.base import BaseTool
from tfrobot.schema.exceptions import TFExecutionError
from tfrobot.schema.meta.tf_field import TFField


class CallIdAwareTool(BaseTool):
    name: ClassVar[str] = "call_id_aware"
    description: ClassVar[str] = "Demo: surface current_tool_call_id in errors."
    params_schema: ClassVar[None] = None
    merge_context: bool = TFField(True, title="合并上下文", description="需要 call_id")

    def _run(self, *args: Any, **kwargs: Any) -> str:
        call_id = kwargs.get("current_tool_call_id", "<local>")
        # 业务失败时把 call_id 带到异常里，方便 Chain 端定位
        if not kwargs:
            raise TFExecutionError(f"empty kwargs (call_id={call_id})", call_id=call_id)
        return f"[Demo] running under call_id={call_id}"

    async def _async_run(self, *args: Any, **kwargs: Any) -> str:
        return self._run(*args, **kwargs)

与 params_schema 的关系¶

params_schema 控制业务参数的解析，merge_context 控制系统参数的注入。二者各管一摊：

来源	进入路径	例
LLM 给的 `tool_params` JSON	`params_schema` 解析 → `tool_args / tool_kwargs`	`{"op": "tail", "increment": 10}`
Chain 注入的运行时 kwargs	`merge_context=True` 时 `tool_kwargs.update(kwargs)`	`intermediate=..., neural=..., current_tool_call_id=...`

注意「同名 key 会覆盖」：合并发生在业务参数之后，因此 Chain 系统 kwargs 优先级更高。业务参数命名应避免使用 intermediate / neural / current_tool_call_id 这三个保留名。

使用建议¶

默认关 merge_context：纯函数工具签名最干净；只有真要读 Chain 状态才开。
永远 kwargs.get(...) 而非 kwargs[...]：本地单测、单步调试时工具会被裸调用，缺字段就降级到提示信息，比抛 KeyError 友好。
业务侧昂贵 IO 提前算：_run 收到 intermediate 后只做读 + 轻量写。intermediate.context.current_input 是引用传递，对其修改会在下一轮 on_enter_thinking 重建 PromptContext 时立即生效，不要在工具里发起昂贵的远程调用，那是 Chain 自身的职责。
想用 neural 召回？先检查 is None：工具可能被某个未连接 Neural 的 Robot 加载（裸单测、迁移期间），缺这层兜底会直接挂掉。

与其他模块的协作¶

Chain (tfrobot/brain/chain/)：注入源。Chain.before_do / before_async_do 把 event_data.intermediate 传给 mini_drive.run / async_run。
Drive (tfrobot/drive/base.py)：中转。把 tool_call.call_id 包装进 kwargs，再把 kwargs 透传给 BaseTool.run。
Neural (tfrobot/neural/)：双向。connect_to_neural 在工具入注册时绑定 self._neural；运行时通过 kwargs 暴露给 _run，让工具有能力反向召回。
IntermediateResult (tfrobot/schema/brain/intermediate.py)：上下文数据载体。详细字段请参考 Chain 思维链与 intermediate-trace 设计。

如果工具需要的是「向 LLM 推 Prompt 段」而不是「读 Chain 状态」，请走另一条路径：工具 Prompt 段位注入。