状态机——Agent 的行为骨架

from enum import Enum, auto

# 1. 定义有限的状态集合 (S)
class AgentState(Enum):
    IDLE      = auto()   # 空闲等待输入
    ANALYZING = auto()   # 分析用户意图
    TOOL_CALL = auto()   # 执行外部工具
    RESPOND   = auto()   # 生成最终回复

class AgentFSM:
    def __init__(self):
        # 2. 设定初始状态 (s0)
        self.state = AgentState.IDLE

        # 3. 定义转移函数表 (Delta)
        # 格式: (当前状态, 触发事件): 目标状态
        self.transitions = {
            (AgentState.IDLE,      "user_input"): AgentState.ANALYZING,
            (AgentState.ANALYZING, "need_tool"):  AgentState.TOOL_CALL,
            (AgentState.ANALYZING, "no_tool"):    AgentState.RESPOND,
            (AgentState.TOOL_CALL, "tool_done"):  AgentState.RESPOND,
            (AgentState.RESPOND,   "sent"):       AgentState.IDLE,
        }

    # 4. 状态转移引擎
    def send(self, event: str):
        key = (self.state, event)
        if key in self.transitions:
            old_state = self.state
            self.state = self.transitions[key]
            print(f"✅ 状态转移: [{old_state.name}] --({event})--> [{self.state.name}]")
        else:
            # 非法转移直接抛出异常，防止 Agent 做出越界行为
            raise ValueError(
                f"❌ 无效转移: 在状态 {self.state.name} 下无法处理事件 '{event}'"
            )

# 测试运行
fsm = AgentFSM()
fsm.send("user_input")  # IDLE -> ANALYZING
fsm.send("need_tool")   # ANALYZING -> TOOL_CALL
fsm.send("tool_done")   # TOOL_CALL -> RESPOND
fsm.send("sent")        # RESPOND -> IDLE

from typing import TypedDict
from langgraph.graph import StateGraph, END

# 1. 定义全局共享的 State 数据结构
class AgentState(TypedDict):
    messages: list[str]     # 聊天记录
    next_action: str        # 路由标记："tool" 或 "respond"
    tool_result: str | None # 工具返回的结果

# 2. 定义处理节点 (Nodes)
def analyze(state: AgentState) -> dict:
    print("正在分析意图...")
    # 模拟 LLM 判断：假设用户问了天气，需要工具
    return {"next_action": "tool"}

def call_tool(state: AgentState) -> dict:
    print("正在调用天气 API...")
    return {"tool_result": "香港今天晴朗，28度"}

def respond(state: AgentState) -> dict:
    print("生成最终回复...")
    return {"messages": state["messages"] + ["最终回复内容"]}

# 3. 构建图状态机
graph = StateGraph(AgentState)

# 注册节点
graph.add_node("analyze", analyze)
graph.add_node("call_tool", call_tool)
graph.add_node("respond", respond)

# 4. 定义边与路由逻辑
graph.set_entry_point("analyze")

# 核心：条件边（Conditional Edge）
# 根据 analyze 节点计算出的 next_action 动态决定走向
graph.add_conditional_edges(
    "analyze",
    lambda state: "call_tool" if state.get("next_action") == "tool" else "respond"
)

# 固定边
graph.add_edge("call_tool", "respond")
graph.add_edge("respond", END)

# 编译为可执行应用
app = graph.compile()

import json
from typing import Any

def save_checkpoint(state: dict[str, Any], step_name: str, session_id: str):
    """将当前状态及停留的节点保存到磁盘"""
    filepath = f"./checkpoints/session_{session_id}.json"
    checkpoint_data = {
        "current_step": step_name,
        "state_payload": state
    }
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(checkpoint_data, f, ensure_ascii=False, indent=2)
    print(f"💾 进度已保存至断点: {step_name}")

def load_checkpoint(session_id: str) -> dict[str, Any]:
    """从磁盘恢复状态，以便继续执行"""
    filepath = f"./checkpoints/session_{session_id}.json"
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        return {}