生产级 Agent 工作流上线清单

// 粗粒度——所有写操作共享一个权限
interface AgentTools {
  read: (query: string) => Promise<unknown>
  write: (action: string, payload: unknown) => Promise<unknown>
}
 
// 权限控制只能区分读和写
if (permission === 'write') {
  await tools.write('update_address', { orderId, address })
  await tools.write('refund', { orderId, amount }) // 同一权限
}

// 细粒度——按业务对象独立授权
interface AgentTools {
  order: {
    query: (id: string) => Promise<Order>
    updateAddress: (id: string, addr: Address) => Promise<void>
  }
  refund: {
    create: (id: string, amount: number) => Promise<Refund>
    // 超过阈值需要人工确认
    createWithApproval: (id: string, amount: number) => Promise<Refund>
  }
}
 
// 每个操作独立鉴权
const refundTool = amount > THRESHOLD
  ? tools.refund.createWithApproval
  : tools.refund.create

# 静默吞错——模型可能误读错误原因
try:
    result = await call_tool("query_order", order_id="12345")
except Exception:
    # 异常被吞掉，模型不知道发生了什么
    result = {"status": "unknown"}

# 结构化重试——每次失败有分类和处理策略
class ToolFailure(Exception):
    def __init__(self, tool: str, error_type: str, retryable: bool):
        self.tool = tool
        self.error_type = error_type  # "timeout" | "permission" | "not_found" | "server_error"
        self.retryable = retryable
 
async def call_tool_with_retry(tool_name, params, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await call_tool(tool_name, params)
        except ToolFailure as e:
            trace.record(tool_name, attempt, e.error_type)
            if not e.retryable:
                raise  # 权限错误不重试，直接上报
            if attempt == max_retries - 1:
                await escalate_to_human(tool_name, params, e)
                raise
            await backoff(attempt)

# 非结构化日志——排查问题时基本靠猜
logger.info(f"Agent started for user {user_id}")
logger.info(f"Called tool: {tool_name}")
logger.info(f"Agent finished, result: {result}")
# 中间发生了什么？不知道

# 结构化 Trace——每一步可追溯
trace = Trace(task_id=task_id, user_id=user_id)
 
with trace.span("plan") as span:
    span.set("input", user_message)
    plan = await agent.plan(user_message)
    span.set("output", plan.steps)
 
for step in plan.steps:
    with trace.span("tool_call") as span:
        span.set("tool", step.tool_name)
        span.set("params", step.params)
        result = await call_tool(step.tool_name, step.params)
        span.set("result", result)
        span.set("latency_ms", elapsed)
 
trace.flush()  # 写入结构化存储，支持查询和回放

# 模糊评分——阈值难以设定，波动难以解读
def evaluate_response(response: str) -> float:
    score = 0.0
    if "correct_format" in response:
        score += 0.3
    if "has_relevant_content" in response:
        score += 0.4
    if "no_obvious_errors" in response:
        score += 0.3
    return score  # 3.7 算通过吗？没人知道
 
# 回归阈值模糊，团队经常争论
assert average_score >= 3.5  # 这个 3.5 是怎么来的？

# 二元判定——每条评测有明确的通过条件
def evaluate_response(response: str, expected: dict) -> EvalResult:
    checks = [
        ("format_valid", validate_json_schema(response, expected.schema)),
        ("contains_required_fields", all(f in response for f in expected.fields)),
        ("no_hallucinated_ids", verify_ids(response, expected.known_ids)),
        ("within_token_budget", count_tokens(response) <= expected.max_tokens),
    ]
    passed = all(result for _, result in checks)
    return EvalResult(passed=passed, checks=checks)
 
# 回归门禁清晰：任何一条不通过就阻断发布
assert all(result.passed for result in eval_results)

目标与边界
  [ ] 任务目标有明确的完成标准
  [ ] 定义了 Agent 不应该做的事情
  [ ] 业务方确认了价值指标
 
工具与权限
  [ ] 每个工具有完整的 JSON Schema
  [ ] 读写操作独立授权
  [ ] 遵循最小权限原则
  [ ] 高风险动作需人工确认
 
状态管理
  [ ] 执行状态可持久化
  [ ] 中断后可恢复
  [ ] 支持状态回放（推荐）
 
评测
  [ ] 主路径自动化评测
  [ ] 边界路径评测（空结果、超时、权限不足）
  [ ] 二元判定替代模糊评分（推荐）
  [ ] 回归评测接入 CI
 
可观测性
  [ ] LLM 调用记录 Prompt、补全、Token、费用
  [ ] 工具调用记录参数、返回值、延迟
  [ ] Trace 包含完整推理链路
  [ ] 配置了告警规则
 
失败处理
  [ ] 可重试错误有重试机制，有上限
  [ ] 不可逆操作有补偿方案
  [ ] 连续失败后主动放弃并通知人工
 
人工兜底
  [ ] 用户可随时取消任务
  [ ] 有人工接管入口
  [ ] 接管操作被记录用于改进
 
发布策略
  [ ] 先内部再灰度
  [ ] 早期限制为只读或草稿模式
  [ ] 权限扩大时同步扩大监控