当前位置：首页 > ops >正文

Causal-Copilot: An Autonomous Causal Analysis Agent 论文解读

ops 2025/8/21 13:37:44

一句话总结：用于探索数据集的因果关系的agent

在这里插入图片描述
从图中可以看到，模块有
用户意图理解，预处理、算法选择、后处理、模拟。

KIMI-k2的总结

下面仍然按 5 大模块 + 报告模块逐块总结，并给出「如果后端是 LangGraph，这一坨东西到底对应哪段代码 / 哪个机制」。你熟悉 LangGraph，就直接看「→ LangGraph 落地方式」即可。

──────────────────
1 用户交互模块
一句话总结
把自然语言转成结构化任务，并在 pipeline 关键节点让用户“插话”改需求。

→ LangGraph 落地方式
• 节点：user_input_node（首次） + human_feedback_node（循环）。
• State 字段：nl_query（str） + structured_query（Pydantic）。
• 人工介入：在 preprocess / post_process 后统一用 interrupt_before=["next_node"]，前端把用户修改 POST 到 /threads/{tid}/state → 引擎内部就是 graph.update_state(...)。

──────────────────
2 预处理模块
一句话总结
洗数据 → 统计体检 → 输出数据摘要，为算法选型提供“体检报告”。

→ LangGraph 落地方式
• 节点：preprocess_node
输入：state[“raw_data”]
输出：state[“clean_data”]、state[“data_report”]
• 工具调用：在节点内部用普通 Python 函数（pandas、scipy），LangGraph 不限制；耗时任务可标记 RunnableConfig(tags=["io_bound"]) 以便线程池调度。
• 记忆：data_report 直接写 State，无需外部 DB，LangGraph checkpoint 会自动持久化。

──────────────────
3 算法选择模块
一句话总结
根据体检报告 + 用户约束，筛 → 排 → 配超参 → 跑算法。

→ LangGraph 落地方式
a) 筛 / 排
• 节点：algo_filter_node（LLM 生成候选列表）
• 输出：state[“algo_candidates”]
b) 超参
• 节点：hyper_config_node（LLM + optuna）
• 输出：state[“hyper_params”]
c) 执行
• 节点：run_algo_node
• 可封装为 Tool：
python @tool def run_causal_algorithm(algo: str, params: dict, df: pd.DataFrame) -> dict: ...
节点里直接 tool.invoke(...)，LangGraph 会自动把调用轨迹写进 checkpoint，方便以后复现。

──────────────────
4 后处理模块
一句话总结
bootstrap 置信度 → LLM 语义合理性 → 用户可再次修改图。

→ LangGraph 落地方式
• 节点：bootstrap_node、llm_refine_node、human_revise_node（带 interrupt）。
• 循环边：

workflow.add_edge("human_revise_node", "llm_refine_node")   # 循环直到满意

• 图对象：networkx.DiGraph 直接塞进 state[“refined_graph”]；序列化用 nx.node_link_data() 转 dict。

──────────────────
5 报告生成模块
一句话总结
把图、效应估计、假设检验打包成 PDF。

→ LangGraph 落地方式
• 节点：gen_report_node
• 工具：

render_latex_template(state) → 返回 .tex 字符串
compile_pdf(tex) → 返回文件路径
• 输出：state[“pdf_path”]
• 节点内可直接用 DockerOperator 跑 LaTeX 容器，LangGraph 支持在节点级别声明 env={"DOCKER_IMAGE": "texlive:latest"}。

──────────────────
跨模块总线（LLM 调度器）
一句话总结
LLM 不只是在某个节点里回答问题，而是负责“下一步去哪个节点”。

→ LangGraph 落地方式
• 整个 workflow 就是一张图；LLM 的“调度”体现在：
– 普通边：固定路径
– 条件边：
python def route_after_preprocess(state): if state["data_report"]["is_timeseries"]: return "ts_algo_filter_node" return "tabular_algo_filter_node" workflow.add_conditional_edges("preprocess_node", route_after_preprocess)
• 全局 LLM 调用：可使用 LangGraph 的 Command 机制，在任意节点里 return Command(goto="back_to_user", update={...}) 实现动态跳转。

──────────────────
部署 & 运行示例（最小可跑片段）

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaverworkflow = StateGraph(CausalState)# 注册节点
workflow.add_node("user_input", user_input_node)
workflow.add_node("preprocess", preprocess_node)
workflow.add_node("algo_filter", algo_filter_node)
workflow.add_node("run_algo", run_algo_node)
workflow.add_node("post_process", post_process_node)
workflow.add_node("gen_report", gen_report_node)# 边
workflow.add_edge(START, "user_input")
workflow.add_edge("user_input", "preprocess")
workflow.add_edge("preprocess", "algo_filter")
workflow.add_edge("algo_filter", "run_algo")
workflow.add_edge("run_algo", "post_process")
workflow.add_edge("post_process", "gen_report")
workflow.add_edge("gen_report", END)# checkpoint
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)# 运行
config = {"configurable": {"thread_id": "abc123"}}
app.invoke({"nl_query": "吸烟是否导致肺癌？"}, config=config)