2024最新LangGraph入门教程：零基础构建你的第一个状态管理AI智能体

【免费下载链接】awesome-LangGraph 项目地址: https://gitcode.com/gh_mirrors/aw/awesome-LangGraph

LangGraph是LangChain生态系统中的核心框架，专为构建状态管理AI智能体和多智能体系统而设计。本教程将帮助零基础开发者快速掌握LangGraph的核心概念，通过简单步骤创建你的第一个具有持久化状态和复杂工作流的AI智能体。

为什么选择LangGraph构建AI智能体？

LangGraph作为LangChain生态的重要组成部分，解决了传统AI应用在状态管理、长期运行和复杂决策流程中的关键挑战。与其他框架相比，它提供了以下独特优势：

• 持久化状态管理：智能体可以在多次对话和系统重启后保持上下文
• 灵活的工作流控制：通过图结构定义复杂的决策路径和条件分支
• 多智能体协作：轻松构建分工明确的智能体团队，实现复杂任务分解
• 强大的工具集成：无缝对接外部API、数据库和工具，扩展智能体能力
• 生产级部署：通过LangGraph Platform实现一键部署和扩展

LangChain生态系统概览

LangGraph并非孤立存在，而是LangChain生态系统的核心组件之一。以下是完整的生态系统架构：

LangChain生态系统架构图，展示了LangGraph与其他组件的关系，包括LangChain核心框架、LangSmith调试平台和LangGraph Platform部署环境

从架构图中可以看到，LangGraph处于生态系统的核心位置，与以下关键组件紧密集成：

• LangChain：提供基础组件和集成能力，是构建AI应用的基础框架
• LangSmith：用于智能体调试、评估和监控的开发工具
• LangGraph Platform：用于部署和扩展LangGraph应用的生产环境
• Integrations：丰富的第三方工具和服务集成

快速开始：环境准备与安装

系统要求

• Python 3.8+ 或 Node.js 18+
• pip 或 npm/yarn 包管理器
• Git 版本控制工具

安装步骤

首先，克隆项目仓库到本地：

git clone https://gitcode.com/gh_mirrors/aw/awesome-LangGraph
cd awesome-LangGraph

Python环境安装

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或在Windows上
venv\Scripts\activate

# 安装LangGraph
pip install langgraph

JavaScript环境安装

# 初始化项目
npm init -y

# 安装LangGraph
npm install @langchain/langgraph

LangGraph核心概念解析

在开始构建智能体之前，让我们先了解几个核心概念：

1. 状态(State)

状态是LangGraph的核心，它定义了智能体在执行过程中需要跟踪的所有信息。状态可以包括：

• 对话历史
• 当前任务
• 已完成的操作
• 外部工具返回的结果
• 任何需要跨步骤保存的信息

状态通常定义为一个Pydantic模型（Python）或接口（TypeScript），例如：

from typing import List, Optional
from pydantic import BaseModel

class AgentState(BaseModel):
    messages: List[str] = []
    current_task: str = ""
    completed_tasks: List[str] = []
    tool_results: dict = {}

2. 节点(Node)

节点代表智能体工作流中的一个步骤或操作，通常是一个函数。节点接收当前状态，执行特定操作，并返回更新后的状态。常见的节点类型包括：

• 工具调用节点：调用外部API或工具
• 决策节点：根据当前状态决定下一步行动
• 处理节点：处理信息或生成响应
• 条件节点：根据条件判断流向

3. 边(Edge)

边定义了节点之间的连接关系，决定了工作流的执行路径。边可以是：

• 直接边：从一个节点直接连接到另一个节点
• 条件边：根据特定条件决定流向哪个节点
• 动态边：根据节点返回结果动态决定下一个节点

4. 图(Graph)

图是LangGraph应用的完整定义，由节点和边组成。它定义了智能体的完整工作流程和状态管理规则。

构建你的第一个状态管理智能体

让我们通过一个简单的"任务跟踪智能体"示例，展示如何使用LangGraph构建具有状态管理能力的AI智能体。

步骤1：定义状态

首先，我们需要定义智能体的状态结构：

from typing import List, Optional
from pydantic import BaseModel

class TaskTrackerState(BaseModel):
    """任务跟踪智能体的状态"""
    tasks: List[str] = []  # 当前任务列表
    completed_tasks: List[str] = []  # 已完成任务
    current_task: Optional[str] = None  # 当前正在执行的任务
    user_query: str = ""  # 用户最新查询
    response: str = ""  # 智能体响应

步骤2：创建节点函数

接下来，我们创建几个核心节点函数：

def add_task(state: TaskTrackerState) -> TaskTrackerState:
    """添加新任务到任务列表"""
    # 从用户查询中提取任务
    task = state.user_query.replace("添加任务", "").strip()
    if task:
        state.tasks.append(task)
        state.response = f"已添加任务: {task}"
    else:
        state.response = "请提供要添加的任务内容"
    return state

def list_tasks(state: TaskTrackerState) -> TaskTrackerState:
    """列出所有任务"""
    if state.tasks:
        task_list = "\n".join([f"- {task}" for task in state.tasks])
        state.response = f"当前任务列表:\n{task_list}"
    else:
        state.response = "任务列表为空"
    return state

def complete_task(state: TaskTrackerState) -> TaskTrackerState:
    """标记任务为已完成"""
    task_number = int(state.user_query.replace("完成任务", "").strip()) - 1
    if 0 <= task_number < len(state.tasks):
        completed = state.tasks.pop(task_number)
        state.completed_tasks.append(completed)
        state.response = f"已完成任务: {completed}"
    else:
        state.response = "无效的任务编号"
    return state

def decide_next_step(state: TaskTrackerState) -> str:
    """根据用户查询决定下一步操作"""
    query = state.user_query.lower()
    if "添加任务" in query:
        return "add_task"
    elif "列出任务" in query:
        return "list_tasks"
    elif "完成任务" in query:
        return "complete_task"
    else:
        return "unknown_command"

def handle_unknown_command(state: TaskTrackerState) -> TaskTrackerState:
    """处理未知命令"""
    state.response = "未知命令。请使用以下命令:\n- 添加任务 [任务内容]\n- 列出任务\n- 完成任务 [任务编号]"
    return state

步骤3：构建图

现在，我们使用LangGraph构建完整的工作流：

from langgraph.graph import Graph, StateGraph

# 创建状态图
workflow = StateGraph(TaskTrackerState)

# 添加节点
workflow.add_node("add_task", add_task)
workflow.add_node("list_tasks", list_tasks)
workflow.add_node("complete_task", complete_task)
workflow.add_node("unknown_command", handle_unknown_command)

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

# 添加条件边
workflow.add_conditional_edges(
    "decide_next_step",
    decide_next_step,
    {
        "add_task": "add_task",
        "list_tasks": "list_tasks",
        "complete_task": "complete_task",
        "unknown_command": "unknown_command"
    }
)

# 添加返回边，使智能体可以继续处理新命令
for node in ["add_task", "list_tasks", "complete_task", "unknown_command"]:
    workflow.add_edge(node, "decide_next_step")

# 编译图
app = workflow.compile()

步骤4：运行智能体

最后，我们可以测试这个简单的任务跟踪智能体：

def run_agent(user_query: str, current_state: TaskTrackerState = None) -> TaskTrackerState:
    """运行智能体并返回更新后的状态"""
    if current_state is None:
        current_state = TaskTrackerState()
    
    current_state.user_query = user_query
    final_state = app.invoke(current_state)
    return final_state

# 测试对话
state = None
print("任务跟踪智能体 (输入'退出'结束)")
while True:
    user_input = input("你: ")
    if user_input.lower() == "退出":
        break
    state = run_agent(user_input, state)
    print(f"智能体: {state.response}")

进阶功能与最佳实践

1. 状态持久化

在实际应用中，你可能需要将状态保存到数据库或文件中，以便在应用重启后恢复：

import json
from pathlib import Path

def save_state(state: TaskTrackerState, file_path: str = "agent_state.json"):
    """保存状态到JSON文件"""
    with open(file_path, "w") as f:
        json.dump(state.dict(), f, indent=2)

def load_state(file_path: str = "agent_state.json") -> TaskTrackerState:
    """从JSON文件加载状态"""
    if Path(file_path).exists():
        with open(file_path, "r") as f:
            state_data = json.load(f)
        return TaskTrackerState(**state_data)
    return TaskTrackerState()

2. 集成LLM增强智能

通过集成大型语言模型(LLM)，可以增强智能体的理解和决策能力：

from langchain_openai import ChatOpenAI

# 初始化LLM
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0)

def analyze_query(state: TaskTrackerState) -> TaskTrackerState:
    """使用LLM分析用户查询，提取意图和任务信息"""
    prompt = f"""
    分析用户查询: "{state.user_query}"
    
    任务:
    1. 判断用户意图是添加任务、列出任务还是完成任务
    2. 如果是添加任务，提取任务内容
    3. 如果是完成任务，提取任务编号
    
    返回JSON格式，包含intent(可能值: add_task, list_tasks, complete_task, unknown)和data(相关数据)
    """
    
    response = llm.invoke(prompt)
    # 解析LLM响应并更新状态
    # ...
    return state

3. 错误处理与重试机制

为提高智能体的健壮性，添加错误处理和重试机制：

def safe_tool_call(state: TaskTrackerState) -> TaskTrackerState:
    """安全调用外部工具，包含错误处理"""
    try:
        # 调用外部工具的代码
        # ...
    except Exception as e:
        state.response = f"操作失败: {str(e)}. 正在重试..."
        # 可以实现重试逻辑或回退策略
    return state

部署与扩展

使用LangGraph CLI部署

LangGraph提供了方便的CLI工具，帮助你快速部署应用：

# 安装LangGraph CLI
pip install langgraph-cli

# 启动开发服务器
langgraph dev

部署到LangGraph Platform

对于生产环境，推荐使用LangGraph Platform：

# 构建应用
langgraph build

# 部署到平台
langgraph deploy

学习资源与社区支持

官方学习资源

• LangGraph Academy：Introduction to LangGraph - 官方入门课程
• LangGraph文档：Python Docs | JavaScript Docs

社区项目与示例

Awesome-LangGraph项目中收录了大量社区贡献的示例和模板，包括：

• RAG应用：agentic-rag-for-dummies - 模块化的Agentic RAG实现
• 多智能体系统：Multi-Agent-Medical-Assistant - 医疗诊断多智能体系统
• 自动化工具：WebRover - 网页自动化智能体

加入社区

• LangChain Community：官方社区 - 获取最新资讯和帮助

总结

通过本教程，你已经了解了LangGraph的核心概念和基本使用方法，并构建了一个简单但功能完整的状态管理AI智能体。LangGraph的强大之处在于其灵活的状态管理和工作流控制能力，使你能够构建复杂的AI应用，如多智能体系统、自动化工作流和持久化对话助手。

随着你对LangGraph的深入了解，你可以探索更多高级功能，如人类在环(Human-in-the-loop)交互、多智能体协作、长期记忆管理等。无论你是AI爱好者、开发者还是企业用户，LangGraph都能帮助你构建更强大、更智能的AI应用。

现在就开始你的LangGraph之旅，构建属于你的下一代AI智能体吧！

【免费下载链接】awesome-LangGraph 项目地址: https://gitcode.com/gh_mirrors/aw/awesome-LangGraph