Agent 执行循环与状态机设计：从 ReAct 到确定性控制流

messages = [SystemMessage, UserPrompt]

while not is_done:
    response = llm.chat(messages, tools=tools)
    if response.has_tool_calls():
        for call in response.tool_calls:
            result = execute_tool(call.name, call.args)
            messages.append(ToolMessage(call_id=call.id, content=result))
    else:
        is_done = True
        return response.content

                   ┌───────────────────────────────────────────────┐
                   │               LLM Tool Call                   │
                   └───────────────────────┬───────────────────────┘
                                           ▼
                   ┌───────────────────────────────────────────────┐
                   │  Layer 1: Deterministic Schema Validation    │
                   │  (Pydantic V2 / JSON Schema 静态校验拦截)       │
                   └───────┬───────────────────────────────┬───────┘
                           │ 合法                          │ 格式非法
                           ▼                               ▼
      ┌────────────────────────────────────┐     ┌─────────────────┐
      │  Layer 2: Circuit Breaker Inspection│     │ Inject Schema   │
      │  (检查当前 Tool 是否处于熔断拦截状态) │     │ Error Hint      │
      └───────┬────────────────────┬───────┘     └────────┬────────┘
              │ 正常 (Closed)       │ 熔断 (Open)          │ (返回自愈)
              ▼                    ▼                      ▼
      ┌───────────────┐    ┌─────────────────┐   ┌─────────────────┐
      │ Execute Tool  │    │ Route to        │   │ Retry in Current│
      │ Real API Call │    │ Fallback Tool   │   │ Node (Max 2)    │
      └───────┬───────┘    └─────────────────┘   └─────────────────┘
              │ 异常
              ▼
      ┌────────────────────────────────────┐
      │ Increment Failure Counter          │
      │ If failures >= 3 -> Open Circuit   │
      └────────────────────────────────────┘

"""
deterministic_agent_fsm.py
工业级确定性 Agent 状态机引擎（包含熔断器、严格参数校验与状态隔离）
"""

import json
import logging
from enum import Enum
from typing import Any, Callable, Dict, List, Optional
from pydantic import BaseModel, Field, ValidationError

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("AgentFSM")


# ==========================================================
# 1. 核心状态与数据模型定义
# ==========================================================

class NodeState(str, Enum):
    IDLE = "IDLE"
    PLANNING = "PLANNING"
    ACTION_SELECT = "ACTION_SELECT"
    TOOL_EXECUTION = "TOOL_EXECUTION"
    VERIFICATION = "VERIFICATION"
    FALLBACK = "FALLBACK"
    TERMINAL_SUCCESS = "TERMINAL_SUCCESS"
    TERMINAL_FAILED = "TERMINAL_FAILED"


class SearchDatabaseArgs(BaseModel):
    query: str = Field(..., min_length=2, description="搜索关键词")
    limit: int = Field(default=5, ge=1, le=50, description="最大返回条数，1-50之间")


class ToolResult(BaseModel):
    success: bool
    data: Optional[Any] = None
    error_message: Optional[str] = None


class AgentMemory(BaseModel):
    user_goal: str
    current_step: int = 0
    max_steps: int = 10
    scratchpad: List[Dict[str, Any]] = Field(default_factory=list)
    final_output: Optional[str] = None
    circuit_failures: Dict[str, int] = Field(default_factory=dict)
    circuit_max_failures: int = 3


# ==========================================================
# 2. 模拟真实环境与熔断工具
# ==========================================================

class ExternalServiceSimulator:
    """模拟外部数据库或 API，可注入故障以验证熔断逻辑"""
    def __init__(self, failure_trigger_count: int = 3):
        self.call_count = 0
        self.failure_trigger_count = failure_trigger_count

    def search_database(self, query: str, limit: int) -> Dict[str, Any]:
        self.call_count += 1
        # 模拟外部服务前 3 次持续故障，触发熔断
        if self.call_count <= self.failure_trigger_count:
            raise ConnectionError(f"Database cluster connection timeout (Attempt {self.call_count})")
        return {"status": "ok", "items": [f"Result #{i} for '{query}'" for i in range(1, limit + 1)]}


# ==========================================================
# 3. 确定性有限状态机执行引擎
# ==========================================================

class DeterministicAgentFSM:
    def __init__(self, user_goal: str, db_service: ExternalServiceSimulator):
        self.memory = AgentMemory(user_goal=user_goal)
        self.db_service = db_service
        self.current_state = NodeState.IDLE
        self.history: List[str] = []

    def transition_to(self, next_state: NodeState, reason: str = ""):
        log_msg = f"Transition: [{self.current_state.value}] -> [{next_state.value}]"
        if reason:
            log_msg += f" (Reason: {reason})"
        logger.info(log_msg)
        self.history.append(f"{self.current_state.value} -> {next_state.value}")
        self.current_state = next_state

    def is_circuit_open(self, tool_name: str) -> bool:
        failures = self.memory.circuit_failures.get(tool_name, 0)
        return failures >= self.memory.circuit_max_failures

    def record_tool_failure(self, tool_name: str):
        self.memory.circuit_failures[tool_name] = self.memory.circuit_failures.get(tool_name, 0) + 1
        logger.warning(f"Tool [{tool_name}] failure count: {self.memory.circuit_failures[tool_name]}/{self.memory.circuit_max_failures}")

    def reset_tool_failure(self, tool_name: str):
        if tool_name in self.memory.circuit_failures:
            self.memory.circuit_failures[tool_name] = 0

    # ---------------- 节点执行逻辑 ----------------

    def step_idle(self):
        logger.info(f"Received goal: '{self.memory.user_goal}'")
        self.transition_to(NodeState.PLANNING, "Goal initialized")

    def step_planning(self):
        # 模拟基于任务目标进行确定性分解规划
        self.memory.current_step += 1
        logger.info("Decomposing goal into verifiable steps...")
        # 确定需要检索目标库
        self.transition_to(NodeState.ACTION_SELECT, "Plan ready")

    def step_action_select(self):
        self.memory.current_step += 1
        if self.memory.current_step > self.memory.max_steps:
            self.transition_to(NodeState.FALLBACK, "Step limit exceeded")
            return

        # 检查工具是否已被熔断
        if self.is_circuit_open("search_database"):
            logger.error("Circuit breaker is OPEN for 'search_database'. Denying execution.")
            self.transition_to(NodeState.FALLBACK, "Tool circuit breaker open")
            return

        # 假装大模型做出了工具调用决策，输入了待校验参数
        mock_model_raw_args = {"query": "production_incident_report", "limit": 3}
        
        # Layer 1: 严格 Pydantic 校验拦截
        try:
            validated_args = SearchDatabaseArgs(**mock_model_raw_args)
            # 暂存工作记忆
            self.memory.scratchpad.append({
                "action": "search_database",
                "args": validated_args.model_dump()
            })
            self.transition_to(NodeState.TOOL_EXECUTION, "Schema validation passed")
        except ValidationError as e:
            logger.error(f"Schema validation rejected: {e.errors()}")
            # 本地自愈或直接拒绝
            self.transition_to(NodeState.PLANNING, "Schema repair requested")

    def step_tool_execution(self):
        last_action = self.memory.scratchpad[-1]
        tool_name = last_action["action"]
        args = last_action["args"]

        logger.info(f"Executing tool '{tool_name}' with args: {args}")
        try:
            # 实际调用外部服务
            raw_response = self.db_service.search_database(args["query"], args["limit"])
            self.reset_tool_failure(tool_name)
            last_action["result"] = ToolResult(success=True, data=raw_response)
            self.transition_to(NodeState.VERIFICATION, "Tool executed successfully")
        except Exception as err:
            logger.error(f"Execution failure: {err}")
            self.record_tool_failure(tool_name)
            last_action["result"] = ToolResult(success=False, error_message=str(err))
            
            # 检查是否刚触发了熔断
            if self.is_circuit_open(tool_name):
                self.transition_to(NodeState.FALLBACK, f"Tool [{tool_name}] reached max failure threshold")
            else:
                self.transition_to(NodeState.ACTION_SELECT, "Retry with remaining budget")

    def step_verification(self):
        # 验证节点：对工具输出进行后置契约校验
        last_action = self.memory.scratchpad[-1]
        result: ToolResult = last_action.get("result")

        if result and result.success:
            logger.info("Verification passed: Tool returned structured data.")
            self.memory.final_output = f"Successfully fetched {len(result.data['items'])} items."
            self.transition_to(NodeState.TERMINAL_SUCCESS, "Goal completed")
        else:
            logger.warning("Verification failed: Data incomplete or invalid.")
            self.transition_to(NodeState.PLANNING, "Re-planning required")

    def step_fallback(self):
        logger.info("Entered FALLBACK node. Activating degradation protocol...")
        # 降级逻辑：例如从本地离线冷备份读取数据，或向管理员通知人工介入
        self.memory.final_output = "System gracefully degraded: Returning cached incident summaries."
        self.transition_to(NodeState.TERMINAL_SUCCESS, "Recovered via Fallback degradation")

    # ---------------- 状态机主循环调度 ----------------

    def run(self) -> str:
        handlers: Dict[NodeState, Callable[[], None]] = {
            NodeState.IDLE: self.step_idle,
            NodeState.PLANNING: self.step_planning,
            NodeState.ACTION_SELECT: self.step_action_select,
            NodeState.TOOL_EXECUTION: self.step_tool_execution,
            NodeState.VERIFICATION: self.step_verification,
            NodeState.FALLBACK: self.step_fallback,
        }

        while self.current_state not in (NodeState.TERMINAL_SUCCESS, NodeState.TERMINAL_FAILED):
            handler = handlers.get(self.current_state)
            if not handler:
                raise RuntimeError(f"Unhandled FSM state: {self.current_state}")
            handler()

        logger.info(f"FSM Terminated at [{self.current_state.value}]. Final Output: {self.memory.final_output}")
        return self.memory.final_output


# ==========================================================
# 4. 运行验证：测试在持续故障下的状态机确定性熔断
# ==========================================================

if __name__ == "__main__":
    print("\n--- [测试场景 1: 模拟服务持续失败触发熔断与优雅降级] ---")
    failing_service = ExternalServiceSimulator(failure_trigger_count=3)
    agent = DeterministicAgentFSM(
        user_goal="查询最新的生产故障诊断报告",
        db_service=failing_service
    )
    result = agent.run()
    
    print("\n--- 状态迁移全链路轨迹 ---")
    for idx, transition in enumerate(agent.history, 1):
        print(f"Step {idx:02d}: {transition}")

2026-09-20 10:15:01 [INFO] Received goal: '查询最新的生产故障诊断报告'
2026-09-20 10:15:01 [INFO] Transition: [IDLE] -> [PLANNING] (Reason: Goal initialized)
2026-09-20 10:15:01 [INFO] Decomposing goal into verifiable steps...
2026-09-20 10:15:01 [INFO] Transition: [PLANNING] -> [ACTION_SELECT] (Reason: Plan ready)
2026-09-20 10:15:01 [INFO] Transition: [ACTION_SELECT] -> [TOOL_EXECUTION] (Reason: Schema validation passed)
2026-09-20 10:15:01 [INFO] Executing tool 'search_database' with args: {'query': 'production_incident_report', 'limit': 3}
2026-09-20 10:15:01 [ERROR] Execution failure: Database cluster connection timeout (Attempt 1)
2026-09-20 10:15:01 [WARNING] Tool [search_database] failure count: 1/3
2026-09-20 10:15:01 [INFO] Transition: [TOOL_EXECUTION] -> [ACTION_SELECT] (Reason: Retry with remaining budget)
2026-09-20 10:15:01 [INFO] Transition: [ACTION_SELECT] -> [TOOL_EXECUTION] (Reason: Schema validation passed)
2026-09-20 10:15:01 [ERROR] Execution failure: Database cluster connection timeout (Attempt 2)
2026-09-20 10:15:01 [WARNING] Tool [search_database] failure count: 2/3
2026-09-20 10:15:01 [INFO] Transition: [TOOL_EXECUTION] -> [ACTION_SELECT] (Reason: Retry with remaining budget)
2026-09-20 10:15:01 [INFO] Transition: [ACTION_SELECT] -> [TOOL_EXECUTION] (Reason: Schema validation passed)
2026-09-20 10:15:01 [ERROR] Execution failure: Database cluster connection timeout (Attempt 3)
2026-09-20 10:15:01 [WARNING] Tool [search_database] failure count: 3/3
2026-09-20 10:15:01 [INFO] Transition: [TOOL_EXECUTION] -> [FALLBACK] (Reason: Tool [search_database] reached max failure threshold)
2026-09-20 10:15:01 [INFO] Entered FALLBACK node. Activating degradation protocol...
2026-09-20 10:15:01 [INFO] Transition: [FALLBACK] -> [TERMINAL_SUCCESS] (Reason: Recovered via Fallback degradation)