Agent 的上下文继承：子 Agent 如何获取和传递父 Agent 的中间推理结果

一、深度引言与场景痛点

多 Agent 协作的场景里，最常见的浪费是：父 Agent 辛辛苦苦分析了半天，得出结论"用户需要查三季度的销售数据"；然后把任务交给子 Agent 时，只传了"查销售数据"五个字。子 Agent 又从头推理一遍，才搞清楚原来是要查三季度的。

这就是上下文断裂。多 Agent 架构里，每个 Agent 的任务拆解、中间推理、工具调用结果都值得被下游复用。如果传递的只有最终结论，下游就得重新推理。上下游重复推理不仅浪费 Token，还可能因为推理路径不同导致结论不一致。

上下文继承要解决的问题就是：父 Agent 的推理不白做，子 Agent 能直接站在父 Agent 的肩膀上干活。

二、底层机制与原理深度剖析

上下文继承不是"把所有信息一股脑传给子 Agent"。信息过多会让子 Agent 迷失，Token 成本也扛不住。需要有一个上下文提取器和压缩器，把父 Agent 的推理挑选出四类信息：关键结论（如"用户要的是三季度的数据"）、已验证事实（如"数据库里 A 表有 230 万条记录"）、失败路径（如"尝试了用 B 字段查询但无结果"）、工具摘要（如"文件搜索返回了 15 篇文档，其中 3 篇相关"）。

子 Agent 接收到的上下文，不是原始文本，而是结构化的 JSON 对象，包含以上四类信息。这样子 Agent 不用再提取一遍，直接引用就行。

三、生产级代码实现

from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Optional import json @dataclass class ReasoningStep: step_id: str conclusion: str confidence: float # 0.0 - 1.0 evidence: list[str] = field(default_factory=list) @dataclass class AgentContext: """多 Agent 之间传递的上下文对象""" task_id: str parent_goal: str key_conclusions: list[ReasoningStep] = field(default_factory=list) verified_facts: list[str] = field(default_factory=list) failed_attempts: list[str] = field(default_factory=list) tool_results: dict[str, Any] = field(default_factory=dict) constraints: list[str] = field(default_factory=list) child_results: dict[str, Any] = field(default_factory=dict) def to_prompt_context(self, max_tokens: int = 2000) -> str: """将上下文压缩为 prompt 可用格式""" parts = [f"任务目标: {self.parent_goal}"] if self.key_conclusions: high_conf = [c for c in self.key_conclusions if c.confidence >= 0.7] if high_conf: parts.append("高置信度结论:") for c in high_conf: parts.append(f"- {c.conclusion} (置信度: {c.confidence:.0%})") if self.verified_facts: facts = "\n".join(f"- {f}" for f in self.verified_facts[:5]) parts.append(f"已验证事实:\n{facts}") if self.failed_attempts: fails = "\n".join(f"- {f}" for f in self.failed_attempts[:3]) parts.append(f"已排除路径:\n{fails}") if self.constraints: cons = "\n".join(f"- {c}" for c in self.constraints) parts.append(f"约束条件:\n{cons}") result = "\n\n".join(parts) if len(result) > max_tokens * 4: # 粗略估算 result = result[: max_tokens * 4] return result def merge_child_result(self, child_name: str, result: Any) -> None: self.child_results[child_name] = result def add_conclusion(self, step: ReasoningStep) -> None: # 避免重复结论 existing = [c for c in self.key_conclusions if c.step_id == step.step_id] if not existing: self.key_conclusions.append(step) class ContextOrchestrator: """上下文编排器：管理多 Agent 之间的上下文传递""" def __init__(self, store=None): self._contexts: dict[str, AgentContext] = {} self._store = store def create_context(self, task_id: str, goal: str) -> AgentContext: ctx = AgentContext(task_id=task_id, parent_goal=goal) self._contexts[task_id] = ctx return ctx def get_context(self, task_id: str) -> Optional[AgentContext]: return self._contexts.get(task_id) async def delegate_to_child( self, parent_task_id: str, child_name: str, child_task: str, child_agent_fn, ) -> Any: """将任务委派给子 Agent，同时传递过滤后的上下文""" parent_ctx = self._contexts.get(parent_task_id) if not parent_ctx: raise ValueError(f"未找到父任务上下文: {parent_task_id}") # 为子 Agent 提取相关上下文 child_ctx = await self._filter_context(parent_ctx, child_task) # 注入上下文到子 Agent 的 prompt context_prompt = child_ctx.to_prompt_context() try: result = await child_agent_fn(child_task, context=context_prompt) except Exception as e: parent_ctx.failed_attempts.append(f"[{child_name}] {child_task} → {e}") raise parent_ctx.merge_child_result(child_name, result) return result async def _filter_context( self, ctx: AgentContext, sub_task: str ) -> AgentContext: """根据子任务的关键词过滤上下文，只传递相关信息""" keywords = set(sub_task.lower().split()) filtered = AgentContext( task_id=f"{ctx.task_id}_sub", parent_goal=sub_task, constraints=list(ctx.constraints), ) for fact in ctx.verified_facts: if any(kw in fact.lower() for kw in keywords): filtered.verified_facts.append(fact) for conclusion in ctx.key_conclusions: if any(kw in conclusion.conclusion.lower() for kw in keywords): filtered.key_conclusions.append(conclusion) for failure in ctx.failed_attempts: if any(kw in failure.lower() for kw in keywords): filtered.failed_attempts.append(failure) return filtered

AgentContext.to_prompt_context把上下文压缩为 prompt 可用格式，核心是"按置信度分层"。高于 70% 置信度的结论直接给子 Agent 当事实用，低于 70% 的只作为参考。这样做的好处是：子 Agent 可以跳过已验证的推理步骤，只做增量推理。

ContextOrchestrator.delegate_to_child是父 Agent 委派任务给子 Agent 的统一入口。它先根据子任务的关键词过滤上下文，只把相关的结论、事实、失败路径传递给子 Agent。不相关的信息不传递，减少干扰。

_filter_context的过滤逻辑很简单：关键词匹配。这篇文章不追求复杂的语义匹配，因为在 Agent 场景里，任务是明确的（如"查销售数据"），关键词匹配已经够用。如果任务语义更模糊，可以换成 embedding 相似度过滤。

四、边界分析与架构权衡

上下文会随着子 Agent 数量增加而膨胀。5 个子 Agent 的结论拼在一起，可能超过 LLM 的 context window。解决方案是设置上下文的大小上限，超过限制时触发压缩——用 LLM 对之前的结论做摘要，减少 Token 数量。

但这带来了新问题：压缩可能丢失关键信息。所以摘要压缩器和结论检查应该配对使用。摘要压缩后，检查被丢弃的结论里有没有包含数值或实体名。如果包含，这些信息应作为"关键事实"单独保留，不能被摘要吞掉。

另一个隐藏成本是上下文的一致性。多个子 Agent 并行执行时，如果都往同一个上下文里写结果，可能出现写冲突。解决方式是用不可变结构或 Copy-on-Write：每个子 Agent 拿到的是父上下文的只读快照，结果通过merge_child_result统一合并。

五、总结

Agent 上下文继承避免了多 Agent 协作中的重复推理。核心机制是：提取父 Agent 的四类信息（结论、事实、失败、工具结果），按关键词过滤后传给子 Agent，子 Agent 基于这些上下文做增量推理而非从零开始。

落地重点：上下文结构化为 JSON 而非自由文本；传递前按子任务关键词过滤；设置大小上限并在超限时触发摘要压缩；并行子 Agent 用快照 + 合并模式避免写冲突。