AI Agent 的确定性工程：当 LLM 遇见可靠性要求

from pydantic import BaseModel, Field

class OrderStatusResponse(BaseModel):
    order_id: str = Field(description="订单号")
    status: Literal["pending", "processing", "shipped", "delivered", "cancelled"]
    estimated_delivery: str | None = Field(default=None, description="预计送达时间")
    actions: list[str] = Field(
        description="用户可执行的操作列表",
        default_factory=list
    )

# 结构化输出减少了理解偏差
response = model.with_structured_output(OrderStatusResponse)
result = response.invoke("帮我查下订单 ORD-2026-0412 的状态")
# result.order_id 永远是字符串
# result.status 永远是枚举中的五个值之一

# 差的情况：只给描述，让模型自由理解
prompt = """
判断用户情感的极性：正面、负面、中性。
用户输入：我对这个功能有点失望。
"""

# 好的情况：给明确的示例覆盖边缘case
prompt = """
判断用户情感的极性：正面、负面、中性。

示例：
输入：「太棒了，完全符合预期」 → 正面
输入：「有点贵」 → 中性（不是负面，只是陈述价格）
输入：「这东西简直是垃圾，浪费我时间」 → 负面
输入：「我有点失望」 → 负面（「有点」表达了明确的负面情绪）

判断：
输入：「说实话，这个功能还算能用」 →
"""

DECISION_FRAMEWORK = """
对于每个候选方案，按以下维度打分（1-5分）：
1. 合规风险：是否可能违反公司政策或法律法规？
2. 用户价值：对用户的实际帮助有多大？
3. 执行成本：开发和维护成本有多高？
4. 可逆性：如果决策错误，恢复难度有多大？

最终决策规则：
- 如果任何维度得分≤2，必须选择风险最低的方案
- 如果合规风险=5（一票否决），直接拒绝
- 否则，选择总分最高的方案

按照上述框架分析并给出最终推荐。不要跳过任何维度。
"""

from pydantic import ValidationError

try:
    result = OrderStatusResponse.model_validate(model_output)
except ValidationError as e:
    # 结构化校验失败，降级处理
    logger.warning(f"LLM output validation failed: {e}, falling back to rule-based")
    return fallback_query(order_id)

from chevron import template

SEMANTIC_CHECK_PROMPT = template(
    """你是一个订单系统的事实核查员。
    
    订单信息：{{order_info}}
    用户问题：{{user_question}}
    LLM回答：{{llm_answer}}
    
    请判断 LLM 回答是否与订单信息一致。
    如果一致，返回 {"consistent": true}
    如果不一致，返回 {"consistent": false, "correct_answer": "正确的回答"}
    
    只返回 JSON，不要有其他内容。
    """,
    {
        "order_info": get_order_info(order_id),
        "user_question": user_input,
        "llm_answer": raw_llm_response
    }
)

# 永远不要让 LLM 决定这些
IMPOSSIBLE_RULES = {
    "discount_rate": lambda cart: 0.1 if cart.total > 1000 else 0,
    "shipping_cutoff": lambda: datetime.now().hour < 18,
    "blocked_users": lambda user_id: user_id in BLOCKED_LIST,
}

# LLM 只负责生成自然语言解释
explanation = llm.generate(f"解释为什么订单{order_id}适用此折扣")

from enum import Enum
from typing import Callable

class OrderAgentState(Enum):
    INITIAL = "initial"
    VALIDATING = "validating"
    INVENTORY_CHECK = "inventory_check"
    PAYMENT_PROCESSING = "payment_processing"
    FULFILLMENT = "fulfillment"
    CONFIRMATION = "confirmation"
    FAILED = "failed"

class StateMachineAgent:
    def __init__(self):
        self.state = OrderAgentState.INITIAL
        self.transitions: dict[OrderAgentState, list[OrderAgentState]] = {
            OrderAgentState.INITIAL: [OrderAgentState.VALIDATING],
            OrderAgentState.VALIDATING: [OrderAgentState.INVENTORY_CHECK, OrderAgentState.FAILED],
            OrderAgentState.INVENTORY_CHECK: [OrderAgentState.PAYMENT_PROCESSING, OrderAgentState.FAILED],
            OrderAgentState.PAYMENT_PROCESSING: [OrderAgentState.FULFILLMENT, OrderAgentState.FAILED],
            OrderAgentState.FULFILLMENT: [OrderAgentState.CONFIRMATION],
            OrderAgentState.CONFIRMATION: [],
            OrderAgentState.FAILED: [],
        }

    def transition(self, next_state: OrderAgentState) -> None:
        if next_state not in self.transitions[self.state]:
            raise AgentDeterminismViolationError(
                f"Invalid transition: {self.state} -> {next_state}"
            )
        self.state = next_state

    def process(self, order: Order) -> OrderResult:
        while self.state != OrderAgentState.CONFIRMATION and self.state != OrderAgentState.FAILED:
            self._execute_state(order)
        return self._generate_result(order)

PLANNING_PROMPT = """
你是一个订单处理 Agent。现在收到用户请求：「{user_request}」

请生成一个结构化的执行计划，格式如下：
1. [ACTION] <具体操作>
2. [DECISION] if <条件> then <路径A> else <路径B>
3. [CALL] <tool_name>(<参数>)
4. [RESPOND] <最终回复内容>

执行计划必须：
- 覆盖用户请求的所有分支
- 包含错误处理分支
- 每个 LLM 判断点都要有明确的决策规则

输出格式：纯文本执行计划，每行一个步骤。
"""

# LLM 生成计划
execution_plan = llm.invoke(PLANNING_PROMPT)

# 计划校验
validated_plan = plan_validator.validate(execution_plan)

# 确定性执行
result = plan_executor.execute(validated_plan)

# 不推荐：始终使用最新版
client = Anthropic()

# 推荐：锁定具体版本
client = Anthropic(
    model="claude-3-7-sonnet-20261111"  # YYYY-MM-DD 格式，锁定到具体日期的版本
)

# 更保守：同时锁定模型和推理参数
response = client.messages.create(
    model="claude-3-7-sonnet-20261111",
    temperature=0.0,  # 确定性输出必须 temperature=0
    top_p=1.0,        # 禁用 nucleus sampling
    # 不传 system 则每次行为完全一致
)

# 第一层：严格的工具描述和参数 schema
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "查询订单状态。仅用于用户明确询问订单物流、发货状态时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "pattern": "^ORD-\\d{4}-\\d{4}$",  # 正则约束参数格式
                        "description": "订单号，格式必须为 ORD-YYYY-NNNN"
                    }
                },
                "required": ["order_id"]
            }
        }
    }
]

# 第二层：调用前用规则引擎校验参数合法性
def validate_tool_call(tool_name: str, arguments: dict) -> None:
    if tool_name == "get_order_status":
        if not re.match(r"^ORD-\d{4}-\d{4}$", arguments["order_id"]):
            raise InvalidParameterError(f"Invalid order_id format: {arguments['order_id']}")
        if arguments["order_id"] in CANCELLED_ORDERS:
            raise OrderCancelledError(f"Order {arguments['order_id']} has been cancelled")

# 差的模式：Agent A 控制 Agent B 的行为（强耦合）
class AgentA:
    def delegate_to_b(self, task):
        response = agent_b.invoke(task)
        if "reject" in response:
            # Agent A 需要猜测 Agent B 的行为
            return self.handle_rejection(response)

# 好的模式：共享状态机，Agent 在预定义状态下自主行动
SHARED_STATE = {
    "current_phase": "order_validation",  # 所有 Agent 都读这个状态
    "validation_result": None,
    "approved_tools": ["get_order_status", "apply_discount"],  # 白名单
}

class OrderValidationAgent:
    def act(self, state: dict) -> dict:
        if state["current_phase"] != "order_validation":
            return {}  # 静默跳过，不是我该管的时候
        result = self.validate_order(state["order"])
        return {"validation_result": result}

class FulfillmentAgent:
    def act(self, state: dict) -> dict:
        if state["current_phase"] != "fulfillment":
            return {}  # 等待状态机推进到这里
        if not state["validation_result"]["approved"]:
            return {}  # 校验没通过，不执行
        return self.fulfill_order(state["order"])