필사 모드: LangGraph 智能体工作流实战指南：从多智能体编排到生产部署

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

# 1) 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, "add_messages"]  # 累积消息
    next_action: str
    result: str

# 2) 定义节点函数
def classify_intent(state: AgentState) -> dict:
    """分类用户意图"""
    last_msg = state["messages"][-1].content
    # 用 LLM 分类意图
    intent = llm.invoke(f"Classify intent: {last_msg}")
    return {"next_action": intent.content}

def handle_question(state: AgentState) -> dict:
    """处理问题"""
    answer = llm.invoke(state["messages"])
    return {"result": answer.content}

def handle_task(state: AgentState) -> dict:
    """执行任务"""
    result = agent_executor.invoke(state["messages"])
    return {"result": result}

# 3) 构建图
graph = StateGraph(AgentState)
graph.add_node("classify", classify_intent)
graph.add_node("question", handle_question)
graph.add_node("task", handle_task)

# 4) 连接边
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    lambda state: state["next_action"],
    {
        "question": "question",
        "task": "task",
    }
)
graph.add_edge("question", END)
graph.add_edge("task", END)

# 5) 编译并执行
app = graph.compile()
result = app.invoke({"messages": [HumanMessage("What is Kubernetes?")]})

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search_docs(query: str) -> str:
    """Search internal documentation."""
    # 向量数据库检索
    results = vectorstore.similarity_search(query, k=3)
    return "\n".join([r.page_content for r in results])

@tool
def run_query(sql: str) -> str:
    """Execute a read-only SQL query."""
    return db.execute(sql).fetchall()

@tool
def create_ticket(title: str, description: str) -> str:
    """Create a Jira ticket."""
    return jira.create_issue(title=title, description=description)

llm = ChatOpenAI(model="gpt-4o")

# 自动创建 ReAct 智能体
agent = create_react_agent(
    llm,
    tools=[search_docs, run_query, create_ticket],
    state_modifier="You are a helpful DevOps assistant."
)

# 执行
result = agent.invoke({
    "messages": [HumanMessage("Check if order-service has errors and create a ticket")]
})

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import create_react_agent

class SupervisorState(TypedDict):
    messages: Annotated[list, "add_messages"]
    next_agent: str
    final_answer: str

# 专业智能体
researcher = create_react_agent(llm, tools=[search_web, search_docs])
coder = create_react_agent(llm, tools=[run_code, read_file])
reviewer = create_react_agent(llm, tools=[analyze_code, lint_code])

def supervisor(state: SupervisorState) -> dict:
    """决定委派给哪个智能体"""
    response = llm.invoke([
        SystemMessage("You are a supervisor. Route to: researcher, coder, reviewer, or FINISH"),
        *state["messages"]
    ])
    return {"next_agent": response.content.strip()}

def run_researcher(state):
    result = researcher.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_coder(state):
    result = coder.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

def run_reviewer(state):
    result = reviewer.invoke({"messages": state["messages"]})
    return {"messages": result["messages"]}

# 构建图
workflow = StateGraph(SupervisorState)
workflow.add_node("supervisor", supervisor)
workflow.add_node("researcher", run_researcher)
workflow.add_node("coder", run_coder)
workflow.add_node("reviewer", run_reviewer)

workflow.add_edge(START, "supervisor")
workflow.add_conditional_edges(
    "supervisor",
    lambda s: s["next_agent"],
    {
        "researcher": "researcher",
        "coder": "coder",
        "reviewer": "reviewer",
        "FINISH": END,
    }
)
# 每个智能体执行后返回 supervisor
for agent_name in ["researcher", "coder", "reviewer"]:
    workflow.add_edge(agent_name, "supervisor")

app = workflow.compile()

from langgraph.types import interrupt, Command

def sensitive_action(state):
    """在敏感操作前请求人工批准"""
    action = state["pending_action"]

    # 暂停执行，等待人工批准
    approval = interrupt({
        "question": f"Approve this action?\n{action}",
        "options": ["approve", "reject", "modify"]
    })

    if approval == "approve":
        return execute_action(action)
    elif approval == "reject":
        return {"result": "Action cancelled by user"}
    else:
        return {"result": f"Action modified: {approval}"}

# 与检查点一起编译
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

# 执行（在 interrupt 处停止）
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [...]}, config)
# result = {"__interrupt__": {...}}

# 人工批准后恢复
app.invoke(Command(resume="approve"), config)

from langgraph.graph import StateGraph

class ParallelState(TypedDict):
    query: str
    web_results: str
    db_results: str
    combined: str

def search_web_node(state):
    return {"web_results": search_web(state["query"])}

def search_db_node(state):
    return {"db_results": search_db(state["query"])}

def combine_results(state):
    combined = f"Web: {state['web_results']}\nDB: {state['db_results']}"
    return {"combined": combined}

graph = StateGraph(ParallelState)
graph.add_node("web", search_web_node)
graph.add_node("db", search_db_node)
graph.add_node("combine", combine_results)

# 并行执行: START → [web, db] → combine
graph.add_edge(START, "web")
graph.add_edge(START, "db")
graph.add_edge("web", "combine")
graph.add_edge("db", "combine")
graph.add_edge("combine", END)

app = graph.compile()

# 按 token 流式输出
async for event in app.astream_events(
    {"messages": [HumanMessage("Explain CQRS")]},
    version="v2"
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

# 按节点流式输出
for chunk in app.stream({"messages": [HumanMessage("...")]}):
    for node_name, output in chunk.items():
        print(f"[{node_name}]: {output}")

from langgraph.checkpoint.postgres import PostgresSaver

# PostgreSQL 检查点（生产环境）
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/langgraph"
)

app = workflow.compile(checkpointer=checkpointer)

# 会话持久化（通过 thread_id 区分）
config = {"configurable": {"thread_id": "user-456"}}

# 第一条消息
app.invoke({"messages": [HumanMessage("Hi")]}, config)

# 第二条消息（保留之前的对话上下文）
app.invoke({"messages": [HumanMessage("What did I just say?")]}, config)

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:app"
  },
  "env": ".env"
}

# 安装 LangGraph CLI
pip install langgraph-cli

# 本地测试
langgraph dev

# 构建 Docker 镜像
langgraph build -t my-agent:latest

# 部署到 LangGraph Cloud
langgraph deploy --app my-agent