Agent可复现工程：回放·快照·回归六栈实测

from agentgit import AgentRepo, WorkflowVersion

repo = AgentRepo("./my_agent_workflow")
repo.init(graph=research_agent_graph)

# 执行一次工作流，自动创建快照
result = repo.run({"topic": "AI agents 2026"})
repo.commit(message="added fact-checker node", tags=["v1.2"])

# 质量下降？回滚到上一个版本
repo.checkout("v1.1")
result = repo.run({"topic": "AI agents 2026"})

# 创建实验分支
repo.branch("experiment-rag")
repo.switch("experiment-rag")
# 修改RAG检索参数...
repo.commit(message="tuned chunk_size to 512")

from temporalio import workflow, activity
from datetime import timedelta

@workflow.defn
class MultiAgentResearchWorkflow:
    @workflow.run
    async def run(self, topic: str) -> str:
        # 步骤1: 研究Agent——自动检查点
        research = await workflow.execute_activity(
            research_agent,
            topic,
            start_to_close_timeout=timedelta(minutes=5),
            # 如果LLM超时，自动重试+检查点恢复
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        # 步骤2: 写作Agent——即使服务重启也能恢复
        draft = await workflow.execute_activity(
            writing_agent,
            research,
            start_to_close_timeout=timedelta(minutes=10),
        )

        # 步骤3: 审校Agent
        final = await workflow.execute_activity(
            review_agent,
            draft,
            start_to_close_timeout=timedelta(minutes=5),
        )
        return final

from maestro import MAESTROEvaluator, AgentSystemConfig

evaluator = MAESTROEvaluator()

# 定义被评估的Multi-Agent系统
config = AgentSystemConfig(
    framework="crewai",  # 支持 crewai, langgraph, autogen, openai-adk
    agents=["researcher", "writer", "reviewer"],
    task="Write a 500-word article about AI agents",
)

# 运行评估套件
results = evaluator.evaluate(
    config,
    metrics=[
        "task_completion_rate",     # 任务完成率
        "inter_agent_coherence",    # Agent间信息一致性
        "error_propagation_depth",  # 错误传播深度
        "reproducibility_score",    # 可复现性得分
    ],
    runs=10,  # 跑10次取平均，评估可靠性
)

print(results.summary())
# Task Completion: 87.3% ± 4.2%
# Inter-Agent Coherence: 72.1% ± 11.5%
# Error Propagation Depth: 1.3 (avg)
# Reproducibility Score: 0.68 (0=完全随机, 1=完全确定性)

        ┌─────────┐
        │  E2E层   │  ← 少量真实LLM调用，验证端到端流程
        │ (真实API) │
        ├─────────┤
        │ 集成层    │  ← Mock LLM + 真实工具，验证编排逻辑
        │(VCR录制) │
        ├─────────┤
        │ 单元层    │  ← FakeLLM + Mock工具，验证单个Agent逻辑
        │(完全Mock) │
        └─────────┘

# pytest + llm-mock: 录制一次，永久回放
import pytest
from llm_mock import vcr

@vcr.use_cassette("fixtures/research_agent.yaml")
def test_research_agent_outputs_structure():
    """首次运行录制真实LLM响应，后续运行精确回放"""
    agent = ResearchAgent(model="gpt-4")
    result = agent.run("What are the latest multi-agent frameworks?")

    assert result.has_section("summary")
    assert result.has_section("sources")
    assert len(result.sources) >= 3

# FakeLLM: 完全确定性测试
def test_router_agent_with_fake_llm():
    from tests.fakes import FakeLLM

    fake = FakeLLM(responses=[
        '{"action": "research", "confidence": 0.9}',
        '{"action": "write", "confidence": 0.85}',
    ])

    router = RouterAgent(llm=fake)
    assert router.route("研究AI Agent") == "research"
    assert router.route("写一篇关于AI的文章") == "write"

# LangSmith的Annotate & Replay模式
from langsmith import Client

client = Client()

# 找到失败的Trace
failed_traces = client.list_runs(
    project_name="multi-agent-research",
    filter="eq(status, 'error')",
)

for trace in failed_traces:
    # 查看第3个Agent节点收到的上下文
    agent_3_input = trace.child_runs[2].inputs
    print(f"Agent 3 saw: {agent_3_input}")

    # 基于原始输入重新实验（无需重跑整个工作流）
    new_prompt = f"{agent_3_input['prompt']}\n\nIMPORTANT: Always cite sources."
    # 直接发送给LLM测试修改效果

# foundry-workflow.yaml
name: research-pipeline
version: "1.2.0"  # 版本锁定

agents:
  researcher:
    model: gpt-4.1
    instructions: |
      You are a research agent. Find 5 key sources on the given topic.
      Always return structured JSON with title, url, and summary fields.
    tools: [bing_search]

  writer:
    model: gpt-4.1
    instructions: |
      Write a 500-word article based on research results.
      Maintain factual accuracy from the provided sources.
    tools: []

  reviewer:
    model: o3-mini
    instructions: |
      Review the article for accuracy, coherence, and completeness.
      Return a score from 1-10 and specific improvement suggestions.
    tools: []

workflow:
  - agent: researcher
    input: ${trigger.topic}
    output: research_data

  - agent: writer
    input: ${research_data}
    output: draft

  - agent: reviewer
    input: ${draft}
    output: review

  # 条件分支：评分低于7则重写
  - condition: ${review.score} < 7
    agent: writer
    input: ${draft} + ${review.suggestions}
    output: final_article

graph TB
    subgraph "可复现工程六层栈"
        direction TB

        subgraph "第6层: CI/CD门控"
            CI[CI Pipeline]
            CI --> |MAESTRO评估| PASS{质量门控}
            PASS --> |通过| DEPLOY[部署]
            PASS --> |不通过| REVERT[自动回滚]
        end

        subgraph "第5层: E2E测试"
            E2E[端到端冒烟测试]
            E2E --> |真实LLM| VALIDATE[输出验证]
        end

        subgraph "第4层: 集成测试"
            INT[VCR录制回放]
            INT --> |Mock LLM + 真实工具| ORCHESTRATION[编排验证]
        end

        subgraph "第3层: 单元测试"
            UNIT[FakeLLM + Mock工具]
            UNIT --> |确定性测试| AGENT_LOGIC[Agent逻辑]
        end

        subgraph "第2层: 运行时持久化"
            TEMPORAL[Temporal检查点]
            AGENT_GIT[AgentGit快照]
            TEMPORAL --> |事件溯源| REPLAY[确定性重放]
            AGENT_GIT --> |分支/回滚| VERSION[版本控制]
        end

        subgraph "第1层: Trace录制"
            LANGSMITH[LangSmith Trace]
            LANGFUSE[Langfuse Trace]
            LANGSMITH --> |Diff对比| DEBUG[调试]
            LANGFUSE --> |回放分析| DEBUG
        end
    end

    style TEMPORAL fill:#e1f5fe
    style AGENT_GIT fill:#e1f5fe
    style LANGSMITH fill:#fff3e0
    style LANGFUSE fill:#fff3e0
    style UNIT fill:#e8f5e9
    style INT fill:#e8f5e9
    style E2E fill:#e8f5e9
    style CI fill:#fce4ec

graph TD
    START[需要Agent可复现性] --> Q1{工作流是否需要<br>长时间运行?}

    Q1 --> |是| TEMPORAL_PATH[选Temporal持久化执行]
    Q1 --> |否| Q2{技术栈是否<br>基于LangGraph?}

    Q2 --> |是| AGENTGIT_PATH[选AgentGit版本控制]
    Q2 --> |否| Q3{是否需要<br>企业级合规?}

    Q3 --> |是| FOUNDRY_PATH[选Foundry YAML工作流]
    Q3 --> |否| Q4{主要痛点是什么?}

    Q4 --> |调试困难| TRACE_PATH[选LangSmith/Langfuse]
    Q4 --> |质量不稳定| MAESTRO_PATH[选MAESTRO+测试金字塔]
    Q4 --> |两者都有| COMBO[组合使用]

    TEMPORAL_PATH --> ADD1[+ MAESTRO评估]
    AGENTGIT_PATH --> ADD2[+ VCR测试]
    FOUNDRY_PATH --> ADD3[+ Foundry内置CI]
    TRACE_PATH --> ADD4[+ 单元测试层]
    MAESTRO_PATH --> ADD5[+ LangSmith Trace]
    COMBO --> FULL[全栈六层]