Intermediate Trace 系统设计方案¶

字段	值
状态	待实现
日期	2026-04-02
涉及模块	`brain/chain`、`brain/dc_brain`、`brain/chain/llms`、`drive/tool`

背景与问题¶

现状¶

intermediate_msgs 是 Chain 生命周期内 LLM 与工具交互产生的过程性消息。当前的传递与生命周期管理存在以下问题：

场景	当前行为	问题
单 Chain 内部	全程累积，LLM 可见	正常
SeqChains 跨链	共享引用，`pass_intermediate_msgs` 控制是否重置	粗粒度控制开关，无法区分"关键事实"与"推理过程"；控制权在容器而非 LLM
ParallelChains	每个子链深拷贝隔离	并行子链之间无法共享中间结果
DCChains 跨链	完全不传递，TFChainInterpreter 调用子链时不传 `intermediate_msgs`	文件读取结果、搜索结果等大型工件跨链丢失
TFUserInterrupt 中断	中断时 intermediate_msgs 直接丢失	下次 run() 时 Robot 完全不知道发生了什么，与用户看到的产生本质冲突

核心矛盾¶

Chain 拆分的设计目标是"提供专属 SOP Prompt + 节约上下文"，但实践中无法做到"结果 100% 重要、过程完全不重要"。关键矛盾：

完全不传递：调度方（DCChains LLM）无法传递几 KB 的文件内容，子链被迫重复调用工具
完全传递：上下文急剧膨胀，节约上下文的设计目标完全废弃
LLM 筛选：成本高、速度慢，不可取
中断上下文丢失：用户发下一条指令时，指令内容基于用户看到的执行过程，而 Robot 毫无记忆

设计目标¶

持久化：intermediate_msgs 不再是运行时临时数据，伴随用户消息一起持久化
树状结构：天然反映 Chain 嵌套层级（DCBrain → DCChains → 子链 → 工具）
渐进式展开：默认注入最近 N 字符，LLM 可通过工具按需扩展，类似屏幕翻页
LLM 层注入：展开逻辑在 BaseLLM.construct_prompt_context() 处理，不改动 Brain 与 Memory
多后端透明：存储在 UserMessage 字段上，InMemory / LocalStorage / PG 天然支持，无需建表
无需 Brain 封装：单独使用 Chain / Chains 时同样可以采集，不强依赖 DCBrain

数据模型¶

不引入新的 Schema 类，直接复用 `BaseMessage`¶

intermediate_msgs 在 Chain 内部本来就是 list[BaseMessage]，BaseMessage 已经具备角色区分、内容封装、序列化能力。因此不需要引入 IntermediateTrace、IntermediateMsgRecord 等新类。

树状元数据存入每条消息的 additional_kwargs，与 __chains_intermediate_results 的模式完全一致：

# Chain 在采集时向每条 BaseMessage 注入的树状元数据
# __trace_meta 的值是 JSON 字符串（str），符合 AttributeValue 约束
import json

msg.additional_kwargs = {
    "__trace_meta": json.dumps({
        "span_id": "abc123",           # 此次 Chain.run() 的 OTel span ID（全局唯一）
        "parent_span_id": "def456",    # 父 Chain 的 span_id（根节点为 None）
        "chain_name": "search_chain",  # 产生此消息的链名称
        "state": "succeeded",          # chain 完成时的状态
    }),
}

四个字段，无冗余： - span_id 全局唯一，天然区分同名链的不同调用（无需 call_index） - parent_span_id 足以重建完整树结构，深度（depth）在 trace_get_tree_summary() 内 post-hoc 计算，不写入消息 - create_ts（BaseMessage 自带）提供全局时序，无需额外计数器

UserMessage 新增字段¶

UserMessage（所有用户消息的基类）新增一个字段，将 trace 记录与视图配置合并为单一结构：

class TraceView(TFBaseModel):
    """控制 trace 注入行为的视图配置，由 ExpandContextTool 修改，注入层只读。"""
    mode: Literal["tail", "span", "all"] = "tail"
    chars: int = 2000                              # tail 模式：展开字符数（累加）
    span_ids: list[str] = Field(default_factory=list)  # span 模式：展开的 span_id 列表

class IntermediateTrace(TFBaseModel):
    """用户消息的执行过程记录，运行时采集与持久化复用同一对象。"""
    records: list[BaseMessage] = Field(default_factory=list)
    view: TraceView = Field(default_factory=TraceView)

class UserMessage(BaseMessage):
    ...
    intermediate_trace: IntermediateTrace = Field(default_factory=IntermediateTrace)  # 新增，非空

records 按时序累积所有 Chain 产生的中间消息，一条用户消息跨多次 Brain.run()（中断重试）时持续追加
view 由 ExpandContextTool 修改，控制下一次 LLM 调用时注入多少 trace 内容
两者合并为一个字段，UserMessage 只多一个字段，语义内聚
timestamp（BaseMessage.create_ts）+ __trace_meta.span_id 足以区分不同 run 的记录边界
非 Optional：default_factory=IntermediateTrace 保证每条 UserMessage 天然持有空 trace，任何 Chain/Brain/LLM 可无条件访问 current_input.intermediate_trace，无需判空
反序列化旧数据时 Pydantic 自动填入空 IntermediateTrace()，向后兼容

为何不放在 `additional_kwargs`¶

additional_kwargs 的类型定义为 dict[str, AttributeValue]，其中：

AttributeValue = Union[str, bool, int, float, Sequence[str], Sequence[bool], Sequence[int], Sequence[float]]

list[BaseMessage] 和嵌套 dict 均不是合法的 AttributeValue。现有的 __chains_intermediate_results 能放入是因为序列化为 JSON 字符串（str）。若对 trace 做同样的序列化，每次追加都要反序列化 → 追加 → 序列化，且无法保持可变引用，需要引入类似 __chains_intermediate_results 的手动同步协议——得不偿失。直接在 UserMessage 上加字段是最干净的方案。

工具函数（新文件 `tfrobot/brain/chain/intermediate_trace_utils.py`）¶

操作 list[BaseMessage] 的独立工具函数，供注入层和展开工具共用。

时序展开：复用现有 MsgExpander，不重复造轮子。length_function 使用 len——trace 注入是软预算，与 Memory 的 recall() 保持一致；用 tokenizer 会引入不必要的 LLM 类型耦合。

from tfrobot.utils.expander.msg_expander import MsgExpander
from tfrobot.utils.expander.base_expander import EChunk

def trace_expand_last_n_chars(
    trace: list[BaseMessage], n: int
) -> list[BaseMessage]:
    """时序展开：从末尾取总字符数不超过 n 的记录列表（保持时序）"""
    def _source() -> Iterator[EChunk[BaseMessage]]:
        for msg in reversed(trace):
            yield {"ele": msg, "direction": "backward"}

    expander = MsgExpander(
        chunk_size=n,
        length_function=len,
        expand_source=_source(),
        strict=False,  # 非严格模式，允许最后一条消息稍微超出
    )
    return expander.expand_eles()


def trace_get_span_records(
    trace: list[BaseMessage], span_id: str
) -> list[BaseMessage]:
    """因果展开：返回某个子链调用（span）的全部消息"""
    result = []
    for m in trace:
        if not isinstance(m.additional_kwargs, dict):
            continue
        raw = m.additional_kwargs.get("__trace_meta")
        if not isinstance(raw, str):
            continue
        meta = json.loads(raw)
        if meta.get("span_id") == span_id:
            result.append(m)
    return result


def trace_get_tree_summary(trace: list[BaseMessage]) -> str:
    """
    生成树摘要，用于告知 LLM 可展开的节点。
    depth 在此函数内通过 parent_span_id 链 post-hoc 计算，不依赖写入时的计数器。

    示例：
      [depth=0] search_chain (span=abc123, 12 msgs, 3.2k chars)
      [depth=1]   file_reader_chain (span=def456, 5 msgs, 8.1k chars)
      [depth=0] validate_chain (span=ghi789, 3 msgs, 0.8k chars)
    """
    # 1. 收集每个 span 的元信息
    # 2. 从 parent_span_id 链计算 depth：根节点 depth=0，逐层 +1
    # 3. 按 create_ts 排序后输出

采集层：完成时批量追加¶

核心设计原则¶

intermediate_msgs 与 intermediate_trace 在时间轴上不重叠：

数据	生命周期	职责
`intermediate_msgs`	当前 chain 运行期间（活跃）	LLM 当前上下文窗口，chain 完成后消费完毕
`intermediate_trace.records`	chain 完成后写入（快照）	已完成子链的持久化记录，供后续 LLM 按需注入

chain 执行期间只操作 intermediate_msgs，在 on_enter_succeeded / on_enter_failed / on_enter_aborted 时一次性批量追加到父 trace。这样无论注入层如何读取 trace，都不可能与当前活跃的 intermediate_msgs 产生重叠。

核心机制¶

trace 存储在 current_input.intermediate_trace（UserMessage 的字段）上。当 TFChainInterpreter 为每个子链构造新的 TextMessage 时，显式传入同一个 IntermediateTrace 引用，使所有子链写入同一个 records 列表。

# tfrobot/grammars/chain_interpreter.py — construct_chain_run_input() 修改
def construct_chain_run_input(self, node: FunctionCallNode) -> UserAndAssMsg:
    ...
    additional_info = cast(dict, self.additional_info).copy()
    additional_info.update(kwargs)
    user_msg = TextMessage(
        content=input_str,
        additional_kwargs=additional_info,
        intermediate_trace=self._trace,   # 显式传入同一个 IntermediateTrace 引用
        creator=self.user,
    )
    return user_msg, None

self._trace 初始化来自 DCChains 传入：

# tfrobot/grammars/chain_interpreter.py — __init__ 修改
def __init__(self, ..., intermediate_trace: Optional[IntermediateTrace] = None):
    ...
    self._trace = intermediate_trace  # 持有引用，不复制

# dc_chains.py — 创建 Interpreter 时传入
interpreter = TFChainInterpreter(
    ...,
    additional_info=current_input.additional_kwargs,
    intermediate_trace=current_input.intermediate_trace,  # 传入当前 IntermediateTrace 引用
)

时序图：

DCBrain.run()
  └── current_input.intermediate_trace = IntermediateTrace()  ← 首次 run 时初始化（Chain on_enter_init 兜底）

DCChains._run()
  └── TFChainInterpreter(intermediate_trace=current_input.intermediate_trace)
        ├── construct_chain_run_input(chain_A)
        │     └── TextMessage(intermediate_trace=self._trace)  ← 同一个 IntermediateTrace 引用
        │           └── chain_A 运行期间：intermediate_msgs 实时累积（trace 此时不写入）
        │           └── chain_A.on_enter_succeeded():
        │                 _batch_append_to_trace(chain_A.intermediate_msgs)
        │                 → trace.records += [taggedA1, taggedA2, ...]  ✓
        │
        └── construct_chain_run_input(chain_B)
              └── TextMessage(intermediate_trace=self._trace)  ← 同一个 IntermediateTrace，已含 chain_A 快照
                    └── chain_B 以全新 intermediate_msgs=[] 启动（不继承 chain_A 的过程消息）
                    └── chain_B.on_enter_succeeded():
                          _batch_append_to_trace(chain_B.intermediate_msgs)
                          → trace.records += [taggedB1, taggedB2, ...]  ✓

`current_input` 类型与 `isinstance` 守卫¶

current_input 保持 UserAndAssMsg 类型不做收窄。原因：Chain.run() 不仅接受用户消息，GraphIndex 通过 sql_element_to_message() 从 DB 恢复历史消息（包含 AssistantTextMessage），也会喂给 Chain.run() 做图谱抽取。

intermediate_trace 是 UserMessage 的非 Optional 字段（default_factory=IntermediateTrace），但 AssistantMessage 上不存在。因此所有访问 intermediate_trace 的入口点必须先检查：

# 所有 trace 相关入口的标准守卫模式：
if not isinstance(current_input, UserMessage):
    return  # 非用户消息不采集 / 不注入 trace

# 通过守卫后，直接访问，无需判空（default_factory 保证非空）：
current_input.intermediate_trace.records.append(...)
current_input.intermediate_trace.view.chars += 2000

需要此守卫的入口点： - _batch_append_to_trace()（Chain 采集层） - DCChains._run() 传入 interpreter 前 - BaseLLM.construct_prompt_context() 遍历 conversation 时

TextMessage（TFChainInterpreter 构造子链 input 时使用）是 UserMessage 的子类，天然通过守卫。

span 上下文跟踪¶

span ID 直接从 OTel 读取，不自行生成 UUID。_append_to_trace() 调用 get_current_span() 即可拿到正确的当前 span——因为 ChainMeta.__init_subclass__ 中 span_decorator 用 start_as_current_span() 包裹了每个 Chain 的 run() / async_run() 方法（context_schema.py L260/L373）。

这带来三个直接推论：

无需在 Chain 上新增任何 span 相关实例属性：span 生命周期完全由 decorator 管理（创建、设为 current、end()、detach 全部自动）
父子层级天然正确：子链 run() 调用时父链 span 已经是 current，TFTracer.start_span() 读到的父 span 就是父链的 span
exit 回调无需操作 span：span.end() 在 run() 退出时由 decorator 自动完成

Chain 上不需要向 additional_kwargs 写入任何 Intermediate Trace 专用状态。span_id 已经全局唯一，parent_span_id 已经足以重建树结构，无需额外计数器。

Chain 基类的接入点¶

修改 tfrobot/brain/chain/base.py：

位置	操作
所有状态回调	不改动
`run()` / `async_run()` 的 `finally` 块	`_batch_append_to_trace(state=self.state)` 批量写入

与 DCBrain 的 update_message_trace() 模式一致：单一退出点，所有状态路径自动覆盖，未来新增状态无需维护。self.state 在 finally 时已是最终状态（succeeded / failed / aborted）。

Chain 上零新增实例属性，additional_kwargs 零新增 key。

_batch_append_to_trace() — 完成时批量写入：

# chain/base.py — 在 on_enter_succeeded/failed/aborted() 末尾调用
from opentelemetry.trace import get_current_span

def _batch_append_to_trace(self, state: str) -> None:
    """将本次 chain 运行产生的所有 intermediate_msgs 批量写入 intermediate_trace。

    在 run()/async_run() finally 块调用，确保写入时机在 chain 完成之后，
    与当前活跃的 intermediate_msgs 不重叠，消除注入层二次展开的风险。
    """
    current_input = self._chain_context.current_input
    # isinstance 守卫：非用户消息（如 GraphIndex 喂入的 AssistantTextMessage）不采集 trace
    if not isinstance(current_input, UserMessage):
        return
    it = current_input.intermediate_trace  # 通过守卫后直接访问，default_factory 保证非空

    msgs = self._chain_context.intermediate_msgs
    if not msgs:
        return

    # span_decorator 用 start_as_current_span 包裹了 run()，此处 span 仍在活跃中
    span = get_current_span()
    span_ctx = span.get_span_context()
    span_id = format(span_ctx.span_id, "016x")
    parent_span = getattr(span, "parent", None)
    parent_span_id = (
        format(parent_span.span_id, "016x")
        if (parent_span is not None and parent_span.is_valid)
        else None
    )

    # 四字段元数据：span_id 全局唯一，depth 由 trace_get_tree_summary() post-hoc 计算
    meta = json.dumps({
        "span_id": span_id,
        "parent_span_id": parent_span_id,
        "chain_name": self.chain_name,
        "state": state,
    })

    for msg in msgs:
        tagged = msg.model_copy()   # 不修改原始消息对象
        if tagged.additional_kwargs is None:
            tagged.additional_kwargs = {}
        cast(dict, tagged.additional_kwargs)["__trace_meta"] = meta
        it.records.append(tagged)

同一个 chain 内所有消息共享相同的 __trace_meta（同一 span、同一 chain_call_index），由 create_ts 区分先后顺序。

持久化层¶

与 UserMessage 的时序¶

所有 Brain（BaseBrain、DCBrain 等）的提交时序完全一致：

run(): commit_message(current_input)          ← 此时 intermediate_trace.records 为空
run() 执行...（Chain 自动追加到 current_input.intermediate_trace.records）
run() finally:
    memory.update_message_trace(
        current_input.msg_id,
        current_input.intermediate_trace,
    )                                         ← 无需提取，字段本身就是结果
run(): commit_message(res_msg)

BaseBrain 与 DCBrain 的 run()/async_run() 均先 commit current_input（此时 trace 空），Chain 执行期间 trace 被 Chain 写满，最后统一在 finally 中 update 回 memory。

BaseMemory 接口扩展¶

新增两个方法，提供 no-op 默认实现，不破坏现有子类：

def update_message_trace(
    self, msg_id: str | int, trace: IntermediateTrace
) -> None:
    """将 trace 写回已存储的用户消息。默认 no-op，子类按需实现。"""
    pass

async def aupdate_message_trace(
    self, msg_id: str | int, trace: IntermediateTrace
) -> None:
    self.update_message_trace(msg_id, trace)

各 backend 实现：

Backend	实现方式
InMemory	在内存 dict 中找到消息对象，设置 `.intermediate_trace = trace`
LocalStorage	读文件 → 更新字段 → 写回
PG	`UPDATE SET intermediate_trace = $1 WHERE msg_id = $2`，字段类型 JSONB

Brain 层集成¶

BaseBrain 和 DCBrain 均在 run()/async_run() 的 finally 块中调用 memory.update_message_trace()。以 DCBrain 为例（BaseBrain 结构相同，复杂度更低）：

def run(self, current_input, tools=None):
    try:
        # ... 现有逻辑不变，Chain 会自动初始化并追加 intermediate_trace ...
        while self._should_continue():
            try:
                ...
            except TFUserInterruptError as e:
                self.abort(...)   # 中断时 finally 块仍会持久化当前 trace
            except TFBrainError as e:
                ...
    finally:
        # 无论何种退出路径，持久化 trace（字段本身即结果，无需提取）
        if current_input.intermediate_trace.records:
            self.memory.update_message_trace(
                current_input.msg_id,
                current_input.intermediate_trace,
            )
    ...

注入层：在 BaseLLM 统一处理¶

不修改 Brain 与 Memory 的业务逻辑，注入发生在 BaseLLM.construct_prompt_context() 内部。所有 LLM 子类共享此路径。

注入模型：每条消息就地展开¶

trace 不是"单独提取最近一条然后集中注入"，而是遍历整个对话窗口 [*conversation, current_input]，每条带有 intermediate_trace 的 UserMessage 都按自己的 view 配置展开，紧随其后就地注入。

LLM 收到的上下文结构（inject_trace=True 时）：

  [conversation[0]]           ← UserMessage: "第一轮提问"
  [trace records of msg[0]]   ← 第一轮执行记录（按 view 展开）
  [assistant[0]]
  [conversation[1]]           ← UserMessage: "第二轮提问"
  [trace records of msg[1]]   ← 第二轮执行记录（按 view 展开）
  [assistant[1]]
  ...
  [current_input]             ← 当前 UserMessage
  [trace records of current]  ← 本次 run 已完成子链的快照（按 view 展开）
  [intermediate_msgs]         ← 当前活跃 chain 的实时过程消息（原有逻辑不变）

inject_trace=False 时跳过所有 trace，退化为原始对话格式，开销为零。

BaseLLM 新增控制位¶

class BaseLLM(TFNeuralModel):
    ...
    inject_trace: bool = True
    """是否将 UserMessage.intermediate_trace 就地展开注入上下文。
    设为 False 时对话格式与原有行为完全一致，适用于不需要感知执行历史的专注型链。
    """

注入逻辑¶

BaseLLM.construct_prompt_context() 在处理 conversation 和 current_input 时，对每条带 trace 的 UserMessage 就地插入展开内容：

def _expand_trace_for_msg(
    msg: UserMessage,
) -> list[BaseMessage]:
    """为单条 UserMessage 按其 view 配置展开 trace，返回要就地插入的消息列表。"""
    it = msg.intermediate_trace  # 直接访问，UserMessage.intermediate_trace 非空
    if not it.records:
        return []
    view = it.view
    if view.mode == "tail":
        expanded = trace_expand_last_n_chars(it.records, view.chars)
    elif view.mode == "span":
        expanded = []
        for sid in view.span_ids:
            expanded.extend(trace_get_span_records(it.records, sid))
    else:  # "all"
        expanded = list(it.records)
    if not expanded:
        return []
    tree_summary = trace_get_tree_summary(it.records)
    header = BaseMessage(
        role="system",
        content=(
            f"【执行记录（展示 {sum(len(str(m.content)) for m in expanded)} 字符，"
            f"可调用 expand_context 展开更多）】\n{tree_summary}"
        ),
    )
    return [header, *expanded]

# construct_prompt_context() 内部：
if self.inject_trace:
    enriched_conversation = []
    for msg in (conversation or []):
        enriched_conversation.append(msg)
        enriched_conversation.extend(_expand_trace_for_msg(msg))
    conversation = enriched_conversation

    # current_input 的 trace（本次 run 已完成子链快照）也就地展开
    current_trace_msgs = _expand_trace_for_msg(current_input)
    if current_trace_msgs:
        intermediate_msgs = current_trace_msgs + list(intermediate_msgs or [])

为何此处是正确的注入点¶

层级	路径
ChatLLM	`format_to_request_msgs()` 处理 `prompt_ctx.conversation` + `intermediate_msgs`，trace 消息按角色转换为 API 消息列表
DeskLLM	`construct_request_params()` 里 `all_msgs = [conversation, current_input, intermediate_msgs]`，trace 随 conversation 拼入请求字符串

两者共享 BaseLLM.construct_prompt_context() 作为上游，注入逻辑写一次，全部 LLM 实现透明继承。

与 SeqChains / ParallelChains 的关系¶

SeqChains：移除 pass_intermediate_msgs 字段，各子链以全新 intermediate_msgs 启动。chain A 的过程消息在 A 完成后写入 trace，chain B 的 LLM 若 inject_trace=True 则能从 current_input.intermediate_trace 读到 chain A 的快照——这正是以前 pass_intermediate_msgs=True 想实现的效果，但粒度更细（按 view 控制而非全量传递）。若某个子链的 LLM 设置 inject_trace=False，则完全隔离，等价于原来的 pass_intermediate_msgs=False。

ParallelChains：各并行子链以 deep_copy(intermediate_msgs) 隔离，完全并发运行，互不干扰。所有子链完成后，由 ParallelChains 串行合并各子链的 intermediate_msgs 到父 trace，按 create_ts 排序以恢复全局时序：

# parallel_chains.py — 所有子链完成后的合并逻辑（示意）
all_msgs: list[BaseMessage] = []
for child_result in results:
    all_msgs.extend(child_result.chain_run_context.intermediate_msgs)
all_msgs.sort(key=lambda m: m.create_ts)
_batch_append_sorted(current_input.intermediate_trace, all_msgs, span_meta)

无竞态，无需锁，ParallelChains 自身的设计不变。

TraceView 协议¶

TraceView 是 IntermediateTrace.view 字段，存储在 current_input.intermediate_trace.view 上，只有 ExpandContextTool 可以修改，注入层只读。

数据结构¶

class TraceView(TFBaseModel):
    mode: Literal["tail", "span", "all"] = "tail"
    # tail 模式：展开字符数（累加，每次工具调用增加，不是绝对值）
    chars: int = 2000
    # span 模式：展开的 span_id 列表（追加，打开 A 再打开 B，A 不会关闭）
    span_ids: list[str] = Field(default_factory=list)

约定规范¶

唯一写入方：只有 ExpandContextTool 修改 view 字段
累加语义：chars 每次工具调用后增加，不是绝对值
追加语义：span_ids 只追加，打开 span A 再打开 span B，A 不会关闭
注入层只读：BaseLLM.construct_prompt_context() 读取后不修改

ExpandContextTool¶

additional_kwargs 传播链路¶

工具通过 merge_context=True 可访问并修改 current_input.additional_kwargs。完整调用链：

Chain.on_enter_doing()
  └── mini_drive.run(tool_call=..., intermediate=event_data.intermediate)
        └── BaseDrive.run(**kwargs)        # intermediate 在 kwargs 里
              └── tool.run(tool_params=..., **kwargs)
                    └── if merge_context:
                              tool_kwargs.update(kwargs)  # intermediate 注入
                    └── _run(**tool_kwargs)  # 可以拿到 intermediate

intermediate.context.current_input 是引用传递，工具修改 additional_kwargs["__trace_view"] 后，该 Chain 下一轮 on_enter_thinking() 重建 PromptContext 时立即生效。

传播边界（天然的隔离优势）¶

TFChainInterpreter 为每个子链构造新 TextMessage 时显式传入同一个 IntermediateTrace 引用。因此：

场景	行为	原因
同一 Chain 内（下一轮 thinking）	立即可见	`current_input` 引用传递，`view` 修改即时生效
同一 DCChains 内的下一个子链	不可见（隔离）	每个子链拿到独立的 `TextMessage`，`intermediate_trace` 字段通过构造传入，但 `view` 是共享对象内的字段——注意：这意味着子链 A 修改 `view` 会影响子链 B
跨 DCChains 轮次	可见	同一 `IntermediateTrace` 对象贯穿整个 `run()`

子链间 view 共享

同一 TFChainInterpreter 内所有子链共享同一个 IntermediateTrace 对象，ExpandContextTool 对 view 的修改会影响后续所有子链的注入行为。这通常是期望行为（一次展开，后续子链都受益），但需要在使用文档中说明。

实现¶

class ExpandContextTool(BaseTool):
    """渐进式展开历史执行过程的中间消息。支持时序翻页和因果定向两种模式。"""
    name: ClassVar[str] = "expand_context"
    merge_context: ClassVar[bool] = True

    def _run(
        self,
        mode: Literal["tail", "span"] = "tail",
        chars: int = 2000,                # tail 模式：本次新增展开量
        span_ids: list[str] = [],         # span 模式：追加展开的 span_id 列表
        intermediate: Optional[IntermediateResult] = None,
        **kwargs: Any,
    ) -> ToolReturn:
        if intermediate is None:
            return ToolReturn(content="无法获取执行上下文。")

        current_input: UserMessage = intermediate.context.current_input
        it = current_input.intermediate_trace  # 直接访问，非空
        if not it.records:
            return ToolReturn(content="当前消息无执行记录。")

        view = it.view
        if mode == "tail":
            view.mode = "tail"
            view.chars += chars                                   # 累加
        elif mode == "span":
            view.mode = "span"
            view.span_ids = list(set(view.span_ids) | set(span_ids))  # 追加不替换

        return ToolReturn(content="展开配置已更新，请继续阅读上下文。")

分阶段实施建议¶

阶段	内容	产出
Phase 1	`UserMessage.intermediate_trace` 字段 + `TFChainInterpreter` 显式传递 + Chain `on_enter_succeeded/failed/aborted` 批量追加 + SeqChains 移除 `pass_intermediate_msgs` + ParallelChains 合并逻辑	运行时可采集 trace，存内存，暂不持久化
Phase 2	`memory.update_message_trace()` + DCBrain `finally` 块持久化	trace 随 UserMessage 持久化，中断恢复问题解决
Phase 3	`BaseLLM.inject_trace` + `construct_prompt_context()` 每消息就地展开注入 + 工具函数	LLM 全量感知历史执行记录，SeqChains/DCChains 子链间信息流通
Phase 4	`ExpandContextTool` + 注册到 DCBrain 默认工具集	LLM 可主动按需展开

变更清单¶

文件	类型	改动概要
`tfrobot/brain/chain/intermediate_trace_utils.py`	新建	`trace_expand_last_n_chars`（基于 `MsgExpander`）/ `trace_get_span_records` / `trace_get_tree_summary`
`tfrobot/drive/tool/expand_context_tool.py`	新建	`ExpandContextTool`
`tfrobot/schema/message/conversation/message_dto.py`	修改	`UserMessage` 新增 `intermediate_trace: IntermediateTrace = Field(default_factory=IntermediateTrace)`（非 Optional）；新增 `TraceView`、`IntermediateTrace` 两个 Pydantic 模型
`tfrobot/brain/chain/base.py`	修改	新增 `_batch_append_to_trace()` 方法（含 `isinstance(current_input, UserMessage)` 守卫）；`run()`/`async_run()` `finally` 块调用之；`current_input` 类型保持 `UserAndAssMsg` 不变
`tfrobot/grammars/chain_interpreter.py`	修改	`__init__` 新增 `intermediate_trace` 参数；`construct_chain_run_input()` 显式传入 `intermediate_trace=self._trace`
`tfrobot/brain/chain/chain_structures/dc_chains.py`	修改	创建 `TFChainInterpreter` 时传入 `intermediate_trace=current_input.intermediate_trace`
`tfrobot/brain/chain/chain_structures/sequence_chains.py`	修改	移除 `pass_intermediate_msgs` 字段；各子链以全新 `intermediate_msgs=[]` 启动
`tfrobot/brain/chain/chain_structures/parallel_chains.py`	修改	所有子链完成后按 `create_ts` 合并 `intermediate_msgs` 到父 trace
`tfrobot/brain/chain/llms/base.py`	修改	新增 `inject_trace: bool = True`；`construct_prompt_context()` 新增每消息就地展开注入逻辑
`tfrobot/brain/memory/base.py`	修改	`update_message_trace` / `aupdate_message_trace`（no-op 默认实现）
`tfrobot/brain/base.py`	修改	`run()`/`async_run()` 的 `finally` 块调用 `memory.update_message_trace()`
`tfrobot/brain/dc_brain.py`	修改	`run()`/`async_run()` 的 `finally` 块调用 `memory.update_message_trace()`