简单理解：
  Agent = 给你目标，你自己想办法完成
  Workflow = 按这个流程走，每步做什么我都规定好了

人工确认输入 → AI 生成 → 自动执行

AI 生成 → 人工审核 → 确认后执行

AI 生成 → 自动执行 → 定期抽样审核 → 统计合格率

# 超时降级策略示例
TIMEOUT_MS = 30_000  # 30 秒超时

async def handle_request_with_fallback(request):
    try:
        # 主路径：AI 处理
        result = await asyncio.wait_for(
            ai_process(request),
            timeout=TIMEOUT_MS / 1000
        )
        return result
    except asyncio.TimeoutError:
        # 降级 1：返回预设的兜底回复
        metrics.record("ai_timeout")
        return {
            "type": "fallback",
            "message": "我们正在处理您的请求，请稍候...",
            "ticket_id": create_ticket(request),  # 创建工单后续处理
        }
    except ExternalServiceError as e:
        # 降级 2：外部服务不可用，切人工
        metrics.record("external_service_error")
        return route_to_human(request, reason=str(e))

置信度 >= 0.9  → AI 自动处理
置信度 0.7~0.9 → AI 生成 + 人工审核
置信度 < 0.7   → 直接转人工 + AI 建议作为参考

from temporalio.client import Client
from temporalio.worker import Worker
from temporalio import activity, workflow

@activity.defn
async def classify_ticket(ticket: dict) -> dict:
    """调用 LLM 对工单分类"""
    response = await llm_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": f"分类此工单: {ticket['body']}"}]
    )
    return {"category": response.choices[0].message.content, "confidence": 0.92}

@activity.defn
async def generate_reply(ticket: dict, customer_data: dict) -> str:
    """基于客户资料和工单内容生成回复"""
    ...

@activity.defn
async def send_email(to: str, subject: str, body: str) -> bool:
    """发送邮件"""
    ...

@workflow.defn
class TicketProcessingWorkflow:
    @workflow.run
    async def run(self, ticket: dict) -> str:
        # Step 1: 分类
        classification = await workflow.execute_activity(
            classify_ticket, ticket, start_to_close_timeout=timedelta(seconds=30)
        )

        # Step 2: 置信度检查
        if classification["confidence"] < 0.7:
            await workflow.execute_activity(escalate_to_human, ticket)
            return "escalated"

        # Step 3: 查询客户资料
        customer_data = await workflow.execute_activity(fetch_customer, ticket["customer_id"])

        # Step 4: 生成回复（等待人工审核）
        reply = await workflow.execute_activity(generate_reply, ticket, customer_data)

        # Step 5: 等待人工审核（Human-in-the-Loop）
        approved = await workflow.execute_activity(wait_for_human_approval, reply)
        if not approved:
            return "rejected"

        # Step 6: 发送邮件
        await workflow.execute_activity(send_email, ticket["email"], "回复", reply)
        return "completed"

# Temporal Worker HPA 配置
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: temporal-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: temporal-worker
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Pods
    pods:
      metric:
        name: temporal_task_queue_depth
      target:
        type: AverageValue
        averageValue: "100"

面试官：设计一个 AI 工单处理系统，日均 10 万工单。

1. 需求分析（1 分钟）
   → 日均 10 万，峰值约 5 万/小时
   → 工单类型：咨询(50%)、投诉(20%)、退款(15%)、技术支持(15%)
   → 目标：减少人工处理量 60%，保证客户满意度不降

2. 架构设计（2 分钟）
   → 流程：接收 → 分类 → 处理 → 审核 → 发送 → 归档
   → 用 Temporal 做工作流编排（需要长期运行、失败重试、状态持久化）
   → Human-in-the-Loop：分类置信度 `<` 0.7 转人工，回复需审核后发送

3. 关键设计决策（2 分钟）
   → 超时处理：AI 30s 超时 → 降级为预设回复 + 创建人工工单
   → 回退规则：基于置信度动态路由（全自动/审核/全人工）
   → 数据飞轮：人工修改的回复用于模型微调，每周更新一次

4. 运维和监控（1 分钟）
   → 技术指标：分类准确率、响应延迟、Worker 队列深度
   → 业务指标：自动处理率、人工审核通过率、客户满意度
   → 告警：自动处理率跌破 50% 触发告警

5. 扩展性（30 秒）
   → Worker 水平扩展，按工单类型分队列
   → 多模型降级策略：主模型 GPT-4o，备用 Claude，兜底规则引擎