LangGraph入门教程：从0到1构建可编排AI智能体

引言

在AI Agent开发领域，LangGraph正逐渐成为构建复杂智能体的首选框架。作为LangChain生态的重要组成部分，LangGraph通过**有向图(Directed Graph)**的方式，让开发者能够直观地设计、调试和运行多步骤、多分支的AI工作流。无论是构建自动客服系统、代码生成助手，还是复杂的决策支持系统，LangGraph都能帮助你轻松实现智能体的编排和控制。

本文将带你从零基础开始，一步步掌握LangGraph的核心概念和使用方法，最终实现一个实用的AI智能体。

核心概念解析

在开始编码之前，我们需要先理解LangGraph中的几个核心概念：

1. 图(Graph)

整个智能体的工作流载体，包含所有节点和边的集合，定义了智能体的执行逻辑。

2. 节点(Node)

工作流中的最小执行单元，可以是：

• LLM调用（如GPT-4o、Claude 3等）
• 工具/函数调用（如数据库查询、API调用）
• 逻辑判断（如条件分支决策）
• 数据处理（如文本格式化、结果整合）

3. 边(Edge)

定义节点之间的连接关系，决定了工作流的执行顺序。分为两种类型：

• 普通边：直接连接两个节点，固定执行顺序
• 条件边：根据节点输出动态决定下一个执行节点

4. 状态(State)

贯穿整个工作流的数据载体，用于在节点之间传递信息。通常使用Python字典或Pydantic模型定义。

快速上手：构建第一个LangGraph智能体

环境准备

首先，我们需要安装必要的依赖：

pip install langgraph langchain-openai python-dotenv

创建一个.env文件，配置你的OpenAI API密钥：

OPENAI_API_KEY=your-api-key-here

基础示例：简单的对话复读机

让我们从一个最简单的例子开始，构建一个能够重复用户输入的智能体：

from langgraph import Graph
from langgraph.state import StateGraph, CompiledGraph
from typing import TypedDict, Annotated
import operator
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 1. 定义状态结构
class AgentState(TypedDict):
    user_input: str
    agent_response: str

# 2. 创建节点函数
def echo_node(state: AgentState) -> AgentState:
    """重复用户输入的节点"""
    return {
        "agent_response": f"你刚才说的是: {state['user_input']}"
    }

# 3. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("echo", echo_node)

# 设置入口点
workflow.set_entry_point("echo")

# 设置结束点
workflow.set_finish_point("echo")

# 4. 编译图
app = workflow.compile()

# 5. 运行智能体
if __name__ == "__main__":
    result = app.invoke({
        "user_input": "你好，LangGraph！"
    })
    print(result["agent_response"])

运行结果：

你刚才说的是: 你好，LangGraph！

进阶示例：带条件分支的智能体

现在让我们构建一个更复杂的智能体，它能够根据用户输入的内容，决定是返回普通回复还是调用工具：

from langgraph import Graph
from langgraph.state import StateGraph
from typing import TypedDict, Literal
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()

# 1. 定义状态结构
class AgentState(TypedDict):
    user_input: str
    agent_response: str
    should_call_tool: bool

# 2. 初始化LLM
llm = ChatOpenAI(model="gpt-4o-mini")

# 3. 创建节点函数
def user_input_node(state: AgentState) -> AgentState:
    """处理用户输入的节点"""
    return state

def decision_node(state: AgentState) -> AgentState:
    """判断是否需要调用工具的节点"""
    user_input = state["user_input"].lower()
    
    # 判断是否包含特定关键词
    if any(keyword in user_input for keyword in ["天气", "温度", "气温"]):
        return {"should_call_tool": True}
    else:
        return {"should_call_tool": False}

def normal_response_node(state: AgentState) -> AgentState:
    """生成普通回复的节点"""
    response = llm.invoke(f"用户说: {state['user_input']}，请友好地回复用户")
    return {"agent_response": response.content}

def tool_call_node(state: AgentState) -> AgentState:
    """模拟调用天气工具的节点"""
    # 这里只是模拟实际工具调用，真实场景中可以接入天气API
    return {"agent_response": "今天天气晴朗，温度25℃，非常适合户外活动！"}

# 4. 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("user_input", user_input_node)
workflow.add_node("decision", decision_node)
workflow.add_node("normal_response", normal_response_node)
workflow.add_node("tool_call", tool_call_node)

# 设置入口点
workflow.set_entry_point("user_input")

# 添加边
workflow.add_edge("user_input", "decision")

# 添加条件边
def decide_next_node(state: AgentState) -> Literal["normal_response", "tool_call"]:
    return "tool_call" if state["should_call_tool"] else "normal_response"

workflow.add_conditional_edges(
    "decision",
    decide_next_node,
    {
        "normal_response": "normal_response",
        "tool_call": "tool_call"
    }
)

# 设置结束点
workflow.add_edge("normal_response", "__end__")
workflow.add_edge("tool_call", "__end__")

# 5. 编译图
app = workflow.compile()

# 6. 运行智能体
if __name__ == "__main__":
    # 测试普通对话
    print("测试普通对话:")
    result1 = app.invoke({"user_input": "你好，我想了解一下LangGraph"})
    print(result1["agent_response"])
    
    print("\n测试工具调用:")
    # 测试工具调用
    result2 = app.invoke({"user_input": "北京今天天气怎么样？"})
    print(result2["agent_response"])

运行结果：

测试普通对话:
你好！LangGraph是一个基于LangChain框架的扩展库，它允许你使用有向图来构建复杂的AI智能体工作流。通过可视化的方式，你可以轻松设计多步骤、多分支的智能体逻辑，实现更灵活、可控的AI应用。

测试工具调用:
今天天气晴朗，温度25℃，非常适合户外活动！

最佳实践与技巧

1. 状态管理最佳实践

• 使用TypedDict/Pydantic模型：明确定义状态结构，提高代码可读性和可维护性
• 最小化状态传递：只在状态中保存必要的数据，避免不必要的内存占用
• 状态版本控制：对于复杂工作流，可以考虑添加状态版本字段，方便调试和回溯

2. 节点设计原则

• 单一职责：每个节点只负责一个功能，保持节点逻辑简洁
• 无状态设计：节点函数尽量只依赖输入的状态，避免使用全局变量
• 错误处理：在节点中添加适当的错误处理逻辑，提高系统健壮性

3. 调试与可视化

LangGraph提供了强大的可视化功能，帮助你直观地理解工作流：

# 可视化图结构
app.get_graph().draw_mermaid_png(output_file="agent_workflow.png")

这将生成一个PNG图片，展示你的智能体工作流结构。

4. 性能优化建议

• 节点并行执行：对于互不依赖的节点，可以使用并行执行提高效率
• 缓存机制：对重复的LLM调用或工具调用结果进行缓存
• 流式输出：对于长时间运行的节点，使用流式输出提高用户体验

高级应用：构建多智能体协作系统

LangGraph的真正强大之处在于能够构建多智能体协作系统。下面是一个简单的多智能体对话示例：

from langgraph import Graph
from langgraph.state import StateGraph
from typing import TypedDict, List
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()

class Message(TypedDict):
    role: str
    content: str

class MultiAgentState(TypedDict):
    messages: List[Message]
    next_agent: str

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

def customer_service_agent(state: MultiAgentState) -> MultiAgentState:
    """客服智能体"""
    messages = state["messages"]
    response = llm.invoke(messages + [{"role": "system", "content": "你是专业的客服人员，请友好回复用户"}])
    new_message = {"role": "assistant", "content": response.content}
    
    # 判断是否需要转交给技术支持
    if any(keyword in response.content.lower() for keyword in ["技术问题", "bug", "报错"]):
        return {
            "messages": messages + [new_message],
            "next_agent": "technical_support"
        }
    else:
        return {
            "messages": messages + [new_message],
            "next_agent": "__end__"
        }

def technical_support_agent(state: MultiAgentState) -> MultiAgentState:
    """技术支持智能体"""
    messages = state["messages"]
    response = llm.invoke(messages + [{"role": "system", "content": "你是资深技术支持工程师，请专业地解决用户的技术问题"}])
    new_message = {"role": "assistant", "content": response.content}
    
    return {
        "messages": messages + [new_message],
        "next_agent": "__end__"
    }

workflow = StateGraph(MultiAgentState)

workflow.add_node("customer_service", customer_service_agent)
workflow.add_node("technical_support", technical_support_agent)

workflow.set_entry_point("customer_service")

def route_agent(state: MultiAgentState) -> str:
    return state["next_agent"]

workflow.add_conditional_edges(
    "customer_service",
    route_agent,
    {
        "technical_support": "technical_support",
        "__end__": "__end__"
    }
)

workflow.add_edge("technical_support", "__end__")

app = workflow.compile()

if __name__ == "__main__":
    result = app.invoke({
        "messages": [{"role": "user", "content": "我的软件总是崩溃，显示错误代码500，这是什么问题？"}],
        "next_agent": ""
    })
    
    for msg in result["messages"]:
        print(f"{msg['role']}: {msg['content']}")

总结与展望

LangGraph为AI Agent开发带来了革命性的变化，它让复杂的智能体工作流设计变得直观、可控。通过本文的学习，你已经掌握了：

1. LangGraph的核心概念和基本使用方法
2. 如何构建简单和复杂的AI智能体
3. LangGraph的最佳实践和调试技巧
4. 如何构建多智能体协作系统

随着AI技术的不断发展，LangGraph也在持续进化，未来将支持更多的功能和扩展：

• 更强大的可视化调试工具
• 与更多LLM和工具的原生集成
• 更灵活的状态管理和持久化机制
• 分布式执行和水平扩展能力

如果你想深入学习LangGraph，推荐访问官方文档，那里有更详细的教程和高级应用示例。

参考资料

1. LangGraph官方文档
2. LangChain官方文档
3. LangGraph GitHub仓库
4. LangGraph入门教程(一):从0到1
5. 万字实战教程:零基础使用LangGraph构建你的第一个大模型Agent

希望本文能帮助你快速入门LangGraph，开启你的AI Agent开发之旅！如果你有任何问题或建议，欢迎在评论区留言交流。

---