案例 1 · Agentic RAG(带自我纠错)

普通 RAG = "检索→塞进 prompt→生成",检索差就答得差。Agentic RAG 给检索加一个"质检+纠错"循环:检索回来先让模型给文档打分,不相关就重写问题再检索(对应面试高频的 CRAG / Self-RAG 思路)。

复制from typing_extensions import TypedDict
from typing import Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class State(TypedDict):
    question: str
    docs: list
    tries: int
    answer: str

def retrieve(state: State):
    # 真实项目换成你的向量库 retriever.invoke(state["question"])
    docs = my_retriever.invoke(state["question"])
    return {"docs": docs, "tries": state.get("tries", 0) + 1}

class Grade(BaseModel):
    relevant: bool = Field(description="检索到的文档是否足以回答问题")

def grade(state: State) -> str:
    # 用结构化输出让模型判定文档相关性(Self-RAG 的"反思"动作)
    g = llm.with_structured_output(Grade).invoke(
        f"问题:{state['question']}\n文档:{state['docs']}\n这些文档够回答吗?")
    if g.relevant:
        return "generate"
    return "generate" if state["tries"] >= 2 else "rewrite"  # 兜底:最多纠错2次

def rewrite(state: State):
    q = llm.invoke(f"把这个问题改写得更利于检索:{state['question']}").content
    return {"question": q}

def generate(state: State):
    ans = llm.invoke(f"根据文档回答。\n文档:{state['docs']}\n问题:{state['question']}").content
    return {"answer": ans}

b = StateGraph(State)
b.add_node("retrieve", retrieve)
b.add_node("rewrite", rewrite)
b.add_node("generate", generate)
b.add_edge(START, "retrieve")
b.add_conditional_edges("retrieve", grade, ["generate", "rewrite"])
b.add_edge("rewrite", "retrieve")   # 纠错循环
b.add_edge("generate", END)
graph = b.compile()

案例 2 · 多智能体研究助手(Supervisor 模式)

一个"主管"调度两个专精 Agent:研究员负责查资料,写手负责成文。对应报告里"Researcher + Writer 多 Agent 怎么设计、怎么防失控"的高频题。

复制from typing import Literal
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.types import Command
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

llm = ChatOpenAI(model="gpt-4o-mini")
MEMBERS = ["researcher", "writer"]

class Route(BaseModel):
    next: Literal["researcher", "writer", "FINISH"]

def supervisor(state: MessagesState) -> Command[Literal["researcher", "writer", "__end__"]]:
    # 主管看当前进展,决定派给谁 / 还是收工
    decision = llm.with_structured_output(Route).invoke(
        [{"role":"system","content":f"你是主管,在 {MEMBERS} 之间调度,完成就回 FINISH"}]
        + state["messages"])
    goto = decision.next
    if goto == "FINISH":
        goto = END
    return Command(goto=goto)

def researcher(state: MessagesState) -> Command[Literal["supervisor"]]:
    out = llm.invoke([{"role":"system","content":"你是研究员,查资料给要点"}] + state["messages"])
    return Command(update={"messages":[out]}, goto="supervisor")   # 干完交回主管

def writer(state: MessagesState) -> Command[Literal["supervisor"]]:
    out = llm.invoke([{"role":"system","content":"你是写手,把要点写成成稿"}] + state["messages"])
    return Command(update={"messages":[out]}, goto="supervisor")

b = StateGraph(MessagesState)
b.add_node("supervisor", supervisor)
b.add_node("researcher", researcher)
b.add_node("writer", writer)
b.add_edge(START, "supervisor")
graph = b.compile()   # 路由全靠 Command(goto),不用画一堆条件边

案例 3 · 自我反思 Reflection(对接"自进化")

生成→自我批判→修订的循环。这是 Reflexion 范式的最小实现,也是你 JD 里"自主反思"的直接落地。

复制class State(TypedDict):
    task: str
    draft: str
    critique: str
    rounds: int

def generate(state: State):
    prompt = state["task"]
    if state.get("critique"):                 # 第二轮起,带着批评意见改
        prompt += f"\n上一版:{state['draft']}\n改进意见:{state['critique']}\n请据此修订。"
    return {"draft": llm.invoke(prompt).content,
            "rounds": state.get("rounds", 0) + 1}

def reflect(state: State):
    c = llm.invoke(f"严格批评这份草稿,指出具体可改进点;若已足够好,只回 OK。\n{state['draft']}").content
    return {"critique": c}

def should_continue(state: State) -> str:
    if "OK" in state["critique"][:5] or state["rounds"] >= 3:
        return END
    return "generate"

b = StateGraph(State)
b.add_node("generate", generate)
b.add_node("reflect", reflect)
b.add_edge(START, "generate")
b.add_edge("generate", "reflect")
b.add_conditional_edges("reflect", should_continue, ["generate", END])
graph = b.compile()

案例 4 · HITL 审批工作流

Agent 要执行高风险操作(发邮件、退款)前,暂停等人工批准。对应报告里"工具权限/人审断点/kill switch"考点。这里用新版 interrupt()。

复制from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver

def draft_action(state: State):
    return {"action": "给客户退款 ¥500"}

def human_approval(state: State):
    # 执行到这里会【暂停】,把 action 抛给外部等待人工决定
    decision = interrupt({"待批准": state["action"]})
    if decision == "approve":
        return {"status": "已执行"}
    return {"status": "已拒绝"}

b = StateGraph(State)
b.add_node("draft_action", draft_action)
b.add_node("human_approval", human_approval)
b.add_edge(START, "draft_action")
b.add_edge("draft_action", "human_approval")
b.add_edge("human_approval", END)
graph = b.compile(checkpointer=MemorySaver())   # interrupt 必须配 checkpointer

config = {"configurable": {"thread_id": "order-1"}}
graph.invoke({}, config)               # 跑到 interrupt 停下
# ……把 state["action"] 推给人审,拿到结果后:
graph.invoke(Command(resume="approve"), config)   # 用人工输入恢复

案例 5 · SQL 数据 Agent

自然语言问数据库:Agent 自己看表结构→写 SQL→执行→把结果讲成人话。工具型 Agent 的经典落地。

复制from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langgraph.prebuilt import create_react_agent

db = SQLDatabase.from_uri("sqlite:///chinook.db")
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
# 工具集:列出表 / 看表结构 / 校验SQL / 执行SQL
tools = toolkit.get_tools()

system = """你是 SQL 专家。先查看有哪些表和表结构,再写语法正确的 SQLite 查询,
执行后根据结果回答。只读查询,禁止任何写操作(INSERT/UPDATE/DELETE/DROP)。"""

agent = create_react_agent(llm, tools, prompt=system)

for step in agent.stream(
        {"messages": [{"role":"user","content":"销售额最高的5个国家是?"}]},
        stream_mode="values"):
    step["messages"][-1].pretty_print()

小结:5 个案例提炼出的通用套路

你会发现:所有复杂 Agent,都是"节点 + 条件边/Command + 循环 + State"这套基本功的不同组合。把第 1、2 章吃透,这些案例你都能自己推导出来。