ai-data-science-team 的 LangChain 对话迁移方案Message-First 架构改造实战指南【免费下载链接】ai-data-science-teamAn AI-powered data science team of agents to help you perform common data science tasks 10X faster.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-data-science-team本篇技术指南围绕仓库内规划文档 planning_docs/package_review_initial_release/langchain_conversation_plan.md 展开系统讲解 ai-data-science-team 将全部 Agent 从字符串指令调用方式迁移至以HumanMessage/AIMessage为核心的消息优先Message-First结构的完整方案。该迁移使各 Agent 的状态统一以messages: Sequence[BaseMessage]为单一事实源从而支撑 supervisor 调度与多智能体团队Multi-Agent Teams的对话式编排。读完本文你将掌握迁移的三大设计原则、按 Agent 逐一改造的 7 个标准步骤、试点 Agent 的源码级实现细节、向后兼容层的设计思路以及回滚与风险控制方法。一、迁移背景为什么需要 Message-First 结构ai-data-science-team 是一套由多个 LangGraph Agent 组成的数据科学工具集早期各 Agent 的对外接口是invoke_agent(user_instructions...)这类字符串指令 业务字段的调用约定。这种模式在单个 Agent 独立使用时没有问题但一旦进入 supervisor/team 架构例如 supervisor_ds_team.py 所实现的监督式数据科学团队就会出现明显的结构性障碍对话上下文无法传递字符串指令只表达这一轮的需求无法携带多轮对话历史路由信息缺失supervisor 需要根据完整的HumanMessage/AIMessage序列判断下一个该派给哪个 worker字符串接口无法提供统一、标准化的输入形态状态难以聚合各 Agent 自定义的返回键如data_wrangled、recommended_steps形态各异团队层无法用统一逻辑读取或评估。因此迁移计划定义的目标是在不破坏现有 API 的前提下将 agents 迁移到以HumanMessage/AIMessage为核心的消息优先结构实现与 supervisor/team 架构的兼容。从源码看这一目标已通过新增消息入口 保留旧入口的双轨方式落地。二、三大设计原则迁移计划明确了三条贯穿始终的原则它们是后续所有改造决策的评判标准原则含义落地证据保持向后兼容保留invoke_agent(user_instructions...)签名新增基于消息的入口各 Agent 类中invoke_agent与invoke_messages并存单一事实源状态State统一携带messages: Sequence[BaseMessage]所有下游逻辑从messages读取GraphState中messages: Annotated[Sequence[BaseMessage], operator.add]工具透明性按需把工具调用消息或摘要纳入messages由开关控制记录详略log_tool_callsTrue参数控制工具调用日志输出其中单一事实源是迁移的核心一旦messages成为所有 Agent 共享的规范化状态键supervisor 就可以用同一条 reducer 逻辑合并来自不同 worker 的对话片段进而路由与评估完整对话。三、迁移的 7 个标准步骤按 Agent 逐一改造迁移计划给出了一套可复用的改造模板每个 Agent 按以下 7 步执行步骤 1输入规范化InputsAgent 的输入首选接受messages当调用方仍传user_instructions字符串时将其包装为[HumanMessage(content...)]并在状态中统一规范化为messages。以试点 Agent 为例data_loader_tools_agent.py 中的prepare_messages节点正是这一原则的实现def prepare_messages(state: GraphState): print(format_agent_name(AGENT_NAME)) print( * PREPARE MESSAGES) if state.get(messages): return {} return {messages: [(user, state.get(user_instructions))]}该节点作为图的入口节点确保无论调用方传入字符串还是消息列表进入下游逻辑前messages必然存在。步骤 2状态 Schema 扩展State Schema状态 Schema 需包含messages: Annotated[Sequence[BaseMessage], operator.add]同时保留各 Agent 特有的业务字段data、errors、summaries等。operator.add是 LangGraph 的累加式 reducer用于把每轮节点返回的新消息追加到既有消息序列中。在 data_loader_tools_agent.py 中GraphState继承自 LangGraph 的AgentState并扩展了业务键class GraphState(AgentState): user_instructions: str messages: Annotated[Sequence[BaseMessage], operator.add] data_loader_artifacts: dict tool_calls: List[str]注意AgentState本身也定义了messages通道这里通过显式声明强化了单一事实源的约定。步骤 3输出规范化OutputsAgent 的最终回答以AIMessage形式追加进messages同时保留原有结果键如data_wrangled、recommended_steps以维持兼容性。也就是说messages负责对话记录业务键负责机器可读产物两者各司其职。在试点 Agent 的post_process节点data_loader_tools_agent.py中可以看到这种双轨输出last_ai_message AIMessage(getattr(last_ai, content, ), roleAGENT_NAME) # ... return { messages: [last_ai_message], internal_messages: internal_messages, data_loader_artifacts: artifacts if artifacts else last_tool_artifact, tool_calls: tool_calls, }此外模板层的报告节点 node_func_report_agent_outputs 也遵循同一约定——将各业务键汇总为 JSON 后封装成AIMessage写入result_keymessages模板层的 node_func_explain_agent_code 同样以AIMessage作为代码解释的输出载体。这说明输出即消息已成为整个仓库的通用惯例。步骤 4工具调用记录Tool Calls对于 ReAct/工具型 Agent可按需将工具调用消息纳入messages并用一个开关参数如log_tool_callsTrue控制转录详略避免对话记录被工具日志淹没。试点 Agent 在 post_process 中先通过 utils/messages.py 的get_tool_call_names从消息中提取工具名再依据log_tool_calls决定是否打印* Tool: name与捕获的 artifacts 键tool_calls get_tool_call_names(internal_messages) if tool_calls and log_tool_calls: for name in tool_calls: # 尝试在上一消息中查找 artifact 路径作为提示 ... print(f * Tool: {name}{path_hint})同一开关模式也出现在 eda_tools_agent.py 与 mlflow_tools_agent.py 中三处默认值均为True。步骤 5访问器更新Accessors将 getter 方法改为从messages中读取最后一条 assistant 消息同时保留返回 artifacts/data 的辅助方法确保旧调用方期望字符串返回不受影响。试点 Agent 的 get_ai_message 演示了倒序扫描最后一条 AI 消息的标准写法msgs self.response.get(messages, []) last_ai None for msg in reversed(msgs): role getattr(msg, role, None) or getattr(msg, type, None) if role in (assistant, ai): last_ai msg break if last_ai is None and msgs: last_ai msgs[-1]与此配套get_internal_messages 返回完整内部消息支持markdownTrue格式化输出get_artifacts 则负责把工具产物按需转换为 DataFrame形成对话与产物分离的访问模式。步骤 6入口节点保障Entry Nodes每个 Agent 图的第一个节点必须确保messages已存在字符串输入在此被包装然后再进入下游逻辑。试点 Agent 的prepare_messages即承担此职责而 data_cleaning_agent.py 的invoke_agent在入口处就完成了字符串到消息的转换self.response self.invoke( { messages: [(user, user_instructions)] if user_instructions else [], user_instructions: user_instructions, data_raw: data_raw.to_dict(), max_retries: max_retries, retry_count: retry_count, }, **kwargs, )步骤 7Supervisor/团队接入Supervisors/Teams当messages在所有子 Agent 间标准化后supervisor 即可基于统一的消息序列完成路由与对话评估。这一点在 supervisor_ds_team.py 中得到充分体现SupervisorDSState以messages为团队级共享对话通道并使用自定义 reducer_supervisor_merge_messages进行合并supervisor 的路由 prompt 通过MessagesPlaceholder(variable_namemessages)把完整对话注入路由链supervisor_ds_team.py再结合 OpenAI function-calling 或文本解析输出下一个 worker。四、试点 Agent 源码深度解析data_loader_tools_agent迁移计划的推出顺序第一步是先在单个 Agent 上试点选中的是data_loader_tools_agent并标记为 ✅ donemessage-first、同步/异步消息入口、工具日志开关、目录/文件产物处理。该 Agent 也是理解整个迁移模板的最佳范本。4.1 四个消息入口DataLoaderToolsAgent 提供四类调用方式入口同步/异步用途invoke_agent(user_instructions...)同步兼容旧调用方内部将字符串包装为消息ainvoke_agent(user_instructions...)异步兼容旧调用方的异步版本invoke_messages(messages)同步面向 supervisor/teams 的首选入口直接传入Sequence[BaseMessage]ainvoke_messages(messages)异步异步版本同样面向 supervisor/teamsinvoke_messages的实现data_loader_tools_agent.py把user_instructions置为None、直接透传messages与旧入口在编译图上完全复用同一套节点逻辑——这正是不破坏现有 API原则的工程体现。4.2 三节点线性图该 Agent 的编译图为prepare_messages → react_agent → post_process的线性流程data_loader_tools_agent.pyprepare_messages入口保障见步骤 1react_agent调用create_react_agent构建的 ReAct 工具调用智能体其内部使用 LangGraph 预置的AgentState作为状态 Schema并注入system_hint指导工具选择例如用户问 LIST 文件时用搜索/列目录工具而不是加载内容见 data_loader_tools_agent.pypost_process从内部消息中提取最后一条 AI 回复、按工具名聚合 artifacts、汇总工具调用名最终返回messages 业务键。该 Agent 挂载的 6 个工具load_directory、load_file、list_directory_contents、list_directory_recursive、get_file_info、search_files_by_pattern定义于 tools/data_loader.py均在 data_loader_tools_agent.py 中注册工具的 artifacts 会通过post_process汇总到data_loader_artifacts键。五、向后兼容层双轨并行的工程保障迁移计划明确要求保留invoke_agent签名与字符串返回的 getter以应对两类既有依赖旧调用方期望字符串输入所有 Agent 的invoke_agent/ainvoke_agent继续接受user_instructions: str旧调用方期望字符串返回get_ai_message(markdownTrue)、get_internal_messages(markdownTrue)等仍可输出 Markdown 格式化文本。对数据清洗这类携带 DataFrame 的 Agent消息入口还额外做了指令回填当invoke_messages未显式传user_instructions时data_cleaning_agent.py 通过 get_last_user_message_content 从消息列表中提取最近一条 Human 消息内容作为指令保证下游推荐清洗步骤等节点无需改动即可工作。此外基类 BaseAgent 在invoke/ainvoke/stream/astream四个方法中统一对返回的messages执行remove_consecutive_duplicates去重agent_templates.py从框架层缓解多轮追加消息可能产生的重复记录问题——这是对State Drift状态漂移风险的又一道防线。六、推出顺序与转换检查清单6.1 分四步的灰度推出迁移计划采用由点及面的灰度策略降低整体回归风险试点单 Agentdata_loader_tools_agent✅ 已完成——message-first、同步/异步消息入口、工具日志开关、目录/文件产物处理全部落地扩展到工具类 Agent清洗、wrangling、可视化扩展到 SQL 与特征工程 Agent更新团队/supervisor 装配层使其完全依赖messages进行调度。6.2 转换检查清单当前状态计划文档记录的检查清单显示核心 Agent 已全部完成转换data_loader_tools_agent试点data_cleaning_agentmessage-first 入口与 demo 已添加data_wrangling_agent同上data_visualization_agent同上sql_database_agent同上feature_engineering_agent同上eda_tools_agentds_agentsh2o_ml_agentmessage-first 入口、校验调整与 demo 已添加mlflow_tools_agentml_agents源码搜索可以印证上述状态invoke_messages/ainvoke_messages方法已出现在 data_cleaning_agent.py、data_visualization_agent.py、data_wrangling_agent.py、feature_engineering_agent.py、sql_database_agent.py、eda_tools_agent.py、h2o_ml_agent.py 与 mlflow_tools_agent.py 等文件中。6.3 下一阶段非核心模块核心 Agent 转换完成后迁移进入非核心模块阶段计划包括multiagents 子包sql_data_analyst以 message-first 方式包装子 Agent、加入系统提示system hint、暴露子图subgraph、补充 demopandas_data_analyst同样实现 message-first 入口、系统提示、子图可见性并添加 demosupervised/team 变体如有应对齐 message-first 并暴露子图。从源码看pandas_data_analyst.py 与 sql_data_analyst.py 均已实现invoke_messages/ainvoke_messages且 pandas 分析器的状态归一化逻辑会保留system/human/assistant三类消息角色pandas_data_analyst.py说明该阶段工作已在持续推进apps更新 notebooks/demos 在合适场景改用invoke_messages。例如 supervisor_ds_team.ipynb 已导入HumanMessage并调用team.invoke_agentdata_loader_tools_agent.ipynb 则大量演示了基于字符串指令的invoke_agent用法两类调用方式在示例库中长期并存。七、风险与缓解措施迁移计划明确列出三大风险及对应缓解策略这些约束也解释了为何仓库中保留大量看似重复的兼容代码7.1 转录膨胀Transcript Bloat工具调用消息若全部写入messages会显著膨胀 LLM 上下文。缓解方式用log_tool_calls开关按需裁剪工具消息的详略。团队层还有更激进的兜底——supervisor_ds_team.py 定义了TEAM_MAX_MESSAGES 20与TEAM_MAX_MESSAGE_CHARS 2000两个常量其自定义 reducer_supervisor_merge_messages会丢弃tool/function角色消息、剔除冗长的 JSON 型 Agent Outputs 报告、截断超长消息体、并仅保留最后 20 条消息supervisor_ds_team.py从源头防止多步骤工作流中的 token 与速率限制问题。7.2 既有调用方期望字符串Existing Callers Expect Strings缓解方式完整保留invoke_agent签名与所有字符串返回型 getter。这正是第四节双轨并行设计的直接动机。7.3 状态漂移State Drift缓解方式节点始终返回新的messages避免对外部状态做原地修改。配合BaseAgent的去重逻辑agent_templates.py保证同一消息不会因 reducer 追加而在多次 invoke 之间反复累积。八、总结ai-data-science-team 的 LangChain 对话迁移方案是一份工程上可复制的 Message-First 改造模板它以messages: Sequence[BaseMessage]为单一事实源通过新增消息入口 保留旧接口 工具日志开关 消息化访问器四件套在零破坏的前提下完成了 9 个核心 Agent 的对话化改造并已向 multiagents 与 apps 层延伸。对于任何基于 LangGraph 构建多智能体系统的团队这份方案在向后兼容策略、状态 Schema 设计、上下文裁剪防膨胀、灰度推出顺序四个维度上都具备直接的借鉴价值——当你想让多个独立 Agent 会聊天并接受统一调度时从invoke_messages开始让对话历史成为唯一的权威数据源。【免费下载链接】ai-data-science-teamAn AI-powered data science team of agents to help you perform common data science tasks 10X faster.项目地址: https://gitcode.com/GitHub_Trending/ai/ai-data-science-team创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
