Agent Systems / Empirical Workflow
Agent 实证研究工作流:从研究问题到可审计证据包
把 Agent 看成带状态、工具、证据门和人工检查点的研究流水线,而不是一次性聊天回答。
Mechanism Lab
动画:研究 Agent 如何沿 DAG 生成证据包
动画展示研究问题进入计划节点,数据和代码并行执行,模型进入诊断 gate;失败会返回计划,成功才进入报告和复现包。
Step 1 / 5
State
研究问题、文件、数据版本、代码和人工反馈组成当前状态。
s_t={task,files,data,code,checks}Animation Control
Reduced-motion users receive the same step states without continuous motion.
01 / 直觉
核心直觉
实证研究 Agent 的对象不是单条 prompt,而是一个长期状态 s:研究问题、数据路径、代码、模型、日志、输出和人工反馈都在状态里持续更新。
每一步 action 都应是可审计的工具调用,例如检索文献、读取数据、运行脚本、生成表格、检查平衡性或渲染论文段落。
工作流必须有 gate:数据清洗是否可复跑、模型是否通过诊断、表格是否来自代码、引用是否存在、结论是否超出识别假设。
人类不是最后才出现的审稿人,而是关键节点的权限控制者:确认数据来源、识别策略、外部发送、删除操作和最终解释。
02 / 数学
把研究流程写成状态机和有向证据图
01 / 状态定义
令 s_t 包含任务说明、文件清单、数据版本、代码版本、模型对象、验证结果和人工反馈。Agent 每步只允许通过工具改变状态。
s_t={task,files,data,code,models,checks,notes}02 / 动作策略
策略 pi 根据当前状态选择下一步工具动作。动作不是自由文本,而是工具名、参数和预期输出。
a_t ~ pi(a | s_t), a_t={tool,args,expected}03 / 执行与观察
环境执行工具,返回 observation,例如日志、错误、表格路径、检索文档或诊断统计量。
o_t=E(a_t,s_t)04 / 证据门
验证器把 observation 映射为通过/失败/需要人工判断。失败不能被语言包装成成功,必须回到计划或修复节点。
g_t=V(o_t,s_t) in {pass,fail,review}05 / 状态更新
只有通过验证或被人工确认的 observation 才能进入可信状态;失败结果也要记录为 trace。
s_{t+1}=U(s_t,a_t,o_t,g_t)06 / 最终证据包
完成不是“生成一段话”,而是得到可复跑的输入、命令、输出、日志、假设和人工审阅记录。
B={D,C,cmd,outputs,logs,assumptions,review}03 / 代码
Python 演示:带 gate 的研究 Agent 状态机
下面代码展示一个最小工作流调度器:每一步都记录 action、observation 和 gate,失败会进入修复路径,而不是伪装成完成。
from dataclasses import dataclass, field
from typing import Callable
@dataclass
class State:
task: str
data_ready: bool = False
model_ready: bool = False
report_ready: bool = False
trace: list[dict] = field(default_factory=list)
def clean_data(state: State) -> dict:
return {"rows": 1280, "missing_rate": 0.018, "artifact": "data/clean.parquet"}
def estimate_model(state: State) -> dict:
if not state.data_ready:
return {"error": "clean data missing"}
return {"coef": 0.12, "se": 0.04, "artifact": "models/did.json"}
def render_report(state: State) -> dict:
if not state.model_ready:
return {"error": "model not verified"}
return {"artifact": "outputs/report.md", "claim": "effect is positive"}
def gate(tool_name: str, obs: dict) -> str:
if "error" in obs:
return "fail"
if tool_name == "clean_data" and obs["missing_rate"] < 0.05:
return "pass"
if tool_name == "estimate_model" and abs(obs["coef"] / obs["se"]) > 1.96:
return "review"
if tool_name == "render_report":
return "pass"
return "fail"
TOOLS: dict[str, Callable[[State], dict]] = {
"clean_data": clean_data,
"estimate_model": estimate_model,
"render_report": render_report,
}
def choose_action(state: State) -> str:
if not state.data_ready:
return "clean_data"
if not state.model_ready:
return "estimate_model"
return "render_report"
state = State(task="Build an auditable DID research note")
for _ in range(5):
action = choose_action(state)
observation = TOOLS[action](state)
status = gate(action, observation)
state.trace.append({"action": action, "observation": observation, "gate": status})
if action == "clean_data" and status == "pass":
state.data_ready = True
elif action == "estimate_model" and status in {"pass", "review"}:
# Human review is required before the claim is externally used.
state.model_ready = True
elif action == "render_report" and status == "pass":
state.report_ready = True
break
print("report ready:", state.report_ready)
print("trace:")
for row in state.trace:
print(row)04 / 案例
案例:从课程项目到审稿级复现包
- 学生或研究者希望 Agent 完成一个政策评估项目:整理原始数据、构造面板、估计 DID、生成回归表、写结果解释并准备复现包。
- 低质量做法是让模型直接写结论。高质量工作流会把任务拆成 DAG:raw data、clean data、features、model、diagnostics、tables、figures、manuscript 和 review notes。
- 每个节点都要有 gate:样本数是否漂移、键是否重复、模型是否收敛、平行趋势图是否生成、表格是否来自模型对象、文字是否引用了正确输出。
- 最终交付不是单个 answer,而是一组可审计 artifact:数据处理日志、命令、表格、截图、人工确认点和仍然开放的识别假设。
05 / 风险
常见误区
参考资料
- Yao et al. (2022), ReAct: Synergizing Reasoning and Acting in Language Modelshttps://arxiv.org/abs/2210.03629
- Project TIER Protocolhttps://www.projecttier.org/tier-protocol/
- ACM Artifact Review and Badginghttps://www.acm.org/publications/policies/artifact-review-and-badging-current
- AEA Data and Code Availability Policyhttps://www.aeaweb.org/journals/data/data-code-policy