经典模式回潮：Agent六栈实测

# ESAA风格的事件溯源实现 - 核心事件定义
from dataclasses import dataclass
from datetime import datetime
from typing import Any
import json

@dataclass
class AgentEvent:
    event_id: str
    aggregate_id: str       # Agent实例ID
    event_type: str         # cognition | decision | tool_call | tool_result | state_change
    timestamp: datetime
    payload: dict
    cognitive_context: dict  # LLM的推理上下文（为什么做这个决定）

class AgentEventStore:
    """Agent事件存储 - 支持回放和审计"""

    def append(self, event: AgentEvent):
        # 追加写入，不可修改
        self._store.append(event)

    def reconstruct_state(self, agent_id: str, up_to: datetime = None) -> dict:
        """从事件流重建Agent在任意时刻的完整状态"""
        state = {}
        for event in self._store.query(agent_id, up_to=up_to):
            if event.event_type == "state_change":
                state.update(event.payload)
            elif event.event_type == "tool_call":
                state["last_tool"] = event.payload["tool_name"]
        return state

    def audit_trail(self, agent_id: str) -> list:
        """导出完整审计轨迹：每个决策+推理过程"""
        return [
            {
                "decision": e.payload,
                "reasoning": e.cognitive_context,
                "timestamp": e.timestamp
            }
            for e in self._store.query(agent_id)
            if e.event_type in ("decision", "tool_call")
        ]

# Katha风格的Saga编排 - 多Agent旅行预订
from katha import Saga, Step, Compensation

travel_saga = Saga("book-trip")

@travel_saga.step(
    compensate=lambda ctx: cancel_hotel(ctx["hotel_id"])
)
async def book_hotel(ctx):
    hotel_id = await hotel_agent.book(
        destination=ctx["destination"],
        dates=ctx["dates"]
    )
    return {"hotel_id": hotel_id}

@travel_saga.step(
    compensate=lambda ctx: cancel_flight(ctx["flight_booking_id"])
)
async def book_flight(ctx):
    booking = await flight_agent.search_and_book(
        origin=ctx["origin"],
        destination=ctx["destination"],
        dates=ctx["dates"]
    )
    return {"flight_booking_id": booking.id}

@travel_saga.step(
    compensate=lambda ctx: refund_payment(ctx["payment_id"])
)
async def process_payment(ctx):
    total = ctx["hotel_cost"] + ctx["flight_cost"]
    payment = await payment_agent.charge(
        amount=total,
        method=ctx["payment_method"]
    )
    return {"payment_id": payment.id}

# 执行 - 任何步骤失败自动补偿
result = await travel_saga.run({
    "origin": "Shanghai",
    "destination": "Tokyo",
    "dates": ("2026-07-01", "2026-07-05"),
    "payment_method": "card"
})

// Akka Agents - 声明式Agent定义 (Java SDK)
import akka.javasdk.agent.Agent;
import akka.javasdk.agent.Effects;

public class ResearchAgent extends Agent {

    private final String systemPrompt = """
        You are a research analyst agent. Analyze the given topic,
        search for relevant information, and produce a structured report.
        """;

    @Override
    public Effects.Effect<String> onMessage(String message) {
        return effects()
            .systemPrompt(systemPrompt)
            .tools("web_search", "document_analyzer")  // MCP工具
            .thenReply(response -> response);
    }

    // 自动暴露为HTTP端点
    // POST /agents/research
    // 自动暴露为MCP工具
    // MCP tool: research_agent
}

# CQRS for AI Agents - 读写分离模式
class AgentCommandHandler:
    """命令侧：处理Agent的决策和动作"""

    def __init__(self):
        self.firestore = FirestoreClient()  # 低延迟写入
        self.event_bus = EventBus()          # 发布事件

    async def handle_decision(self, agent_id: str, decision: dict):
        # 1. 写入命令（Firestore保证强一致）
        await self.firestore.collection("agent_commands").add({
            "agent_id": agent_id,
            "decision": decision,
            "timestamp": datetime.utcnow()
        })
        # 2. 发布事件触发读侧更新
        await self.event_bus.publish("agent.decision.made", {
            "agent_id": agent_id,
            "decision_type": decision["type"],
            "result": decision["result"]
        })

class AgentQueryHandler:
    """查询侧：优化过的Agent状态视图"""

    def __init__(self):
        self.bigquery = BigQueryClient()  # 分析型查询
        self.cache = RedisClient()          # 热数据缓存

    async def get_agent_status(self, agent_id: str) -> dict:
        # 从缓存或物化视图读取（毫秒级）
        cached = await self.cache.get(f"agent_status:{agent_id}")
        if cached:
            return cached
        # 回退到BigQuery物化视图
        result = await self.bigquery.query(
            "SELECT * FROM agent_status_materialized WHERE agent_id = @aid",
            params={"aid": agent_id}
        )
        await self.cache.set(f"agent_status:{agent_id}", result, ttl=60)
        return result

# Temporal + Agent工作流 - 持久化多Agent协作
from temporalio import workflow, activity
from datetime import timedelta

@activity.defn
async def call_research_agent(topic: str) -> dict:
    """可中断、可重试的Agent调用"""
    return await research_agent.run(topic)

@activity.defn
async def call_writer_agent(research: dict) -> str:
    return await writer_agent.run(research)

@activity.defn
async def call_reviewer_agent(draft: str) -> dict:
    return await reviewer_agent.run(draft)

@workflow.defn
class ArticleProductionWorkflow:
    """文章生产流水线 - 保证不丢不重"""

    @workflow.run
    async def run(self, topic: str) -> str:
        # Step 1: 研究 - 最多重试3次，每次超时5分钟
        research = await workflow.execute_activity(
            call_research_agent,
            topic,
            start_to_close_timeout=timedelta(minutes=5),
            retry_policy=RetryPolicy(maximum_attempts=3)
        )

        # Step 2: 写作
        draft = await workflow.execute_activity(
            call_writer_agent,
            research,
            start_to_close_timeout=timedelta(minutes=10),
            retry_policy=RetryPolicy(maximum_attempts=2)
        )

        # Step 3: 审核（可挂起等待人工审批）
        review = await workflow.execute_activity(
            call_reviewer_agent,
            draft,
            start_to_close_timeout=timedelta(minutes=15),
        )

        if review["approved"]:
            return draft
        else:
            # 不通过 - 自动回到Step 2带修改意见重写
            return await self.run_with_feedback(topic, review["feedback"])

graph TB
    subgraph Event Bus
        EB[Kafka / NATS / Redis Streams]
    end

    subgraph Producer Agents
        A1[研究Agent]
        A2[数据Agent]
    end

    subgraph Consumer Agents
        A3[写作Agent]
        A4[审核Agent]
        A5[发布Agent]
    end

    subgraph Event Store
        ES[事件溯源存储]
    end

    A1 -->|research.completed| EB
    A2 -->|data.available| EB
    EB -->|research.completed| A3
    EB -->|data.available| A3
    A3 -->|draft.ready| EB
    EB -->|draft.ready| A4
    A4 -->|review.approved| EB
    EB -->|review.approved| A5
    EB -->|all events| ES

    style EB fill:#f9f,stroke:#333,stroke-width:2px
    style ES fill:#bbf,stroke:#333,stroke-width:2px

graph LR
    subgraph "经典模式层"
        ES[Event Sourcing<br/>事件溯源]
        SG[Saga<br/>分布式事务]
        AC[Actor Model<br/>并发原语]
        CQ[CQRS<br/>读写分离]
        DE[Durable Execution<br/>持久执行]
        ED[Event-Driven<br/>事件驱动]
    end

    subgraph "Agent场景层"
        AUDIT[审计合规]
        COMP[失败补偿]
        SCALE[弹性伸缩]
        PERF[性能优化]
        RELI[可靠执行]
        DECOUP[解耦通信]
    end

    subgraph "工具/框架层"
        ESAA[ESAA]
        KATHA[Katha/agent-saga]
        AKKA[Akka Agents]
        BQ[Firestore+BigQuery]
        TEMP[Temporal]
        KAFKA[Kafka/NATS]
    end

    ES --> AUDIT
    SG --> COMP
    AC --> SCALE
    CQ --> PERF
    DE --> RELI
    ED --> DECOUP

    AUDIT --> ESAA
    COMP --> KATHA
    SCALE --> AKKA
    PERF --> BQ
    RELI --> TEMP
    DECOUP --> KAFKA

    style ES fill:#e1f5fe
    style SG fill:#fff3e0
    style AC fill:#e8f5e9
    style CQ fill:#fce4ec
    style DE fill:#f3e5f5
    style ED fill:#e0f2f1