LangGraph 实战教程：构建有状态的 AI 应用

pip install langgraph langchain-openai langchain-core python-dotenv httpx

OPENAI_API_KEY=your-api-key
OPENAI_BASE_URL=https://api.xiaomi.com/v1

# step1_simple_chatbot.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

load_dotenv()

# 创建自定义 httpx 客户端（禁用 SSL 验证以解决某些环境的连接问题）
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 创建状态图
graph_builder = StateGraph(State)

# 定义聊天节点
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 添加节点和边
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 编译图
graph = graph_builder.compile()

# 测试对话
def test_chat():
    print("=== 简单聊天机器人测试 ===")
    
    response = graph.invoke({"messages": [{"role": "user", "content": "你好，请介绍一下自己"}]})
    print("用户: 你好，请介绍一下自己")
    print("助手:", response["messages"][-1].content)
    
    response = graph.invoke({"messages": [{"role": "user", "content": "什么是 LangGraph？"}]})
    print("\n用户: 什么是 LangGraph？")
    print("助手:", response["messages"][-1].content)

if __name__ == "__main__":
    test_chat()

py step1_simple_chatbot.py

# step2_streaming_chatbot.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM（启用流式输出）
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0.7, streaming=True, http_client=custom_client)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 创建状态图
graph_builder = StateGraph(State)

# 定义聊天节点（使用流式输出）
def chatbot(state: State):
    # 获取完整响应
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 添加节点和边
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 编译图
graph = graph_builder.compile()

# 流式对话
def streaming_chat():
    print("=== 流式聊天机器人测试 ===")
    print("输入 'quit' 退出\n")
    
    while True:
        user_input = input("你: ")
        if user_input.lower() in ['quit', 'exit', '退出']:
            break
        
        print("助手: ", end="")
        # 使用 stream 获取流式输出
        for chunk in graph.stream(
            {"messages": [HumanMessage(content=user_input)]},
            stream_mode="messages"
        ):
            # chunk 是 (message, metadata) 元组
            if isinstance(chunk, tuple) and len(chunk) > 0:
                message = chunk[0]
                if hasattr(message, 'content') and message.content:
                    print(message.content, end="", flush=True)
        print()

if __name__ == "__main__":
    streaming_chat()

py step2_streaming_chatbot.py

# step3_tool_agent.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 定义工具
@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息"""
    results = {
        "天气": "今天北京天气晴朗，气温 25°C",
        "新闻": "最新科技新闻：AI 技术取得重大突破",
        "股票": "苹果公司股价今日上涨 2%"
    }
    
    for key, value in results.items():
        if key in query:
            return value
    
    return f"搜索结果：关于 '{query}' 的信息"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

# 绑定工具到 LLM
tools = [search_web, calculate]
llm_with_tools = llm.bind_tools(tools)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 创建状态图
graph_builder = StateGraph(State)

# 定义工具节点
def tool_node(state: State):
    # 获取最后一条消息
    last_message = state["messages"][-1]
    
    # 检查是否有工具调用
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        results = []
        for tool_call in last_message.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            
            # 执行工具
            if tool_name == "search_web":
                result = search_web.invoke(tool_args)
            elif tool_name == "calculate":
                result = calculate.invoke(tool_args)
            else:
                result = f"未知工具: {tool_name}"
            
            results.append({"role": "tool", "content": result, "tool_call_id": tool_call["id"]})
        
        return {"messages": results}
    
    return {"messages": []}

# 定义聊天节点
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 定义路由函数
def should_use_tools(state: State):
    last_message = state["messages"][-1]
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "tools"
    return END

# 添加节点
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)

# 添加边
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", should_use_tools, {"tools": "tools", END: END})
graph_builder.add_edge("tools", "chatbot")

# 编译图
graph = graph_builder.compile()

# 测试对话
def test_tool_agent():
    print("=== 工具调用 Agent 测试 ===")
    
    # 测试天气查询
    response = graph.invoke({"messages": [{"role": "user", "content": "今天北京天气怎么样？"}]})
    print("用户: 今天北京天气怎么样？")
    print("助手:", response["messages"][-1].content)
    
    # 测试计算
    response = graph.invoke({"messages": [{"role": "user", "content": "计算 (15 + 27) * 3"}]})
    print("\n用户: 计算 (15 + 27) * 3")
    print("助手:", response["messages"][-1].content)
    
    # 测试普通对话
    response = graph.invoke({"messages": [{"role": "user", "content": "你好，请介绍一下自己"}]})
    print("\n用户: 你好，请介绍一下自己")
    print("助手:", response["messages"][-1].content)

if __name__ == "__main__":
    test_tool_agent()

py step3_tool_agent.py

# step4_memory_chatbot.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 创建状态图
graph_builder = StateGraph(State)

# 定义聊天节点
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 添加节点和边
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 创建记忆存储
memory = MemorySaver()

# 编译图（带记忆）
graph = graph_builder.compile(checkpointer=memory)

# 测试多轮对话
def test_memory_chat():
    print("=== 带记忆的聊天机器人测试 ===")
    
    # 创建线程配置
    config = {"configurable": {"thread_id": "user_1"}}
    
    # 第一轮对话
    response = graph.invoke(
        {"messages": [{"role": "user", "content": "你好，我叫张三"}]},
        config=config
    )
    print("用户: 你好，我叫张三")
    print("助手:", response["messages"][-1].content)
    
    # 第二轮对话
    response = graph.invoke(
        {"messages": [{"role": "user", "content": "我是一名 Python 开发者"}]},
        config=config
    )
    print("\n用户: 我是一名 Python 开发者")
    print("助手:", response["messages"][-1].content)
    
    # 第三轮对话（测试记忆）
    response = graph.invoke(
        {"messages": [{"role": "user", "content": "你还记得我叫什么吗？"}]},
        config=config
    )
    print("\n用户: 你还记得我叫什么吗？")
    print("助手:", response["messages"][-1].content)

if __name__ == "__main__":
    test_memory_chat()

py step4_memory_chatbot.py

# step5_conditional_routing.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]
    category: str
    sentiment: str

# 创建状态图
graph_builder = StateGraph(State)

# 定义分类节点
def classify(state: State):
    last_message = state["messages"][-1].content
    
    prompt = f"""请分析以下用户消息，返回 JSON 格式：
{{
    "category": "技术问题/生活问题/其他",
    "sentiment": "正面/负面/中性"
}}

用户消息：{last_message}"""
    
    response = llm.invoke([{"role": "user", "content": prompt}])
    
    # 简单解析（实际应用中应该使用更健壮的解析）
    content = response.content
    if "技术问题" in content:
        category = "技术问题"
    elif "生活问题" in content:
        category = "生活问题"
    else:
        category = "其他"
    
    if "正面" in content:
        sentiment = "正面"
    elif "负面" in content:
        sentiment = "负面"
    else:
        sentiment = "中性"
    
    return {"category": category, "sentiment": sentiment}

# 定义技术问题处理节点
def handle_tech(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义生活问题处理节点
def handle_life(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义其他问题处理节点
def handle_other(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义路由函数
def route_by_category(state: State):
    category = state.get("category", "其他")
    if category == "技术问题":
        return "handle_tech"
    elif category == "生活问题":
        return "handle_life"
    else:
        return "handle_other"

# 添加节点
graph_builder.add_node("classify", classify)
graph_builder.add_node("handle_tech", handle_tech)
graph_builder.add_node("handle_life", handle_life)
graph_builder.add_node("handle_other", handle_other)

# 添加边
graph_builder.add_edge(START, "classify")
graph_builder.add_conditional_edges("classify", route_by_category, {
    "handle_tech": "handle_tech",
    "handle_life": "handle_life",
    "handle_other": "handle_other"
})
graph_builder.add_edge("handle_tech", END)
graph_builder.add_edge("handle_life", END)
graph_builder.add_edge("handle_other", END)

# 编译图
graph = graph_builder.compile()

# 测试条件路由
def test_conditional_routing():
    print("=== 条件路由测试 ===")
    
    # 测试技术问题
    response = graph.invoke({"messages": [{"role": "user", "content": "Python 的列表和元组有什么区别？"}]})
    print("用户: Python 的列表和元组有什么区别？")
    print("分类:", response.get("category"))
    print("情感:", response.get("sentiment"))
    print("助手:", response["messages"][-1].content[:100] + "...")
    
    # 测试生活问题
    response = graph.invoke({"messages": [{"role": "user", "content": "今天天气真好，适合出去玩"}]})
    print("\n用户: 今天天气真好，适合出去玩")
    print("分类:", response.get("category"))
    print("情感:", response.get("sentiment"))
    print("助手:", response["messages"][-1].content[:100] + "...")

if __name__ == "__main__":
    test_conditional_routing()

py step5_conditional_routing.py

# step6_human_approval.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]
    needs_approval: bool
    approved: bool

# 创建状态图
graph_builder = StateGraph(State)

# 定义生成节点
def generate(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response], "needs_approval": True}

# 定义审核节点
def human_approval(state: State):
    # 在实际应用中，这里会暂停并等待人工审核
    # 这里我们模拟人工审核
    print("\n[人工审核] 生成的内容需要审核")
    print("内容:", state["messages"][-1].content[:200] + "...")
    
    # 模拟审核通过
    return {"approved": True, "needs_approval": False}

# 定义发布节点
def publish(state: State):
    print("\n[发布] 内容已发布")
    return {"messages": state["messages"]}

# 定义路由函数
def should_publish(state: State):
    if state.get("approved", False):
        return "publish"
    return END

# 添加节点
graph_builder.add_node("generate", generate)
graph_builder.add_node("human_approval", human_approval)
graph_builder.add_node("publish", publish)

# 添加边
graph_builder.add_edge(START, "generate")
graph_builder.add_edge("generate", "human_approval")
graph_builder.add_conditional_edges("human_approval", should_publish, {
    "publish": "publish",
    END: END
})
graph_builder.add_edge("publish", END)

# 编译图
graph = graph_builder.compile()

# 测试人工审核
def test_human_approval():
    print("=== 人工审核流程测试 ===")
    
    response = graph.invoke({
        "messages": [{"role": "user", "content": "请写一篇关于 AI 的短文"}],
        "needs_approval": False,
        "approved": False
    })
    
    print("\n最终内容:", response["messages"][-1].content[:200] + "...")

if __name__ == "__main__":
    test_human_approval()

py step6_human_approval.py

# step7_multi_agent.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]
    research: str
    draft: str
    final: str

# 创建状态图
graph_builder = StateGraph(State)

# 定义研究员节点
def researcher(state: State):
    last_message = state["messages"][-1].content
    
    prompt = f"""你是一位专业的研究员。请对以下主题进行研究，提供关键信息和要点：
    
主题：{last_message}

请提供：
1. 主题概述
2. 关键要点
3. 相关数据或事实
4. 注意事项"""
    
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"research": response.content}

# 定义写作员节点
def writer(state: State):
    prompt = f"""你是一位专业的写作员。请根据以下研究资料，撰写一篇结构清晰的文章：
    
研究资料：
{state['research']}

请确保文章：
1. 有清晰的标题
2. 结构合理
3. 语言流畅
4. 包含总结"""
    
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"draft": response.content}

# 定义编辑节点
def editor(state: State):
    prompt = f"""你是一位专业的编辑。请审阅以下文章，提供改进建议：
    
文章内容：
{state['draft']}

请检查：
1. 语法错误
2. 逻辑连贯性
3. 表达清晰度
4. 格式规范

如果文章质量良好，请回复"通过"。否则，请提供具体的修改建议。"""
    
    response = llm.invoke([{"role": "user", "content": prompt}])
    
    # 检查是否通过
    if "通过" in response.content:
        return {"final": state["draft"]}
    else:
        # 如果需要修改，返回修改后的内容
        revise_prompt = f"""请根据以下建议修改文章：
        
原文：
{state['draft']}

修改建议：
{response.content}

请直接输出修改后的完整文章。"""
        
        revised_response = llm.invoke([{"role": "user", "content": revise_prompt}])
        return {"final": revised_response.content}

# 添加节点
graph_builder.add_node("researcher", researcher)
graph_builder.add_node("writer", writer)
graph_builder.add_node("editor", editor)

# 添加边
graph_builder.add_edge(START, "researcher")
graph_builder.add_edge("researcher", "writer")
graph_builder.add_edge("writer", "editor")
graph_builder.add_edge("editor", END)

# 编译图
graph = graph_builder.compile()

# 测试多 Agent 协作
def test_multi_agent():
    print("=== 多 Agent 协作测试 ===")
    
    response = graph.invoke({
        "messages": [{"role": "user", "content": "人工智能在医疗领域的应用"}],
        "research": "",
        "draft": "",
        "final": ""
    })
    
    print("\n研究资料:")
    print(response["research"][:300] + "...")
    
    print("\n最终文章:")
    print(response["final"][:500] + "...")

if __name__ == "__main__":
    test_multi_agent()

py step7_multi_agent.py

# customer_service.py
from dotenv import load_dotenv
import httpx
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

load_dotenv()

# 创建自定义 httpx 客户端
transport = httpx.HTTPTransport(http2=False, retries=3, verify=False)
custom_client = httpx.Client(transport=transport, timeout=30)

# 创建 LLM
llm = ChatOpenAI(model="mimo-v2.5-pro", temperature=0, http_client=custom_client)

# 定义状态
class State(TypedDict):
    messages: Annotated[list, add_messages]
    customer_info: dict
    issue_type: str
    resolved: bool

# 创建状态图
graph_builder = StateGraph(State)

# 定义识别节点
def identify_customer(state: State):
    # 在实际应用中，这里会查询数据库
    return {"customer_info": {"name": "张三", "vip_level": "黄金"}}

# 定义分类节点
def classify_issue(state: State):
    last_message = state["messages"][-1].content
    
    prompt = f"""请分析以下客户问题，返回问题类型：
    
客户问题：{last_message}

可能的类型：
1. 技术支持
2. 账户问题
3. 产品咨询
4. 投诉建议
5. 其他

请只返回类型名称。"""
    
    response = llm.invoke([{"role": "user", "content": prompt}])
    return {"issue_type": response.content.strip()}

# 定义技术支持节点
def tech_support(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义账户问题节点
def account_issue(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义产品咨询节点
def product_consult(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义投诉建议节点
def complaint(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义通用节点
def general(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# 定义路由函数
def route_by_issue(state: State):
    issue_type = state.get("issue_type", "其他")
    if "技术" in issue_type:
        return "tech_support"
    elif "账户" in issue_type:
        return "account_issue"
    elif "产品" in issue_type:
        return "product_consult"
    elif "投诉" in issue_type:
        return "complaint"
    else:
        return "general"

# 添加节点
graph_builder.add_node("identify_customer", identify_customer)
graph_builder.add_node("classify_issue", classify_issue)
graph_builder.add_node("tech_support", tech_support)
graph_builder.add_node("account_issue", account_issue)
graph_builder.add_node("product_consult", product_consult)
graph_builder.add_node("complaint", complaint)
graph_builder.add_node("general", general)

# 添加边
graph_builder.add_edge(START, "identify_customer")
graph_builder.add_edge("identify_customer", "classify_issue")
graph_builder.add_conditional_edges("classify_issue", route_by_issue, {
    "tech_support": "tech_support",
    "account_issue": "account_issue",
    "product_consult": "product_consult",
    "complaint": "complaint",
    "general": "general"
})
graph_builder.add_edge("tech_support", END)
graph_builder.add_edge("account_issue", END)
graph_builder.add_edge("product_consult", END)
graph_builder.add_edge("complaint", END)
graph_builder.add_edge("general", END)

# 创建记忆存储
memory = MemorySaver()

# 编译图（带记忆）
graph = graph_builder.compile(checkpointer=memory)

# 测试智能客服
def test_customer_service():
    print("=== 智能客服系统测试 ===")
    
    config = {"configurable": {"thread_id": "customer_1"}}
    
    # 测试技术支持
    response = graph.invoke({
        "messages": [{"role": "user", "content": "我的电脑无法开机，怎么办？"}],
        "customer_info": {},
        "issue_type": "",
        "resolved": False
    }, config=config)
    
    print("客户: 我的电脑无法开机，怎么办？")
    print("问题类型:", response.get("issue_type"))
    print("客服:", response["messages"][-1].content[:200] + "...")
    
    # 测试产品咨询
    response = graph.invoke({
        "messages": [{"role": "user", "content": "你们最新的产品有什么功能？"}],
        "customer_info": {},
        "issue_type": "",
        "resolved": False
    }, config=config)
    
    print("\n客户: 你们最新的产品有什么功能？")
    print("问题类型:", response.get("issue_type"))
    print("客服:", response["messages"][-1].content[:200] + "...")

if __name__ == "__main__":
    test_customer_service()

py customer_service.py

使用 LangGraph 的中断功能，可以在需要时暂停图的执行。

在条件边中使用多个条件判断，或者使用子图来组织复杂的逻辑。

使用 MemorySaver 或数据库存储（如 SQLite、PostgreSQL）来持久化状态。

使用 LangSmith 进行监控和调试，或者在节点中添加日志输出。