Agent 工作流编排设计：DAG、状态机、暂停恢复与长期运行任务

from langgraph.graph import StateGraph, END
from typing import TypedDict, Any
import json

# 共享状态
class WorkflowState(TypedDict):
    input_text: str
    intent: str
    extracted_entities: dict
    api_response: Any
    final_answer: str

# 节点函数
def intent_recognition(state: WorkflowState):
    # 意图识别 Agent
    return {"intent": "book_flight"}

def entity_extraction(state: WorkflowState):
    # 实体提取
    return {"extracted_entities": {"origin": "上海", "destination": "北京"}}

def api_call(state: WorkflowState):
    # API 查询
    return {"api_response": {"flights": ["MU5161", "CA1850"]}}

def generate_response(state: WorkflowState):
    # 回复生成
    return {"final_answer": "为您找到 2 班航班..."}

# 构建工作流
builder = StateGraph(WorkflowState)
builder.add_node("intent", intent_recognition)
builder.add_node("extract", entity_extraction)
builder.add_node("api", api_call)
builder.add_node("respond", generate_response)

# 定义边（顺序执行）
builder.add_edge("intent", "extract")
builder.add_edge("extract", "api")
builder.add_edge("api", "respond")
builder.add_edge("respond", END)
builder.set_entry_point("intent")

# 编译执行
workflow = builder.compile()
result = workflow.invoke({"input_text": "帮我订上海到北京的机票"})

def route_by_intent(state):
    """根据意图动态路由"""
    return state["intent"]  # 返回字符串决定下一个节点

builder.add_conditional_edges(
    "intent",
    route_by_intent,
    {
        "book_flight": "extract",
        "check_weather": "api",
        "general_chat": "respond",
    }
)

from langgraph.graph import ANNOTATED_STATE
import operator

class ParallelState(TypedDict):
    queries: list[str]
    results: Annotated[list, operator.add]  # 并行结果累加

# researchA 和 researchB 同时运行
builder.add_node("research_a", research_agent_a)
builder.add_node("research_b", research_agent_b)
builder.add_node("synthesize", synthesis_agent)

builder.add_edge("synthesize", "synthesize")

def check_quality(state):
    if state["quality_score"] < 0.8:
        return "retry"  # → 回到 edit 节点
    return "done"      # → END

builder.add_conditional_edges(
    "evaluate",
    check_quality,
    {"retry": "edit", "done": END}
)

from langgraph.types import interrupt

async def human_review(state):
    # 暂停等待人类审核
    result = interrupt(
        value={"draft": state["draft"]},
        # 弹出 UI 让用户确认
    )
    # result 是人类返回的数据
    return {"approved": result["approved"]}

builder.add_node("human_review", human_review)
builder.add_edge("draft", "human_review")
builder.add_edge("human_review", "publish")

import uuid
from datetime import datetime

class PersistentWorkflow:
    def __init__(self, db):
        self.db = db

    async def start(self, workflow_def, initial_state):
        run_id = str(uuid.uuid4())
        await self.db.execute(
            "INSERT INTO workflow_runs (id, status, state) VALUES (?, 'running', ?)",
            (run_id, json.dumps(initial_state))
        )
        return run_id

    async def checkpoint(self, run_id, state):
        await self.db.execute(
            "UPDATE workflow_runs SET state = ?, checkpoint_time = ? WHERE id = ?",
            (json.dumps(state), datetime.utcnow().isoformat(), run_id)
        )

    async def resume(self, run_id, human_input):
        row = await self.db.fetchone(
            "SELECT * FROM workflow_runs WHERE id = ?", (run_id,)
        )
        state = json.loads(row["state"])
        # 注入人类输入并继续执行
        state["human_input"] = human_input
        return await self._run_from_checkpoint(state)

from temporalio import workflow
from temporalio.client import Client
import asyncio

# Workflow 定义
@workflow.defn
class ResearchAgentWorkflow:
    @workflow.run
    async def run(self, research_topic: str) -> str:
        # Step 1: 搜索（可中断）
        search_results = await workflow.execute_activity(
            "search_activity",
            research_topic,
            start_to_close_timeout=timedelta(minutes=1),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        # Step 2: 分析（可并行）
        analyses = await asyncio.gather(*[
            workflow.execute_activity("analysis_activity", result)
            for result in search_results
        ])

        # Step 3: 人类审核（长时间等待）
        draft = await workflow.execute_activity("synthesis_activity", analyses)
        
        # 暂停等待人类评审
        review = await workflow.execute_activity(
            "request_human_review",
            draft,
            start_to_close_timeout=timedelta(days=7),  # 人类有 7 天时间
            schedule_to_close_timeout=timedelta(days=30),
        )

        # Step 4: 修改
        if not review["approved"]:
            draft = await workflow.execute_activity(
                "revise_activity",
                {"draft": draft, "feedback": review["feedback"]},
            )

        # Step 5: 发布
        return await workflow.execute_activity("publish_activity", draft)

from temporalio.worker import Worker

async def main():
    client = await Client.connect("temporal:7233")
    
    activities = [search_activity, analysis_activity, synthesis_activity]
    
    worker = Worker(
        client,
        task_queue="agent-research-queue",
        workflows=[ResearchAgentWorkflow],
        activities=activities,
    )
    await worker.run()

from dataclasses import dataclass
from typing import Callable, List
import asyncio

@dataclass
class AgentEvent:
    event_type: str  # "order.placed", "review.pending", etc.
    payload: dict
    correlation_id: str

class EventBus:
    def __init__(self):
        self.subscribers: dict[str, List[Callable]] = {}
        self.event_queue = asyncio.Queue()

    def subscribe(self, event_type: str, handler: Callable):
        self.subscribers.setdefault(event_type, []).append(handler)

    async def emit(self, event: AgentEvent):
        await self.event_queue.put(event)

    async def run(self):
        while True:
            event = await self.event_queue.get()
            handlers = self.subscribers.get(event.event_type, [])
            for handler in handlers:
                asyncio.create_task(handler(event))

# Agent 订阅事件
bus = EventBus()

def order_agent(event: AgentEvent):
    print(f"处理订单: {event.payload}")
    # 发起支付、检查库存...

bus.subscribe("order.placed", order_agent)

# 触发事件
asyncio.run(bus.emit(AgentEvent("order.placed", {"sku": "A001", "qty": 2}, "uuid-123")))