工作流与 AI Agent 编排

确定性（工作流控制）          非确定性（模型决策）
---------------------------  ---------------------------
步骤顺序与分支条件            自然语言理解与意图识别
重试与超时策略                内容生成与摘要
数据校验与格式转换            工具选择的优先级判断
权限与安全校验                模糊输入的分类
持久化与幂等                  开放式问题的回答

# 反模式：让模型决定下一步走哪个节点
next_step = llm.invoke(f"根据订单 {order} 决定下一步是 A 还是 B")
if next_step == "A":
    do_a()

# 正例：规则判断分支，模型只做内容理解
category = classify_by_rules(order)      # 确定性
if category == "NEEDS_REVIEW":
    summary = llm.summarize(order.raw_text)   # 非确定性，但边界清晰

@workflow.defn
class ResearchWorkflow:
    @workflow.run
    async def run(self, topic: str) -> Report:
        docs = await workflow.execute_activity(fetch_docs, topic)      # 确定性
        draft = await workflow.execute_activity(run_agent_loop, docs)  # 内层 Agent 循环
        await workflow.execute_activity(validate_report, draft)        # 确定性校验
        approved = await workflow.wait_condition(lambda: self.approved)  # 人工确认
        return await workflow.execute_activity(archive, draft)

TOOLS = {
    # 只读工具：可重试，无副作用
    "search_docs": Tool(fn=search_docs, read_only=True),
    "query_db": Tool(fn=query_db, read_only=True),
    # 写工具：必须幂等，需要审批
    "send_email": Tool(fn=send_email, read_only=False, requires_approval=True),
    "create_ticket": Tool(fn=create_ticket, read_only=False, requires_approval=True),
}

@activity.defn
async def send_email(order_id: str, recipient: str, body: str) -> str:
    # 幂等键由工作流生成并传入，重试时不变
    idem_key = f"email:{order_id}"
    if await email_log.exists(idem_key):
        return await email_log.get_result(idem_key)
    msg_id = await smtp.send(recipient, body)
    await email_log.save(idem_key, msg_id)
    return msg_id

// 反模式：在 Workflow 里直接调 LLM
public void execute(TaskInput input) {
    String result = llmClient.complete(input.prompt());  // 破坏确定性
    activities.save(result);
}

// 正例：LLM 调用在 Activity 里
public void execute(TaskInput input) {
    String result = activities.callLlm(input.prompt());
    activities.save(result);
}

@ActivityMethod
public String callLlm(String prompt) {
    String content = llmClient.complete(prompt);
    String key = blobStore.put(content);      // 大内容存对象存储
    return key;                               // 历史里只存 key
}

@workflow.defn
class AgentWithApproval:
    def __init__(self):
        self.approved = None

    @workflow.signal
    def approve(self, decision: bool, comment: str = ""):
        self.approved = (decision, comment)

    @workflow.run
    async def run(self, task: str):
        plan = await workflow.execute_activity(agent_plan, task)
        await workflow.execute_activity(notify_reviewer, plan)

        # 最多等 24 小时，超时视为拒绝
        ok = await workflow.wait_condition(
            lambda: self.approved is not None,
            timeout=timedelta(hours=24))
        if not ok or not self.approved[0]:
            return {"status": "REJECTED"}

        result = await workflow.execute_activity(agent_execute, plan)
        return {"status": "DONE", "result": result}

agent_workflow:
  total_budget:
    max_steps: 20
    max_tokens: 200000
    max_duration: 10m
  steps:
    classify:
      model: gpt-4o-mini
      max_tokens: 200
      cache: true
    summarize:
      model: gpt-4o
      max_tokens: 4000
      cache: true
    execute:
      model: gpt-4o
      max_tokens: 8000
      cache: false        # 有副作用的步骤不缓存

# 做法一：把 LLM 输出先持久化，副作用基于持久化结果执行
analysis = await workflow.execute_activity(llm_analyze, doc)   # 结果进历史
# 重放时直接读历史，不会重新分析，因此后续副作用也不会重复

# 做法二：副作用带幂等键
await workflow.execute_activity(
    send_notification,
    NotificationArgs(idem_key=f"notify:{task_id}", content=analysis))

# 做法三：把 LLM 输出当决策，用状态前置条件保护
await workflow.execute_activity(
    apply_decision,
    DecisionArgs(task_id=task_id, decision=analysis.decision))
# apply_decision 内部用「WHERE status='PENDING'」保证只生效一次

@workflow.defn
class OrchestratorWorkflow:
    @workflow.run
    async def run(self, goal: str):
        subtasks = await workflow.execute_activity(plan_subtasks, goal)
        # 并行执行子任务
        handles = [
            workflow.start_child_workflow(
                WorkerWorkflow.run, st, id=f"{goal}-{i}")
            for i, st in enumerate(subtasks)
        ]
        results = [await h for h in handles]
        return await workflow.execute_activity(synthesize, results)

@activity.defn
async def llm_with_fallback(prompt: str) -> str:
    for model in ["gpt-4o", "gpt-4o-mini", "local-model"]:
        try:
            return await call_llm(model, prompt)
        except RateLimitError:
            await asyncio.sleep(backoff())
        except (TimeoutError, ServiceUnavailable):
            continue          # 降级到下一个模型
        except ContentFilterError:
            raise             # 内容问题不降级，直接上抛
    raise AllModelsFailed("所有模型均不可用")

# 评估集：输入 + 期望的关键特征（不是精确输出）
EVAL_CASES = [
    {"input": "退款申请：订单 1001，原因商品破损",
     "expect": {"category": "REFUND", "requires_approval": True}},
    {"input": "咨询：什么时候发货",
     "expect": {"category": "INQUIRY", "requires_approval": False}},
]

async def run_eval(pipeline) -> EvalReport:
    results = []
    for case in EVAL_CASES:
        out = await pipeline(case["input"])
        results.append({
            "case": case,
            "passed": matches(out, case["expect"]),
        })
    return EvalReport(results)

# prompts/refund-classify/v3.yaml
version: v3
model: gpt-4o-mini
temperature: 0
system: |
  你是退款申请分类器。只输出 JSON，不要解释。
  字段：category（REFUND/INQUIRY/COMPLAINT）、requires_approval（bool）
user: |
  申请内容：{{ text }}
  订单金额：{{ amount }}

result = await workflow.execute_activity(
    llm_classify,
    ClassifyArgs(text=text, prompt_version="refund-classify/v3"))

def execute_tool(name: str, args: dict, allowed: set[str], principal: str):
    if name not in allowed:
        raise PermissionError(f"tool {name} not allowed in this step")
    tool = TOOLS[name]
    validated = tool.schema.validate(args)      # 参数结构校验
    if not tool.authorize(principal, validated):  # 业务权限校验
        raise PermissionError(f"principal {principal} cannot {name}")
    return tool.fn(validated)

# 每个步骤的 token 消耗
agent_tokens_total{workflow="RefundFlow", step="classify", model="gpt-4o-mini", direction="input"} 125000
agent_tokens_total{workflow="RefundFlow", step="classify", model="gpt-4o-mini", direction="output"} 8000

# 每个工作流实例的成本
agent_cost_usd_sum{workflow="RefundFlow"} 42.17

# 质量指标
agent_eval_pass_rate{workflow="RefundFlow", eval_set="core"} 0.94
agent_fallback_total{workflow="RefundFlow", from="gpt-4o", to="gpt-4o-mini"} 12
agent_human_review_total{workflow="RefundFlow", reason="content_filter"} 3