AI Agent 开发学习与面试指南：04-Agent 架构与 LangGraph 状态编排

输入 -> 决定动作 -> 调用工具 -> 观察 -> 再决定 -> 完成

from typing import Literal, TypedDict


class AgentState(TypedDict, total=False):
    run_id: str
    user_id: str
    goal: str
    route: Literal["knowledge", "ticket", "human"]
    retrieved_evidence: list[dict]
    tool_calls: list[dict]
    tool_results: list[dict]
    step_count: int
    max_steps: int
    final_answer: str
    error_code: str | None

from typing import Literal, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph


class SupportState(TypedDict, total=False):
    query: str
    route: Literal["knowledge", "ticket", "human"]
    evidence: list[str]
    ticket_id: str
    answer: str


def classify(state: SupportState) -> dict:
    query = state["query"]
    if "退款" in query or "故障" in query:
        return {"route": "ticket"}
    if "人工" in query:
        return {"route": "human"}
    return {"route": "knowledge"}


def route_after_classify(state: SupportState) -> str:
    return state["route"]


def retrieve(state: SupportState) -> dict:
#    # 实际项目中调用 Retriever，并返回带来源的结构化 Evidence。
    return {"evidence": ["[S1] 示例制度：普通问题先查知识库。"]}


def answer(state: SupportState) -> dict:
    evidence = "\n".join(state.get("evidence", []))
    return {"answer": f"根据以下资料回答：\n{evidence}"}


def create_ticket(state: SupportState) -> dict:
#    # 真正实现必须有权限、幂等键和审计。
    return {"ticket_id": "T-001", "answer": "已创建工单 T-001"}


def handoff(state: SupportState) -> dict:
    return {"answer": "已转接人工客服"}


builder = StateGraph(SupportState)
builder.add_node("classify", classify)
builder.add_node("retrieve", retrieve)
builder.add_node("answer", answer)
builder.add_node("create_ticket", create_ticket)
builder.add_node("handoff", handoff)

builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_after_classify,
    {
        "knowledge": "retrieve",
        "ticket": "create_ticket",
        "human": "handoff",
    },
)
builder.add_edge("retrieve", "answer")
builder.add_edge("answer", END)
builder.add_edge("create_ticket", END)
builder.add_edge("handoff", END)

graph = builder.compile(checkpointer=InMemorySaver())

config = {"configurable": {"thread_id": "demo-thread-1"}}
result = graph.invoke({"query": "年假有多少天？"}, config=config)
print(result["answer"])

from typing import Literal
from pydantic import BaseModel, Field


class NextAction(BaseModel):
    action: Literal["search", "create_ticket", "ask_user", "finish"]
    tool_name: str | None = None
    tool_arguments: dict = Field(default_factory=dict)
    user_question: str | None = None
    final_answer: str | None = None

def guard_limits(state: AgentState) -> Literal["continue", "stop"]:
    if state["step_count"] >= state["max_steps"]:
        return "stop"

    calls = state.get("tool_calls", [])
    if len(calls) >= 2 and calls[-1] == calls[-2]:
        return "stop"

    return "continue"

from langgraph.types import Command, interrupt


def approve_ticket(state: SupportState) -> dict:
    decision = interrupt({
        "question": "是否创建工单？",
        "proposed_title": state["query"][:80],
    })
    return {"approved": bool(decision)}


## 首次运行遇到 interrupt 后暂停；恢复时必须使用同一 thread_id。
## graph.invoke(Command(resume=True), config=config)