AI 数据分析 Agent 协作：模型之间怎么传上下文不丢信息

一、Agent 协作：多个模型的"接力赛"

单个 AI 模型能做很多事，但复杂的数据分析任务往往需要多个模型分工合作——一个擅长写 SQL，一个擅长统计分析，一个擅长可视化，一个擅长写报告。就像接力赛：每个运动员跑自己擅长的段落，交接棒时不能掉，否则整场比赛就废了。

Agent 之间的"交接棒"就是上下文传递。模型 A 的分析结果要完整准确地传给模型 B，B 基于这些结果继续分析，再传给模型 C。如果交接棒掉了——关键信息丢失、数据被误读、结论被扭曲——后续模型的分析就是空中楼阁。

上下文丢失的典型场景：

场景	丢失什么	后果
SQL Agent → 分析 Agent	查询的过滤条件和聚合逻辑	分析 Agent 不知道数据范围，结论偏差
分析 Agent → 报告 Agent	统计检验的置信区间和效应大小	报告只写"显著"，不写"效应多大"
分析 Agent → 可视化 Agent	数据的分层数和异常点	可视化忽略异常点，图表误导
任意 Agent → 下一步	前置假设和排除的路径	后续模型重复探索已排除的路径

二、上下文传递的三种模式

Agent 之间传上下文不是简单地把上一个模型的输出塞给下一个模型。不同场景需要不同的传递模式：

flowchart TB
    A[上下文传递模式] --> B[模式1: 结构化接力]
    A --> C[模式2: 共享记忆池]
    A --> D[模式3: 分层压缩]

    B --> E[适合: 固定流程<br>SQL→分析→报告]
    C --> F[适合: 并行探索<br>多个Agent同时工作]
    D --> G[适合: 长链路任务<br>6步以上的分析]

    style B fill:#4ecdc4
    style C fill:#ffe66d
    style D fill:#ff6b6b

模式 1：结构化接力

每个 Agent 输出标准化的结构化数据，下一个 Agent 只接收结构化输入。像接力赛交接棒——棒的形状是固定的，不会变形。

# 结构化接力模式
from typing import TypedDict, List, Optional
from datetime import datetime

# 定义标准化的上下文结构
class SQLAgentOutput(TypedDict):
    """SQL Agent 的标准输出格式"""
    task_id: str                        # 任务唯一标识
    sql: str                            # 执行的SQL语句
    result_shape: tuple                 # 结果数据形状 (rows, cols)
    columns: List[dict]                 # 列信息：名称、类型、含义
    filters_applied: List[str]          # 查询中应用的过滤条件
    aggregation_logic: List[str]        # 聚合逻辑说明
    execution_time: float               # SQL执行耗时（秒）
    data_summary: dict                  # 数据概要统计
    anomalies_found: List[str]          # SQL阶段发现的异常

class AnalysisAgentOutput(TypedDict):
    """分析 Agent 的标准输出格式"""
    task_id: str
    input_task_id: str                  # 上一步的任务ID，形成链路
    analysis_type: str                  # 分析类型（对比/趋势/分布/归因）
    findings: List[dict]                # 发现的结论列表
    statistical_tests: List[dict]       # 统计检验结果（含置信区间）
    assumptions_made: List[str]         # 做了哪些假设
    paths_excluded: List[str]           # 排除了哪些分析路径
    confidence_level: str               # 整体置信度评级
    next_steps_suggested: List[str]     # 建议的下一步分析

class ReportAgentOutput(TypedDict):
    """报告 Agent 的标准输出格式"""
    task_id: str
    input_task_id: str
    title: str                          # 报告标题
    sections: List[dict]                # 报告章节
    key_metrics: List[dict]             # 核心指标（含数值+置信区间）
    visualizations_needed: List[dict]   # 需要的可视化类型和数据
    limitations: List[str]              # 分析局限说明


# Agent 协作链路
def run_structured_pipeline(user_query: str) -> dict:
    """结构化接力：SQL → 分析 → 报告"""

    # 第一步：SQL Agent
    sql_output = sql_agent(user_query)

    # 第二步：分析 Agent（接收SQL Agent的完整输出）
    analysis_input = {
        "data_description": sql_output,
        "user_intent": user_query,
        "prior_context": None   # 首步没有前置上下文
    }
    analysis_output = analysis_agent(analysis_input)

    # 第三步：报告 Agent（接收分析Agent的完整输出+SQL Agent的关键信息）
    report_input = {
        "analysis_results": analysis_output,
        "data_scope": {         # 从SQL Agent传递关键元信息
            "filters": sql_output["filters_applied"],
            "shape": sql_output["result_shape"],
            "anomalies": sql_output["anomalies_found"]
        },
        "user_intent": user_query
    }
    report_output = report_agent(report_input)

    return {
        "pipeline_trace": {
            "sql_task_id": sql_output["task_id"],
            "analysis_task_id": analysis_output["task_id"],
            "report_task_id": report_output["task_id"]
        },
        "final_report": report_output
    }

结构化接力模式的关键：每个 Agent 的输出格式是预先定义的，不能随意增减字段。这保证了下一个 Agent 总是能拿到它需要的所有信息。

三、共享记忆池：并行 Agent 的信息交换

多个 Agent 并行工作时，它们需要共享一个记忆池——像团队协作时的共享文档，每个人都能读到其他人写的内容，也能往里追加自己的发现。

# 共享记忆池模式
class SharedMemoryPool:
    """Agent协作的共享记忆池"""

    def __init__(self):
        self.entries = []       # 所有Agent的输出记录
        self.facts = {}         # 已确认的事实
        self.hypotheses = {}    # 待验证的假设
        self.contradictions = [] # 发现的矛盾点

    def add_entry(self, agent_name: str, entry_type: str,
                  content: dict, confidence: float) -> str:
        """Agent向记忆池追加一条记录"""
        entry_id = f"{agent_name}_{len(self.entries)}"

        entry = {
            "id": entry_id,
            "agent": agent_name,
            "type": entry_type,         # fact/hypothesis/observation/question
            "content": content,
            "confidence": confidence,   # 0-1置信度
            "timestamp": datetime.now().isoformat(),
            "references": []            # 引用了哪些其他entry
        }

        self.entries.append(entry)

        # 如果是高置信度的事实，加入事实库
        if entry_type == "fact" and confidence >= 0.8:
            self.facts[entry_id] = content

        # 如果是假设，加入假设库
        elif entry_type == "hypothesis":
            self.hypotheses[entry_id] = content

        # 检查是否与已有信息矛盾
        self._check_contradictions(entry)

        return entry_id

    def get_relevant_context(self, agent_name: str,
                              task_description: str) -> dict:
        """为Agent提取与当前任务相关的上下文"""
        relevant_facts = []
        relevant_hypotheses = []

        for entry in self.entries:
            # 简单相关性判断：检查内容是否包含任务关键词
            if self._is_relevant(entry["content"], task_description):
                if entry["type"] == "fact":
                    relevant_facts.append(entry)
                elif entry["type"] == "hypothesis":
                    relevant_hypotheses.append(entry)

        return {
            "relevant_facts": relevant_facts,
            "relevant_hypotheses": relevant_hypotheses,
            "contradictions": self.contradictions,
            "total_entries": len(self.entries)
        }

    def _check_contradictions(self, new_entry: dict):
        """检查新entry与已有信息是否矛盾"""
        if new_entry["type"] == "fact":
            for fact_id, fact_content in self.facts.items():
                if self._is_contradictory(new_entry["content"], fact_content):
                    self.contradictions.append({
                        "new_entry": new_entry["id"],
                        "existing_fact": fact_id,
                        "description": "新发现与已确认事实冲突"
                    })

    def _is_relevant(self, content: dict, description: str) -> bool:
        """简单相关性判断"""
        # 将内容转为字符串后做关键词匹配
        content_str = str(content).lower()
        # 提取任务描述的关键词
        keywords = description.lower().split()
        return any(kw in content_str for kw in keywords)

    def _is_contradictory(self, new_content: dict,
                          existing_content: dict) -> bool:
        """判断两个内容是否矛盾（简化实现）"""
        # 实际实现会用AI判断语义冲突
        return False   # 临时实现

共享记忆池的好处是——Agent 之间不需要直接"交接棒"，而是通过公共区域交换信息。这样多个 Agent 可以并行工作，互相看到对方的发现，避免重复探索和结论冲突。

四、分层压缩：长链路的信息保真

当分析链路超过 5 步时，上下文会越来越长。如果每一步都把上一步的全部输出传下去，最后一步的模型可能收到上万字的上下文——信息太多反而难以聚焦。

分层压缩的核心思路：关键信息全量传递，细节信息摘要压缩。

flowchart TB
    A[Agent1 全量输出] --> B[压缩层]
    B --> C[关键信息: 全量传递]
    B --> D[细节信息: 摘要压缩]
    C --> E[Agent2 接收]
    D --> E

    E --> F[Agent2 全量输出]
    F --> G[压缩层]
    G --> H[关键信息: 全量传递]
    G --> I[Agent1的关键信息: 二次压缩]
    I --> J[Agent3 接收]
    H --> J

    style B fill:#4ecdc4
    style G fill:#4ecdc4

# 分层压缩模式
def compress_context(full_output: dict, layer: int) -> dict:
    """根据链路层数压缩上下文

    参数：
        full_output: Agent的完整输出
        layer: 当前在第几层传递（1=最近一步，2=两步前...）
    """

    if layer <= 2:
        # 最近2步：几乎全量传递，只去掉大块原始数据
        compressed = {
            "task_id": full_output["task_id"],
            "analysis_type": full_output.get("analysis_type", ""),
            "findings": full_output.get("findings", []),
            "statistical_tests": full_output.get("statistical_tests", []),
            "confidence_level": full_output.get("confidence_level", ""),
            "filters_applied": full_output.get("filters_applied", []),
            "assumptions_made": full_output.get("assumptions_made", []),
            "paths_excluded": full_output.get("paths_excluded", []),
            # 压缩：去掉原始数据详情，只保留概要
            "data_summary": full_output.get("data_summary", {}),
        }

    elif layer <= 4:
        # 3-4步前：摘要级压缩，只保留核心发现
        compressed = {
            "task_id": full_output["task_id"],
            "key_findings": [
                f["summary"] for f in full_output.get("findings", [])
            ],  # 只保留发现的一句话摘要
            "confidence_level": full_output.get("confidence_level", ""),
            "assumptions_made": full_output.get("assumptions_made", []),
            "paths_excluded": full_output.get("paths_excluded", []),
        }

    else:
        # 5步以上：极简压缩，只保留结论和关键假设
        compressed = {
            "task_id": full_output["task_id"],
            "conclusion": " ".join([
                f["summary"] for f in full_output.get("findings", [])[:3]
            ]),
            "key_assumptions": full_output.get("assumptions_made", [])[:5],
            "confidence_level": full_output.get("confidence_level", ""),
        }

    return compressed


# 长链路协作示例
def run_long_pipeline(user_query: str, num_steps: int = 6) -> dict:
    """6步以上的长链路协作，使用分层压缩传递上下文"""

    memory = SharedMemoryPool()
    pipeline_trace = []
    context_stack = []     # 累积的上下文栈

    agents = [
        ("sql_agent", sql_agent),
        ("data_quality_agent", data_quality_agent),
        ("stats_agent", stats_agent),
        ("trend_agent", trend_agent),
        ("insight_agent", insight_agent),
        ("report_agent", report_agent)
    ]

    for step, (agent_name, agent_fn) in enumerate(agents[:num_steps]):
        # 构建当前Agent的输入上下文
        if step == 0:
            # 第一步：只有用户原始查询
            input_context = {"user_query": user_query}
        else:
            # 后续步骤：压缩后的历史上下文 + 当前任务
            input_context = {
                "current_task": user_query,
                "recent_context": compress_context(
                    context_stack[-1], layer=1
                ),
                "earlier_context": [
                    compress_context(ctx, layer=step - i)
                    for i, ctx in enumerate(context_stack[:-1])
                ],
                "shared_memory": memory.get_relevant_context(
                    agent_name, user_query
                )
            }

        # 执行Agent
        output = agent_fn(input_context)

        # 存入上下文栈和共享记忆池
        context_stack.append(output)
        memory.add_entry(
            agent_name, "observation",
            output, float(output.get("confidence_level", "0.5"))
        )

        pipeline_trace.append({
            "step": step + 1,
            "agent": agent_name,
            "task_id": output["task_id"]
        })

    return {
        "pipeline_trace": pipeline_trace,
        "final_output": context_stack[-1],
        "shared_memory_summary": {
            "total_facts": len(memory.facts),
            "total_hypotheses": len(memory.hypotheses),
            "contradictions": len(memory.contradictions)
        }
    }


# Agent函数示例（简化实现）
def sql_agent(input_ctx: dict) -> dict:
    """SQL Agent: 从用户查询生成并执行SQL"""
    return {
        "task_id": "sql_001",
        "sql": "SELECT ... FROM ... WHERE ...",
        "findings": [{"summary": "GMV环比下降15%", "detail": "..."}],
        "filters_applied": ["dt >= 2026-06-01", "status = paid"],
        "confidence_level": "0.9",
        "assumptions_made": ["数据延迟不超过24小时"],
        "paths_excluded": ["排除取消订单的影响"]
    }

def data_quality_agent(input_ctx: dict) -> dict:
    """数据质量 Agent: 检查数据质量"""
    return {
        "task_id": "dq_001",
        "analysis_type": "quality_check",
        "findings": [{"summary": "6月数据完整度99.2%", "detail": "..."}],
        "confidence_level": "0.95",
        "assumptions_made": [],
        "paths_excluded": []
    }

def stats_agent(input_ctx: dict) -> dict:
    """统计分析 Agent: 执行统计检验"""
    return {
        "task_id": "stats_001",
        "analysis_type": "hypothesis_test",
        "findings": [{"summary": "GMV下降统计显著(p<0.01), Cohen's d=0.8", "detail": "..."}],
        "statistical_tests": [{"test": "t-test", "p_value": 0.003, "effect_size": 0.8}],
        "confidence_level": "0.85",
        "assumptions_made": ["数据近似正态分布"],
        "paths_excluded": []
    }

def trend_agent(input_ctx: dict) -> dict:
    return {"task_id": "trend_001", "analysis_type": "trend", "findings": [{"summary": "下降趋势持续3周", "detail": "..."}], "confidence_level": "0.7", "assumptions_made": [], "paths_excluded": []}

def insight_agent(input_ctx: dict) -> dict:
    return {"task_id": "insight_001", "analysis_type": "insight", "findings": [{"summary": "核心原因是新用户转化率下降", "detail": "..."}], "confidence_level": "0.75", "assumptions_made": ["假设转化率是主要原因"], "paths_excluded": ["排除竞品影响（无数据支持）"]}

def report_agent(input_ctx: dict) -> dict:
    return {"task_id": "report_001", "title": "GMV异动分析报告", "sections": [], "confidence_level": "0.8", "assumptions_made": [], "paths_excluded": []}

分层压缩的规则：

• 最近 2 步几乎全量传递——最新信息最可能被当前 Agent 直接引用
• 3-4 步前只保留核心发现的一句话摘要——细节已经不那么重要了
• 5 步以上只保留结论和关键假设——避免远古信息干扰当前判断

必须全量传递的关键信息：

1. 过滤条件——不知道数据范围就无法正确解读结论
2. 假设列表——不知道前置假设就无法判断结论的适用条件
3. 排除路径——不知道哪些路径已排除就会重复探索

这三类信息无论压缩到什么程度都不能丢。它们是"交接棒"的核心——棒的形状可以简化，但棒的材质不能变。

五、总结

AI 数据分析 Agent 协作的核心问题是上下文传递——模型之间怎么传信息不丢关键内容。

三种传递模式各有适用场景：

• 结构化接力：适合固定流程（SQL→分析→报告），每个 Agent 输出标准化格式，下一个 Agent 只接收结构化输入
• 共享记忆池：适合并行探索，多个 Agent 通过公共记忆池交换发现，避免重复探索和结论冲突
• 分层压缩：适合长链路任务，最近 2 步全量传递，3-4 步保留摘要，5 步以上只留结论和假设

三类不能丢的信息：

1. 过滤条件——不知道数据范围就无法正确解读结论
2. 前置假设——不知道假设就无法判断结论的适用条件
3. 排除路径——不知道已排除路径就会重复探索

Agent 协作不是简单的串行调用——它是接力赛，交接棒不能掉。上下文传递的质量，决定了协作链路的整体质量。结构化、共享、压缩，三种模式组合使用，才能在信息保真和效率之间找到平衡。