消息系统（Message System）¶

TFRobot 的消息系统是整个 Agent 框架的「通用血液」：用户输入、LLM 回复、工具返回、DSL 执行结果、跨轮次记忆，全部以统一的 BaseMessage 结构在 Brain / Chain / Memory / Drive 之间流转。本文档阐述消息系统的设计背景、分层结构、Pydantic 多态机制、中间消息在上下文中的生产-消费-持久化视图，以及标准格式到各 LLM Provider API 的转换层。

设计理念¶

为什么需要「消息系统」而不是直接用 LLM 的原生格式¶

早期 LLM 应用大多直接使用 OpenAI 的 {"role": "user", "content": "..."} 作为通用格式。这种做法在简单对话里够用，但在 TFRobot 这样的 Agent 框架里会迅速失控：

多 Provider 差异：OpenAI 允许 system role，Anthropic 把 system 抽出为独立参数；OpenAI 把 tool_calls 放在 assistant 消息的字段上，Anthropic 把它放进 content block；Zhipu / 字节 / Gemini 又各有差异。若让业务层感知这些差异，Agent 逻辑会被 Provider 细节淹没。
多模态原生支持：用户输入可能是图片、音频、视频、PDF、多段混合。不同 Provider 对多模态的支持能力和降级策略不一（不支持视频时要落回 detail 文本说明），需要一套中间表达承载这些语义。
TFRobot 特色结构：框架有两类 Provider 没有的消息形态
- ConsoleResult：DSL 解释器（TFL grammar）执行后的「控制台回显」，既不是纯工具返回，也不是助理文本，它可能携带附件、摘要、trace 元信息
- TFL / Grammar 脚本：Agent 在推理中产生的 DSL 代码片段，它是 assistant 的产物，但要被后续解释器消费
跨轮次 Trace 持久化：Agent 不只记录「你问我答」，还要在数据库里保存一次推理内部的完整执行轨迹（IntermediateTrace），这些数据需要在下一轮被还原成原始 Python 类型才能继续推理。

因此 TFRobot 定义自己的 BaseMessage 族，作为「项目内部标准格式」；LLM 适配层再将它翻译成各 Provider 的 API 格式。业务层只关心标准格式，Provider 差异锁在适配层内部。

核心设计原则¶

原则	含义
单一数据源	一条消息从产生到持久化，始终是同一个 Pydantic 对象，不会被拷贝成多种 DTO
类型保持优先	反序列化后，`AssistantTextMessage` 必须还是 `AssistantTextMessage`，不能降级成 `BaseMessage(role="assistant")`
转换集中化	「项目格式 → LLM API 格式」只发生在 `ChatLLM.format_to_request_msgs` 及其 Provider Hook 里，业务层一律用 `BaseMessage`
Provider 可插拔	新增 LLM Provider 时，业务层不需要任何改动，只新增适配器

分层架构¶

┌─────────────────────────────────────────────────────────────────┐
│                  业务层 (Brain / Chain / Memory)                 │
│  只感知 BaseMessage 及其子类，通过 to_llm_request() 触发转换     │
└─────────────────────────────────────────────────────────────────┘
                              ↕
┌─────────────────────────────────────────────────────────────────┐
│             标准消息层 (tfrobot/schema/message/)                 │
│  BaseMessage ─┬─ UserMessage ──┬─ TextMessage / ImageMessage /  │
│               │                └─ AudioMessage / MultiPartMessage/…│
│               ├─ AssistantMessage ── AssistantTextMessage        │
│               └─ (tool / console / system ← 走 BaseMessage 兜底) │
│                                                                  │
│  Message: TypeAlias = Annotated[Union[子类...], Discriminator]   │
└─────────────────────────────────────────────────────────────────┘
                              ↕ to_llm_request()
┌─────────────────────────────────────────────────────────────────┐
│           LLM 中间格式层 (llm_requests.py)                       │
│  LLMUserMessage / LLMAssistantMessage / LLMSystemMessage /       │
│  LLMToolMessage / LLMFunctionMessage / LLMTFLMessage            │
│  （Provider 无关，但更贴近 LLM API 结构）                         │
└─────────────────────────────────────────────────────────────────┘
                              ↕ reformat_request_msg_to_api()
┌─────────────────────────────────────────────────────────────────┐
│       Provider 适配层 (brain/chain/llms/<provider>.py)           │
│  Anthropic / OpenAI / Zhipu / Doubao / Gemini …                 │
│  把 LLMMessage → Provider SDK 要求的 dict / 对象                 │
└─────────────────────────────────────────────────────────────────┘

BaseMessage 及其子类¶

BaseMessage 字段图谱¶

tfrobot/schema/message/base.py：

字段	类型	用途
`role`	`Literal["system","user","assistant","function","tool","console"]`	角色标签，控制转换逻辑分支
`content`	`Union[str, Path, AnyUrl, BaseUser, list[BaseMessage], list[MsgPart]]`	多态内容，子类通过泛型 `StrSubclass` 收窄
`tool_calls`	`Optional[list[ToolCall]]`	LLM 返回的工具调用列表（消息级字段，非 content 内）
`tool_reses`	`Optional[Sequence[str \\| ToolReturn]]`	工具返回值
`function_call` / `function_res`	兼容 OpenAI 旧版 function calling	已逐步被 `tool_calls` 取代
`console_reses`	`Optional[Sequence[str \\| ConsoleResult]]`	TFRobot 特有：DSL 解释器输出（含 result / summary_result / attachments）
`tfl_scripts`	`Optional[list[str]]`	assistant 产出的 TFL 脚本代码
`additional_kwargs`	`dict[str, AttributeValue]`	元数据自由字段（msg_meta、强调标记等）
`create_ts`	`int`	毫秒级时间戳，`IntermediateTrace` 展开时的排序依据
`creator` / `group`	`BaseUser` / `BaseGroup`	消息归属
`msg_id` / `conversation_id`	`str`	唯一标识

子类分布¶

tfrobot/schema/message/conversation/message_dto.py 按 msg_type 精细区分：

UserMessage 家族（9 类）：TextMessage / AttachmentMessage / AudioMessage / ContactMessage / ImageMessage / VideoMessage / UrlMessage / HistoryMessage / MultiPartMessage。携带 UserMessage.intermediate_trace: IntermediateTrace，是跨轮次 trace 持久化的载体。
AssistantMessage 家族：目前仅 AssistantTextMessage（含 content + reasoning_content，后者支持 Anthropic/DeepSeek 思维链结构）。
System / Tool / Console / Function：没有独立子类，直接用 BaseMessage(role=...) 表达，通过 role 和字段存在性（tool_calls / console_reses / function_res）在运行期分支处理。

为什么 tool/console/system 没有独立子类¶

这是一个历史遗留的设计决策，不是最佳设计：

早期优先级：UserMessage 和 AssistantMessage 的内容结构对 LLM 上下文构建影响最大（多模态 / ImageMessage 等），优先做了精确类型收敛
Tool/Console 的结构差异小：一条 tool 消息主要由 tool_reses[n].call_id + result 决定，字段差异不足以单独立类
代价：在反序列化时，tool/console 消息无法恢复子类信息——但因为它们本来就没有子类，这个「代价」被掩盖了

真正的隐患出现在 user/assistant 子类落入 list[BaseMessage] 这样的宽类型容器里：Pydantic 失去判别联合的上下文，反序列化时一律退回基类。这就是 TFROB-299 的成因。

多态反序列化：Callable Discriminator¶

方案与实现¶

message_dto.py 中的 get_discriminator_value 函数与 Message / UserAndAssMsg TypeAlias 定义了两级判别联合：

def get_discriminator_value(v: Any) -> str:
    """复合判别键：role:msg_type，缺 msg_type 的走 DEFAULT_TAG"""
    if isinstance(v, dict):
        role = v.get("role")
        if "msg_type" not in v:
            return DEFAULT_TAG
        msg_type = v.get("msg_type")
    else:
        role = getattr(v, "role")
        if not hasattr(v, "msg_type"):
            return DEFAULT_TAG
        msg_type = getattr(v, "msg_type")
    return f"{role}:{msg_type}" if role in ("user", "assistant") else DEFAULT_TAG


Message: TypeAlias = Annotated[
    Annotated[TextMessage, Tag("user:text")]
    | Annotated[ImageMessage, Tag("user:image")]
    | ...
    | Annotated[AssistantTextMessage, Tag("assistant:text")]
    | Annotated[BaseMessage, Tag(DEFAULT_TAG)],
    Discriminator(get_discriminator_value),
]

为什么选 Callable Discriminator 而不是 Literal Discriminator¶

Pydantic v2 支持两种判别联合：

Literal Discriminator：Field(discriminator="msg_type") — 直接按单字段字面量分派。简单，但只能用单字段。
Callable Discriminator：Discriminator(callable) — 用函数计算判别键。灵活，支持复合逻辑。

TFRobot 选 Callable 有三个原因：

复合键需求：同一个 msg_type="text" 既可能是 TextMessage（user）也可能是 AssistantTextMessage（assistant），必须用 role:msg_type 复合键区分
DEFAULT_TAG 兜底：system/tool/console 等没有 msg_type 的消息需要退化到 BaseMessage——Literal 风格无法表达「无字段时走兜底」
向后兼容：历史数据里存在 msg_type 字段缺失的旧记录，兜底分支保证老 JSON 也能反序列化

业界最佳实践对比¶

方案	适用场景	在 TFRobot 的可行性
Literal + 单字段 discriminator	清晰的封闭枚举，无复合键	❌ 无法表达 `role:msg_type`
Callable Discriminator（当前）	复合键或运行期计算	✅ 当前使用
RootModel + 自定义 validator	单字段包装器、非 Union	❌ 不适合多态 Union
在基类写 `model_validator(mode="before")` 做类型分派	灵活但脱离 Pydantic 原生多态机制	⚠️ 性能差、IDE 类型推断丢失
外部序列化元信息（保存 `__class__`）	极端场景	❌ 耦合实现细节、跨语言不友好

结论：Callable Discriminator 是当前场景的最佳实践。但它有一个边界陷阱——只有当 Pydantic 看到 Message 联合类型注解时才触发；若字段注解是宽的 list[BaseMessage]，判别联合不会被激活，子类信息全部丢失。

常见陷阱：list[BaseMessage] vs list[Message]¶

# ❌ 反序列化后所有 user/assistant 子类降级为 BaseMessage
records: list[BaseMessage] = Field(default_factory=list)

# ✅ 反序列化时 Pydantic 激活 Message 判别联合，保留子类
records: list[Message] = Field(default_factory=list)

TFROB-299 就是这一陷阱的直接后果——IntermediateTrace.records 当时被写成了 list[BaseMessage]，导致 trace 持久化-加载后 AssistantTextMessage 降级为 BaseMessage(role="assistant")，调用 to_llm_request() 时命中 case "assistant": raise ValueError。

中间消息的生产-消费-持久化¶

生产者视图¶

产生位置	产生什么	写入哪里
`Chain.on_enter_thinking`（LLM 返回）	`AssistantTextMessage`（含 reasoning_content、tool_calls）	`ChainContext.intermediate_msgs`
`Chain.on_enter_doing`（工具执行返回）	`BaseMessage(role="tool", tool_reses=[...])`	`ChainContext.intermediate_msgs`
`DCBrain.plan` / `DCChains.plan_validate`（DSL 解释器）	`BaseMessage(role="console", console_reses=[ConsoleResult])`	`ChainContext.intermediate_msgs` 或直接进 trace
`BaseChain._batch_append_to_trace`（采集层）	将上述 intermediate_msgs 拷贝 + 标记 `__trace_meta`	`UserMessage.intermediate_trace.records`（事故字段）

持久化¶

UserMessage 作为一个整体被 sync_msg（tfrobot/brain/memory/）通过 .model_dump_json() 写入 PostgreSQL。其 intermediate_trace.records 作为嵌套字段一并序列化，每条记录的 role、msg_type、content 等字段都进入 JSON。

消费者视图¶

下一轮对话：

Brain.run(user_input_2)
  │
  ├─ memory.recall()  → 从 DB 加载历史 UserMessage
  │    └─ TextMessage.model_validate_json(...)   ← 触发反序列化
  │         └─ intermediate_trace.records 按 list[Message] 解析
  │              ↑ 若类型注解是 list[BaseMessage]，这一步就丢子类
  │
  ├─ BaseLLM.construct_prompt_context(user_input_2, conversation=[recalled])
  │    └─ _expand_trace_for_msg(msg)
  │         └─ 按 create_ts 将 records 分 pre/post，注入 Conversation.msgs
  │
  └─ ChatLLM.construct_request_params(prompt_ctx)
       └─ format_to_request_msgs
            └─ for msg in prompt_ctx.conversation.msgs:
                 msg.to_llm_request()   ← TFROB-299 爆点

类型风险面清单¶

所有可能承载混合消息类型的容器字段，只要用了宽类型 BaseMessage，都是类型丢失的风险点。以下是 TFROB-299 修复前后的对照：

字段	定义文件	修复前	修复后	风险说明
`IntermediateTrace.records`	`message_dto.py`	`list[BaseMessage]`	`list[Message]`	🔴 TFROB-299 直接原因
`Conversation.msgs`	`context_vars.py`	`list[BaseMessage]`	`list[Message]`	🔴 LLM 请求上游
`IntermediateTraceSnapshot.records`	`context_vars.py`	`tuple[BaseMessage, ...]`	`tuple[Message, ...]`	🟠 Prompt 模板渲染
`PromptContext.intermediate_msgs`	`context_vars.py`	`list[BaseMessage] \\| None`	`list[Message] \\| None`	🟠 链内传递
`ChainContext.intermediate_msgs`	`chain_schema.py`	`list[BaseMessage]`	`list[Message]`	🟠 采集源头（运行期写入，不经 JSON，但改类型统一语义）

LLM API 转换层¶

to_llm_request() 契约¶

BaseMessage.to_llm_request() 方法（base.py）是从项目标准格式进入 LLM 适配层的总入口。它的实现采用 match self.role 分派：

match self.role:
    case "assistant":
        raise ValueError("助理消息需要子类实现转换过程")  # TFROB-299 爆点
    case "user":
        raise ValueError("用户消息需要子类实现转换过程")
    case "system":
        return LLMSystemMessage(content=...)
    case "console":
        # 展开 ConsoleResult → LLMTFLMessage + 附件 ImageMessage
        return list[BaseLLMMessage]
    case "tool":
        # 展开 ToolReturn → LLMToolMessage + 附件
        return list[BaseLLMMessage]
    case "function":
        return LLMFunctionMessage(...)

为什么 user / assistant 必须抛异常强制子类实现？

User/Assistant 消息的 content 是最丰富的：可能是 str、Path、list[MsgPart]（多模态）、list[BaseMessage]（历史消息嵌套）。BaseMessage 层无法知道具体子类应当如何展开
Assistant 还有 tool_calls 这个消息级字段（不在 content 内），如果走默认的 to_dict() → LLMAssistantMessage(content=str(...)) 会静默丢失 tool_calls
强制子类覆写是一种「契约保护」：宁可抛异常，也不要产生看起来能工作但丢信息的消息
Tool/Console/System 走默认分支是因为它们没有子类，结构单一，基类可以正确处理

TFROB-299 暴露的裂缝：如果上游的 Conversation.msgs 允许反序列化出纯 BaseMessage(role="assistant")，这条「契约保护」就变成了「生产崩溃」。修复的根本方式是堵住上游的类型丢失，而不是在 case 里做 best-effort 兜底——兜底会再次静默丢信息，违背契约设计的初衷。

多 Provider 差异对照¶

维度	Anthropic	OpenAI	TFRobot 标准
system role	❌ 禁用：抽出到 `system` 请求参数；若出现在 messages 里则转为 user text block	✅ 允许 role="system"	`LLMSystemMessage`
tool_calls	content block `tool_use` / `tool_result`	assistant message 的 `tool_calls` 字段	`LLMToolCall` on `LLMAssistantMessage.tool_calls`
console role	→ user text block（降级）	→ user（role 重写）	`LLMTFLMessage(role="console")`
图片	content block `image` + media_type	content parts `image_url`	`LLMUserMessage.content = [ImageUserMsgPart(...)]`
视频/音频	⚠️ 部分支持，不支持时走 `detail` 文本降级	⚠️ 同左	`Video/AudioUserMsgPart`
reasoning_content	原生 thinking block	部分模型支持	`LLMAssistantMessage.reasoning_content`

Provider 适配器通过三个 Hook 接管差异：

_configure_api_params：调整请求参数（如 Anthropic 把 system 抽出来）
reformat_request_msg_to_api：LLMMessage → Provider SDK dict
_assemble_api_request：组装最终请求（含工具定义等）

七段式请求组装（`ChatLLM.construct_request_params` / `format_to_request_msgs`）¶

construct_request_params
  ├─ Phase 1: construct_prompt_context   ← inject_trace、conversation 构建
  ├─ Phase 2: _configure_api_params       ← Provider Hook（如提取 system）
  ├─ Phase 3: format_to_request_msgs      ← msg.to_llm_request() 批量转换
  ├─ Phase 4: req_msgs.to_list            ← 展平 & 归并
  ├─ Phase 5: reformat_request_msg_to_api ← Provider Hook（→ SDK dict）
  └─ Phase 6: _assemble_api_request       ← 拼装工具 / 参数

配置与扩展¶

新增一个 UserMessage 子类¶

在 message_dto.py 定义子类，设置 msg_type: Literal[UserMessageType.xxx]
覆写 to_llm_request()
在 Message 和 UserAndAssMsg TypeAlias 里添加 Annotated[NewMessage, Tag("user:xxx")] 分支（容易遗漏）
补充单元测试：TextMessage.model_validate_json(msg.model_dump_json()) 要能还原类型

新增一个 LLM Provider¶

继承 ChatLLM 实现新的子类
覆写 _configure_api_params / reformat_request_msg_to_api / _assemble_api_request
不需要改业务层——消息系统是完全 Provider 无关的

使用建议¶

声明消息容器字段时的原则¶

永远不要写 list[BaseMessage] 或 tuple[BaseMessage, ...]，除非你确信这个字段永远只承载 tool/console/system 这类无子类的消息
跨 JSON 边界（DB / 消息队列 / RPC）的字段，一律用 list[Message]——让 Pydantic 的判别联合接管反序列化
如果字段只在进程内传递（没有 model_dump_json），类型丢失风险较低，但为了统一语义，也建议用 list[Message]

写测试时的原则¶

涉及消息序列化的测试，必须验证 isinstance(restored, ExpectedSubclass)，不能只断言字段值相等
TestSerializationRoundtrip 类应当覆盖所有涉及嵌套消息的容器字段
model_dump_json → model_validate_json 是最严格的反序列化验证路径，比 model_dump → model_validate 更能暴露类型丢失

与其他模块的协作¶

模块	如何使用消息系统
Brain	`BrainContext.conversation` 是历史对话，`UserMessage.intermediate_trace` 记录每一轮推理的完整 trace
Chain	`ChainContext.intermediate_msgs` 采集 thinking/doing 阶段的产物；run 结束时 `_batch_append_to_trace` 持久化
Memory	`sync_msg` 将整个 `UserMessage`（含 trace）序列化入库；`recall` 反向还原
Drive/Tool	工具执行返回 `ToolReturn`，被包成 `BaseMessage(role="tool")` 流入 Chain
Grammars/DSL	DSL 解释器产出 `ConsoleResult`，被包成 `BaseMessage(role="console")`
SaverLoader	提供通用的 Pydantic 持久化协议，消息系统通过它落盘

历史教训：TFROB-299¶

现象：多轮对话第二条消息处理时抛 ValueError: 助理消息需要子类实现转换过程

根因：IntermediateTrace.records 声明为 list[BaseMessage]，Pydantic 反序列化时没有走 Message 判别联合，AssistantTextMessage 被降级为 BaseMessage(role="assistant")，进入 LLM 请求构建链路后命中契约保护抛异常

教训：Pydantic 判别联合必须在「能看到 Union 注解」的地方才生效；任何写成 list[BaseMessage] 的跨 JSON 边界字段都是类型丢失的定时炸弹。

关键文件一览¶

目的	路径
BaseMessage 基类与 to_llm_request 契约	`tfrobot/schema/message/base.py`
子类定义 + Message TypeAlias	`tfrobot/schema/message/conversation/message_dto.py`
多模态 Part	`tfrobot/schema/message/msg_part.py`
LLM 中间格式	`tfrobot/schema/message/llms/llm_requests.py` / `llm_response.py`
中间消息流转	`tfrobot/brain/chain/base.py`（`_batch_append_to_trace`）、`tfrobot/brain/chain/llms/base.py`（`_expand_trace_for_msg`）
Conversation / PromptContext	`tfrobot/schema/brain/chain/prompt/context_vars.py`
Provider 适配器	`tfrobot/brain/chain/llms/{chat_llm,claude_desk,openai,zhipuai,...}.py`