AI 数据分析 Agent 协作模型之间怎么传上下文不丢信息一、Agent 协作多个模型的接力赛单个 AI 模型能做很多事但复杂的数据分析任务往往需要多个模型分工合作——一个擅长写 SQL一个擅长统计分析一个擅长可视化一个擅长写报告。就像接力赛每个运动员跑自己擅长的段落交接棒时不能掉否则整场比赛就废了。Agent 之间的交接棒就是上下文传递。模型 A 的分析结果要完整准确地传给模型 BB 基于这些结果继续分析再传给模型 C。如果交接棒掉了——关键信息丢失、数据被误读、结论被扭曲——后续模型的分析就是空中楼阁。上下文丢失的典型场景场景丢失什么后果SQL Agent → 分析 Agent查询的过滤条件和聚合逻辑分析 Agent 不知道数据范围结论偏差分析 Agent → 报告 Agent统计检验的置信区间和效应大小报告只写显著不写效应多大分析 Agent → 可视化 Agent数据的分层数和异常点可视化忽略异常点图表误导任意 Agent → 下一步前置假设和排除的路径后续模型重复探索已排除的路径二、上下文传递的三种模式Agent 之间传上下文不是简单地把上一个模型的输出塞给下一个模型。不同场景需要不同的传递模式flowchart TB A[上下文传递模式] -- B[模式1: 结构化接力] A -- C[模式2: 共享记忆池] A -- D[模式3: 分层压缩] B -- E[适合: 固定流程brSQL→分析→报告] C -- F[适合: 并行探索br多个Agent同时工作] D -- G[适合: 长链路任务br6步以上的分析] 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], layer1 ), earlier_context: [ compress_context(ctx, layerstep - 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下降统计显著(p0.01), Cohens d0.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 步以上只保留结论和关键假设——避免远古信息干扰当前判断必须全量传递的关键信息过滤条件——不知道数据范围就无法正确解读结论假设列表——不知道前置假设就无法判断结论的适用条件排除路径——不知道哪些路径已排除就会重复探索这三类信息无论压缩到什么程度都不能丢。它们是交接棒的核心——棒的形状可以简化但棒的材质不能变。五、总结AI 数据分析 Agent 协作的核心问题是上下文传递——模型之间怎么传信息不丢关键内容。三种传递模式各有适用场景结构化接力适合固定流程SQL→分析→报告每个 Agent 输出标准化格式下一个 Agent 只接收结构化输入共享记忆池适合并行探索多个 Agent 通过公共记忆池交换发现避免重复探索和结论冲突分层压缩适合长链路任务最近 2 步全量传递3-4 步保留摘要5 步以上只留结论和假设三类不能丢的信息过滤条件——不知道数据范围就无法正确解读结论前置假设——不知道假设就无法判断结论的适用条件排除路径——不知道已排除路径就会重复探索Agent 协作不是简单的串行调用——它是接力赛交接棒不能掉。上下文传递的质量决定了协作链路的整体质量。结构化、共享、压缩三种模式组合使用才能在信息保真和效率之间找到平衡。
AI 数据分析 Agent 协作模型之间怎么传上下文不丢信息一、Agent 协作多个模型的接力赛单个 AI 模型能做很多事但复杂的数据分析任务往往需要多个模型分工合作——一个擅长写 SQL一个擅长统计分析一个擅长可视化一个擅长写报告。就像接力赛每个运动员跑自己擅长的段落交接棒时不能掉否则整场比赛就废了。Agent 之间的交接棒就是上下文传递。模型 A 的分析结果要完整准确地传给模型 BB 基于这些结果继续分析再传给模型 C。如果交接棒掉了——关键信息丢失、数据被误读、结论被扭曲——后续模型的分析就是空中楼阁。上下文丢失的典型场景场景丢失什么后果SQL Agent → 分析 Agent查询的过滤条件和聚合逻辑分析 Agent 不知道数据范围结论偏差分析 Agent → 报告 Agent统计检验的置信区间和效应大小报告只写显著不写效应多大分析 Agent → 可视化 Agent数据的分层数和异常点可视化忽略异常点图表误导任意 Agent → 下一步前置假设和排除的路径后续模型重复探索已排除的路径二、上下文传递的三种模式Agent 之间传上下文不是简单地把上一个模型的输出塞给下一个模型。不同场景需要不同的传递模式flowchart TB A[上下文传递模式] -- B[模式1: 结构化接力] A -- C[模式2: 共享记忆池] A -- D[模式3: 分层压缩] B -- E[适合: 固定流程brSQL→分析→报告] C -- F[适合: 并行探索br多个Agent同时工作] D -- G[适合: 长链路任务br6步以上的分析] 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], layer1 ), earlier_context: [ compress_context(ctx, layerstep - 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下降统计显著(p0.01), Cohens d0.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 步以上只保留结论和关键假设——避免远古信息干扰当前判断必须全量传递的关键信息过滤条件——不知道数据范围就无法正确解读结论假设列表——不知道前置假设就无法判断结论的适用条件排除路径——不知道哪些路径已排除就会重复探索这三类信息无论压缩到什么程度都不能丢。它们是交接棒的核心——棒的形状可以简化但棒的材质不能变。五、总结AI 数据分析 Agent 协作的核心问题是上下文传递——模型之间怎么传信息不丢关键内容。三种传递模式各有适用场景结构化接力适合固定流程SQL→分析→报告每个 Agent 输出标准化格式下一个 Agent 只接收结构化输入共享记忆池适合并行探索多个 Agent 通过公共记忆池交换发现避免重复探索和结论冲突分层压缩适合长链路任务最近 2 步全量传递3-4 步保留摘要5 步以上只留结论和假设三类不能丢的信息过滤条件——不知道数据范围就无法正确解读结论前置假设——不知道假设就无法判断结论的适用条件排除路径——不知道已排除路径就会重复探索Agent 协作不是简单的串行调用——它是接力赛交接棒不能掉。上下文传递的质量决定了协作链路的整体质量。结构化、共享、压缩三种模式组合使用才能在信息保真和效率之间找到平衡。
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
