从0到1搭建Multi-Agent决策系统：LangGraph完整指南

标题选项

  1. 从零到一：用LangChain LangGraph构建会思考、会决策、会协作的企业级Multi-Agent系统
  2. LangGraph实战指南：Multi-Agent决策系统的设计逻辑、状态管理与落地全流程
  3. 告别单Agent局限！手把手带你用LangGraph搭建可解释、可扩展的复杂决策AI应用
  4. 2025多Agent开发首选：LangGraph从概念到实战，覆盖状态流转、工具调用、Agent协作全栈技能
  5. 理解LangGraph的核心：以「旅行规划+酒店比价Multi-Agent系统」为例，搞定复杂决策流

引言 (Introduction)

痛点引入 (Hook)

“唉，单Agent太难用了！”最近听到好几个AI应用开发者朋友吐槽：

  • 我让GPT-4o做「从上海→东京→首尔5天4晚」的旅行规划，它能给出景点，但要查实时机票、酒店、迪士尼门票？它直接说「我没有联网权限」——明明我已经给它配了Serper搜索和Expedia插件啊！
  • 好不容易让插件链串起来了，结果出问题时根本找不到原因：是搜索API返回的数据格式错了？还是Agent没正确理解“家庭亲子游优先靠近东京浅草寺”的需求？日志里全是乱哄哄的LLM对话历史，根本没有结构化的决策记录。
  • 更头疼的是扩展：现在只有一个规划Agent和一个搜索Agent，如果想再加一个「预算审核Agent」，或者「日本签证办理提醒Agent」，是不是得把整个链的代码重写一遍？原来的SequentialChain、RouterChain根本扛不住这种复杂的循环决策和多Agent协作场景。

是啊，单Agent和简单的链式结构，在面对「需要多步推理、多轮工具调用、多角色分工协作」的复杂决策任务时，简直捉襟见肘——就像用Excel处理大数据分析，不是不能做，但效率极低、扩展性极差、调试成本极高。

文章内容概述 (What)

那么，有没有一种专门为复杂Multi-Agent决策系统设计的框架？答案是肯定的——那就是LangChain官方在2024年初推出的LangGraph！

LangGraph的核心思想，是把Multi-Agent系统建模成**“状态机（State Machine）”**：

  • 系统的所有「当前情况」（比如用户的需求、已搜索的数据、当前的决策轮数）都封装在一个统一的状态对象（State Object） 里；
  • 每个Agent（或者工具调用逻辑、数据处理逻辑）都是状态机里的一个节点（Node）；
  • 节点之间的跳转由边（Edge） 控制，既可以是顺序边（Conditional Edge？不，普通顺序是Conditional的特殊情况），也可以是条件边（Conditional Edge）——甚至可以是循环边（Loop Edge）！这就让处理“搜索→验证→搜索→再验证”的循环决策变得易如反掌。

本文将带你从0到1，用LangGraph搭建一个完整的、可落地的「家庭亲子旅行规划+酒店比价+预算审核Multi-Agent系统」：

  1. 先彻底搞懂LangGraph的核心概念（状态、节点、边、图、编译、运行）；
  2. 再一步步设计我们的系统架构和状态结构；
  3. 然后实现每个具体的Agent节点（需求解析Agent、景点筛选Agent、机票搜索Agent、酒店比价Agent、预算审核Agent）和工具调用逻辑；
  4. 接着处理复杂的条件跳转和循环决策（比如预算审核不通过时，要不要调整酒店档次？要不要调整景点？）；
  5. 最后给系统加上可解释性日志和简单的Web界面，让它成为一个真正能用的产品原型！

读者收益 (Why)

读完本文并动手实践后，你将：

  • ✅ 彻底理解为什么LangGraph是比LangChain普通链更好的Multi-Agent开发选择；
  • ✅ 掌握LangGraph的核心API和设计模式（StatefulGraph、ConditionalEdge、ToolNode、Memory、Checkpointing）；
  • ✅ 学会如何把一个复杂的业务决策问题拆解成状态机；
  • ✅ 独立搭建至少3种不同类型的Multi-Agent系统原型（除了旅行规划，还能举一反三做客服、代码审查、数据分析师助手）；
  • ✅ 了解Multi-Agent系统的最佳实践（比如状态管理、错误处理、调试技巧）；
  • ✅ 跟上AI应用开发的最新趋势——因为现在几乎所有主流的企业级多Agent项目，都在用LangGraph或者它的设计思想！

准备工作 (Prerequisites)

技术栈/知识

在开始之前，请确保你已经具备以下技术背景和知识储备：

  1. Python 3.10+基础：熟悉Python的基本语法、面向对象编程（类、继承、多态）、异步编程（async/await，虽然本文用同步代码讲解，但实际生产中推荐异步）、装饰器（虽然暂时用不到，但了解LangChain的底层原理会有帮助）；
  2. LangChain基础组件：至少用过LangChain的LLM链（LLMChain、SimpleSequentialChain）、提示词模板（PromptTemplate、ChatPromptTemplate）、Memory（ConversationBufferMemory，虽然LangGraph有自己的状态管理，但理解状态的概念会更轻松）、工具（Tools、Toolkit）——如果完全没用过LangChain，建议先花1-2天时间看一下LangChain的官方入门教程（https://python.langchain.com/v0.2/docs/get_started/introduction/）；
  3. API Key准备：你需要准备至少3个API Key（本文会详细告诉你怎么获取）：
    • OpenAI API Key：用来驱动大语言模型（LLM），我们用GPT-4o Mini或者GPT-4o（Mini足够用，成本极低）；
    • Serper API Key：用来做联网搜索（查实时机票、酒店、景点信息），每月有2500次免费调用；
    • 可选：OpenWeatherMap API Key：如果想加一个「天气查询Agent」，可以用这个；
  4. 基本的Markdown和命令行操作：因为我们要写代码示例，还要用终端运行项目。

环境/工具

  1. 安装Python 3.10+：可以从Python官网（https://www.python.org/downloads/）下载安装，或者用pyenv/conda管理Python版本（推荐用conda，因为可以隔离不同项目的依赖）；
  2. 创建一个新的Python虚拟环境： 
    # 用conda创建（推荐）
conda create -n langgraph-multi-agent python=3.11
conda activate langgraph-multi-agent

# 或者用venv创建（Python内置）
python3 -m venv langgraph-multi-agent
# Windows激活
.\langgraph-multi-agent\Scripts\activate
# Mac/Linux激活
source langgraph-multi-agent/bin/activate
  3. 安装必要的依赖库：我们需要安装LangGraph、LangChain OpenAI、LangChain Community（里面有Serper和其他常用工具）、Python-dotenv（用来管理API Key）、Streamlit（用来做简单的Web界面）： 
    pip install langgraph==0.2.36 langchain-openai==0.2.15 langchain-community==0.3.12 python-dotenv==1.0.1 streamlit==1.39.0
    注意：我在这里指定了具体的版本号，因为LangGraph和LangChain的更新非常快，不同版本之间的API可能会有变化——用指定的版本可以确保你跟着本文的代码一步一步走，不会出错！
  4. 创建项目目录结构：为了让代码更清晰、更易维护，我们先创建一个规范的项目目录： 
    langgraph-travel-planner/
├── .env                # 用来存储API Key
├── README.md           # 项目说明
├── src/                # 源代码目录
│   ├── __init__.py
│   ├── state.py        # 定义系统的状态结构
│   ├── agents/         # 存放所有的Agent节点
│   │   ├── __init__.py
│   │   ├── requirement_parser.py  # 需求解析Agent
│   │   ├── attraction_filter.py   # 景点筛选Agent
│   │   ├── flight_searcher.py     # 机票搜索Agent
│   │   ├── hotel_comparator.py    # 酒店比价Agent
│   │   └── budget_checker.py      # 预算审核Agent
│   ├── tools/          # 存放所有的工具
│   │   ├── __init__.py
│   │   ├── serper_search.py       # Serper联网搜索工具
│   │   └── weather_search.py      # 可选的天气查询工具
│   ├── graph.py        # 定义LangGraph的状态机
│   ├── prompts.py      # 存放所有的提示词模板
│   └── utils.py        # 存放通用的工具函数
└── app.py              # Streamlit Web界面

核心概念：彻底搞懂LangGraph的底层逻辑

在开始写代码之前，我们必须先花一些时间，彻底搞懂LangGraph的核心概念——这就像盖房子之前，必须先搞懂什么是地基、什么是梁、什么是柱一样。如果跳过这一步，直接写代码，你可能会遇到很多莫名其妙的问题，而且根本不知道怎么解决。

核心概念清单

LangGraph的核心概念可以用「1个核心，3个基本要素，2个关键特性，1个运行流程」来概括：

  1. 1个核心：状态机（State Machine）；
  2. 3个基本要素：状态（State）、节点（Node）、边（Edge）；
  3. 2个关键特性：Checkpointing（状态检查点）、Memory集成（虽然状态已经包含了Memory，但可以和LangChain的Memory无缝对接）；
  4. 1个运行流程：编译（Compile）→ 初始化（Initialize）→ 迭代（Iterate）→ 终止（Terminate）。

接下来，我们会逐个详细讲解这些概念，并且会用「简单的数学公式」、「直观的Mermaid架构图」、「概念对比表格」来帮助你理解。

核心概念1：状态机（State Machine）—— LangGraph的灵魂
问题背景

为什么LangGraph要把Multi-Agent系统建模成状态机？而不是像LangChain普通链那样，建模成「线性的、单向的流程」？

我们先来看一个简单的单Agent决策问题：「帮我把1000美元换成日元」。这个问题的流程非常清晰：

  1. 解析用户的需求：货币对是USD/JPY，金额是1000美元；
  2. 调用货币换算工具，获取实时汇率；
  3. 计算最终的日元金额；
  4. 把结果返回给用户。

这个流程是线性的、单向的、没有分支和循环的——用LangChain的SimpleSequentialChain就能完美解决。

但是，我们再来看本文的核心业务问题：「帮我规划上海→东京→首尔5天4晚的家庭亲子游，总预算不超过25000元人民币，2大1小（5岁小孩），优先靠近迪士尼乐园和浅草寺，需要含早餐的4星级酒店」。这个问题的流程就复杂多了：

  1. 解析用户的需求：可能需要追问（比如「具体的出发日期是哪天？」「迪士尼乐园的门票要不要包含在预算里？」）；
  2. 搜索上海→东京、东京→首尔的实时机票；
  3. 筛选东京（靠近浅草寺和迪士尼）、首尔的景点（适合5岁小孩）；
  4. 搜索并比价东京、首尔的含早餐4星级酒店；
  5. 计算总预算（机票+酒店+门票+餐饮估算）；
  6. 预算审核：如果总预算≤25000元，直接返回结果；如果总预算>25000元，怎么办？
    • 方案A：调整酒店档次（从4星降到3星）；
    • 方案B：调整景点（去掉一些需要付费的热门景点，比如迪士尼海洋公园）；
    • 方案C：调整出行日期（比如避开周末和节假日）；
    • 方案D：调整出发地/目的地（比如改成上海→首尔→东京，或者上海→大阪→东京）；
  7. 循环决策：选一个方案调整后，再重新计算总预算，直到总预算≤25000元，或者尝试了所有方案都不行（这时候需要把所有尝试过的方案和结果返回给用户，让用户自己选择）。

你看！这个流程有分支（预算审核通过/不通过）、有循环（调整→再审核→再调整→…）——用LangChain的普通链根本无法处理这种场景！因为LangChain的普通链是「固定的、单向的、一次性执行完毕的」，而状态机是「动态的、可分支的、可循环的、可暂停的、可恢复的」——这就是为什么LangGraph要把Multi-Agent系统建模成状态机！

问题描述

状态机的数学定义是什么？（虽然看起来有点枯燥，但理解数学定义能让你更深刻地理解LangGraph的底层逻辑）

根据《计算理论导论》（Introduction to the Theory of Computation），一个确定性有限状态机（Deterministic Finite Automaton, DFA） 可以用一个五元组来表示：
DFA=(Q,Σ,δ,q0,F)DFA = (Q, \Sigma, \delta, q_0, F)DFA=(Q,Σ,δ,q0​,F)
其中：

  1. QQQ：有限状态集合（Finite Set of States）——比如我们旅行规划系统的状态集合是{「需求解析中」,「机票搜索中」,「景点筛选中」,「酒店比价中」,「预算审核中」,「调整酒店档次」,「调整景点」,「完成」,「失败」}；
  2. Σ\SigmaΣ：有限输入字母表（Finite Input Alphabet）——比如我们旅行规划系统的输入字母表是{「用户初始需求」,「机票搜索结果」,「景点筛选结果」,「酒店比价结果」,「预算审核通过」,「预算审核不通过」}；
  3. δ\deltaδ：状态转移函数（Transition Function）——也就是「给定当前状态和输入，下一个状态是什么」的规则，比如δ(「预算审核中」,「预算审核不通过」)=「调整酒店档次」\delta(「预算审核中」, 「预算审核不通过」) = 「调整酒店档次」δ(「预算审核中」,「预算审核不通过」)=「调整酒店档次」；
  4. q0q_0q0​：初始状态（Initial State）——系统启动时的第一个状态，比如我们旅行规划系统的初始状态是「需求解析中」；
  5. FFF：终止状态集合（Set of Final States）——系统停止运行时的状态，比如我们旅行规划系统的终止状态集合是{「完成」,「失败」}。

不过，LangGraph的状态机和传统的DFA有三个重要的区别：

  1. 状态不是有限的、离散的：LangGraph的状态是一个任意的Python对象（可以是字典、类实例、Pydantic模型等等），它的取值可以是无限的、连续的——比如我们旅行规划系统的状态里有「总预算」这个字段，它的取值可以是0到1000000元之间的任意实数；
  2. 输入不是外部的、独立的：LangGraph的「输入」其实就是状态的变化——当一个节点执行完毕后，它会修改状态对象，然后状态转移函数会根据修改后的状态来决定下一个节点是什么；
  3. 可以暂停和恢复：因为LangGraph支持Checkpointing（状态检查点），所以你可以在任意节点执行完毕后暂停系统，保存当前的状态，然后在任意时间恢复系统继续运行——这对于处理「需要用户输入的场景」（比如预算审核不通过时，让用户选择调整方案）非常有用！
概念结构与核心要素组成

LangGraph的状态机的概念结构可以用下面的Mermaid架构图来表示：

是

否

用户初始输入

初始化状态对象 State

编译状态机 CompiledGraph

从初始状态开始迭代

当前状态是否是终止状态？

返回最终状态和结果

找到当前状态对应的节点 Node

执行节点逻辑，修改状态对象

保存状态检查点 Checkpoint（可选）

找到下一个节点 Edge（顺序边/条件边）

从这个架构图可以看出，LangGraph的状态机的核心运行流程是：

  1. 初始化：根据用户的初始输入，创建一个初始的状态对象；
  2. 编译：把我们定义的「节点」和「边」编译成一个可执行的「状态机对象（CompiledGraph）」；
  3. 迭代：从初始状态开始，不断地「找到当前节点→执行节点逻辑→修改状态→找到下一个节点」；
  4. 终止：当遇到「终止状态」时，停止迭代，返回最终的状态和结果。
核心概念2：状态（State）—— Multi-Agent系统的“大脑记忆”
问题背景

在单Agent系统中，我们通常用LangChain的Memory组件来存储对话历史和上下文信息——比如ConversationBufferMemory、ConversationSummaryMemory、ConversationBufferWindowMemory等等。但是，在Multi-Agent系统中，我们需要存储的信息远不止对话历史：

  • 我们需要存储用户的结构化需求（比如出发地、目的地、出发日期、预算、人数、偏好）；
  • 我们需要存储每个Agent的执行结果（比如机票搜索结果、景点筛选结果、酒店比价结果）；
  • 我们需要存储当前的决策状态（比如当前是第几次预算审核？已经尝试过哪些调整方案？）；
  • 我们需要存储对话历史（虽然结构化需求更重要，但自然语言的对话历史可以帮助Agent更好地理解上下文）。

如果我们还用LangChain的Memory组件来存储这些信息，会遇到三个问题：

  1. Memory组件是无结构的（或者半结构化的）：比如ConversationBufferMemory存储的是「字符串形式的对话历史」，如果我们想提取其中的「结构化需求」，还得用LLM再解析一遍——效率很低，而且容易出错；
  2. Memory组件是全局的、共享的，但没有明确的访问控制：所有的Agent都可以读取和修改Memory组件里的所有信息——如果某个Agent不小心修改了另一个Agent需要的信息，整个系统就会崩溃；
  3. Memory组件无法支持「状态检查点」和「状态回滚」：如果我们想在某个Agent执行完毕后暂停系统，保存当前的状态，然后在任意时间恢复系统继续运行——Memory组件根本做不到！

而LangGraph的状态对象，正好解决了这三个问题！

概念核心属性

LangGraph的状态对象有五个核心属性：

  1. 结构化的（Structured）：状态对象可以是任意的Python对象，但推荐用Pydantic模型（BaseModel） 或者TypedDict来定义——这样可以确保状态对象的结构是固定的，所有的字段都有明确的类型和默认值；
  2. 全局的、共享的（Global & Shared）：所有的节点（Agent、工具、数据处理逻辑）都可以读取和修改状态对象里的信息——但因为状态对象是结构化的，我们可以通过「字段命名规范」或者「自定义的访问控制逻辑」来避免信息被误修改；
  3. 可序列化的（Serializable）：状态对象必须是可序列化的（比如可以用json.dumps()序列化，或者用pickle序列化）——这样才能支持「状态检查点」和「状态回滚」；
  4. 可合并的（Mergeable）：如果我们有多个节点同时执行（并行执行），每个节点都会修改状态对象里的不同字段——这时候我们需要一个「合并规则」来把这些修改合并成一个新的状态对象；
  5. 有历史的（With History）：LangGraph会自动保存状态对象的历史变化记录——这对于调试和可解释性非常有用！
问题解决：如何定义LangGraph的状态对象？

LangGraph提供了三种定义状态对象的方式，我们可以根据自己的需求选择：

  1. 方式一：使用TypedDict（Python 3.8+内置）：适合简单的、不需要验证字段类型的状态对象；
  2. 方式二：使用Pydantic BaseModel（推荐）：适合复杂的、需要验证字段类型和默认值的状态对象；
  3. 方式三：使用自定义的Python类：适合需要添加「自定义方法」或者「自定义合并规则」的状态对象。

接下来，我们用本文的核心业务问题为例，分别用这三种方式定义状态对象，并且对比它们的优缺点。

方式一：使用TypedDict

TypedDict是Python 3.8+内置的一个类型注解工具，它可以用来定义「有固定键和固定值类型的字典」。LangGraph支持使用TypedDict作为状态对象。

下面是用TypedDict定义的旅行规划系统的状态对象：

# src/state.py
from typing import TypedDict, List, Dict, Optional
from datetime import date

# 定义结构化的旅行需求
class TravelRequirement(TypedDict):
    departure_city: str  # 出发城市
    destination_cities: List[str]  # 目的地城市列表（比如["东京", "首尔"]）
    departure_date: Optional[date]  # 出发日期
    return_date: Optional[date]  # 返回日期
    total_budget: float  # 总预算（单位：元人民币）
    num_adults: int  # 成人人数
    num_children: int  # 儿童人数
    children_ages: List[int]  # 儿童年龄列表
    preferences: List[str]  # 偏好列表（比如["靠近迪士尼", "含早餐", "4星级酒店"]）
    include_tickets: bool  # 是否包含景点门票

# 定义结构化的搜索结果
class SearchResult(TypedDict):
    source: str  # 搜索来源（比如"Serper"、"Expedia"）
    query: str  # 搜索关键词
    results: List[Dict]  # 具体的搜索结果列表

# 定义LangGraph的状态对象
class TravelState(TypedDict):
    # 1. 对话历史（可选，用来帮助Agent理解上下文）
    messages: List[Dict]  # 每个元素是一个字典，比如{"role": "user", "content": "帮我规划..."}
    # 2. 结构化的旅行需求（可能是用户初始输入的，也可能是需求解析Agent解析出来的）
    travel_requirement: Optional[TravelRequirement]
    # 3. 机票搜索结果
    flight_results: Optional[SearchResult]
    # 4. 景点筛选结果
    attraction_results: Optional[SearchResult]
    # 5. 酒店比价结果
    hotel_results: Optional[SearchResult]
    # 6. 当前的总预算估算
    current_total_budget: Optional[float]
    # 7. 当前的决策状态
    decision_state: Optional[str]  # 比如"INITIAL", "FLIGHT_SEARCHED", "ATTRACTION_FILTERED", "HOTEL_COMPARED", "BUDGET_CHECKED", "ADJUSTING_HOTEL", "ADJUSTING_ATTRACTION", "DONE", "FAILED"
    # 8. 已经尝试过的调整方案
    tried_adjustments: List[str]  # 比如["HOTEL_STAR_DOWN", "ATTRACTION_REMOVE", "DATE_CHANGE"]
    # 9. 调整方案的选择（如果需要用户输入）
    adjustment_choice: Optional[str]
    # 10. 最终的旅行规划
    final_plan: Optional[str]

优点：

  • 简单、易用，不需要安装额外的依赖库（Python内置）；
  • 可以用类型注解工具（比如mypy）来检查字段类型。

缺点：

  • 无法验证字段的默认值；
  • 无法验证字段的取值范围（比如总预算必须>0，成人人数必须≥1）；
  • 无法添加自定义的方法（比如计算总预算的方法）；
  • 无法自定义合并规则（如果有多个节点同时执行，只能用LangGraph默认的「字典合并规则」——也就是后面的修改覆盖前面的修改）。
方式二：使用Pydantic BaseModel（推荐）

Pydantic是一个Python的数据验证和设置管理库，它可以用来定义「有固定键、固定值类型、默认值、取值范围验证」的类。LangGraph官方推荐使用Pydantic BaseModel作为状态对象，因为它解决了TypedDict的所有缺点！

接下来，我们用Pydantic BaseModel重新定义旅行规划系统的状态对象——注意，我们用的是Pydantic v2（因为LangGraph 0.2.x版本只支持Pydantic v2）：

# src/state.py
from typing import List, Dict, Optional, Annotated
from datetime import date
from pydantic import BaseModel, Field, field_validator, model_validator
from langgraph.graph import add_messages  # LangGraph提供的专门用来合并消息的函数

# 定义结构化的旅行需求
class TravelRequirement(BaseModel):
    departure_city: str = Field(..., description="出发城市，比如上海、北京")
    destination_cities: List[str] = Field(..., description="目的地城市列表，比如['东京', '首尔']")
    departure_date: Optional[date] = Field(None, description="出发日期，格式为YYYY-MM-DD")
    return_date: Optional[date] = Field(None, description="返回日期，格式为YYYY-MM-DD")
    total_budget: float = Field(..., gt=0, description="总预算，单位为元人民币，必须大于0")
    num_adults: int = Field(..., ge=1, description="成人人数，必须大于等于1")
    num_children: int = Field(..., ge=0, description="儿童人数，必须大于等于0")
    children_ages: List[int] = Field(default_factory=list, description="儿童年龄列表，比如[5, 7]")
    preferences: List[str] = Field(default_factory=list, description="偏好列表，比如['靠近迪士尼', '含早餐', '4星级酒店']")
    include_tickets: bool = Field(default=False, description="是否包含景点门票")

    # 字段验证器：验证儿童年龄列表的长度是否等于儿童人数
    @field_validator("children_ages")
    @classmethod
    def check_children_ages_length(cls, v, info):
        if "num_children" in info.data and len(v) != info.data["num_children"]:
            raise ValueError(f"儿童年龄列表的长度必须等于儿童人数（{info.data['num_children']}）")
        return v

    # 模型验证器：验证返回日期是否晚于出发日期
    @model_validator(mode="after")
    def check_return_date_after_departure_date(self):
        if self.departure_date and self.return_date and self.return_date <= self.departure_date:
            raise ValueError("返回日期必须晚于出发日期")
        return self

# 定义结构化的搜索结果
class SearchResult(BaseModel):
    source: str = Field(..., description="搜索来源，比如Serper、Expedia")
    query: str = Field(..., description="搜索关键词")
    results: List[Dict] = Field(default_factory=list, description="具体的搜索结果列表")

# 定义LangGraph的状态对象——注意：我们用Annotated来指定字段的合并规则！
class TravelState(BaseModel):
    # 1. 对话历史：用LangGraph提供的add_messages函数来合并消息
    messages: Annotated[List[Dict], add_messages] = Field(default_factory=list, description="对话历史，每个元素是一个字典，比如{'role': 'user', 'content': '帮我规划...'}")
    # 2. 结构化的旅行需求：用"replace"规则合并（也就是后面的修改覆盖前面的修改）
    travel_requirement: Annotated[Optional[TravelRequirement], "replace"] = Field(None, description="结构化的旅行需求")
    # 3. 机票搜索结果：用"replace"规则合并
    flight_results: Annotated[Optional[SearchResult], "replace"] = Field(None, description="机票搜索结果")
    # 4. 景点筛选结果：用"replace"规则合并
    attraction_results: Annotated[Optional[SearchResult], "replace"] = Field(None, description="景点筛选结果")
    # 5. 酒店比价结果：用"replace"规则合并
    hotel_results: Annotated[Optional[SearchResult], "replace"] = Field(None, description="酒店比价结果")
    # 6. 当前的总预算估算：用"replace"规则合并
    current_total_budget: Annotated[Optional[float], "replace"] = Field(None, description="当前的总预算估算，单位为元人民币")
    # 7. 当前的决策状态：用"replace"规则合并
    decision_state: Annotated[Optional[str], "replace"] = Field("INITIAL", description="当前的决策状态，比如INITIAL、FLIGHT_SEARCHED、DONE、FAILED")
    # 8. 已经尝试过的调整方案：用"extend"规则合并（也就是后面的列表会扩展到前面的列表后面）
    tried_adjustments: Annotated[List[str], "extend"] = Field(default_factory=list, description="已经尝试过的调整方案，比如['HOTEL_STAR_DOWN', 'ATTRACTION_REMOVE']")
    # 9. 调整方案的选择：用"replace"规则合并
    adjustment_choice: Annotated[Optional[str], "replace"] = Field(None, description="调整方案的选择，如果需要用户输入")
    # 10. 最终的旅行规划：用"replace"规则合并
    final_plan: Annotated[Optional[str], "replace"] = Field(None, description="最终的旅行规划")

    # 模型验证器：验证已经尝试过的调整方案是否有重复
    @model_validator(mode="after")
    def check_tried_adjustments_unique(self):
        if len(self.tried_adjustments) != len(set(self.tried_adjustments)):
            raise ValueError("已经尝试过的调整方案不能有重复")
        return self

注意！ 这里有一个LangGraph的关键特性：使用Annotated来指定字段的合并规则！

LangGraph默认提供了三种合并规则：

  1. add_messages：专门用来合并对话历史的规则——它会把新的消息列表添加到旧的消息列表后面，并且会自动处理「相同ID的消息被替换」的情况；
  2. "replace"：最简单的合并规则——新的值会完全覆盖旧的值；
  3. "extend"：用来合并列表的规则——新的列表会扩展到旧的列表后面（注意：如果旧的值不是列表，会抛出异常）。

除了这三种默认的合并规则，你还可以自定义合并规则——只需要写一个Python函数，接受两个参数（旧的值和新的值），返回合并后的新值即可。比如，你可以写一个自定义的合并规则，用来合并两个字典（只更新新字典里有的字段，保留旧字典里的其他字段）：

# 自定义合并规则：只更新新字典里有的字段，保留旧字典里的其他字段
def merge_dicts(old: Optional[Dict], new: Optional[Dict]) -> Dict:
    if old is None:
        old = {}
    if new is None:
        new = {}
    return {**old, **new}

# 然后在状态对象里使用这个自定义合并规则
class MyState(BaseModel):
    my_dict: Annotated[Optional[Dict], merge_dicts] = Field(default_factory=dict)

优点：

  • 可以验证字段的类型、默认值、取值范围；
  • 可以添加自定义的字段验证器和模型验证器；
  • 可以添加自定义的方法；
  • 可以指定字段的合并规则（包括默认规则和自定义规则）；
  • 支持序列化和反序列化（用Pydantic的model_dump()和model_validate()方法）；
  • LangGraph官方推荐，兼容性最好！

缺点：

  • 需要安装额外的依赖库（不过我们已经在准备工作里安装了langchain-openai，它会自动安装Pydantic v2）；
  • 稍微比TypedDict复杂一点，但完全值得！
方式三：使用自定义的Python类

如果你需要添加「非常复杂的自定义方法」或者「非常复杂的自定义合并规则」，可以使用自定义的Python类作为状态对象——不过这种方式用得比较少，因为Pydantic BaseModel已经能满足绝大多数需求了。

下面是一个简单的自定义Python类作为状态对象的例子：

# src/state.py
from typing import List, Dict, Optional
from langgraph.graph import add_messages

class CustomTravelState:
    def __init__(
        self,
        messages: Optional[List[Dict]] = None,
        travel_requirement: Optional[Dict] = None,
        # 其他字段...
    ):
        self.messages = messages or []
        self.travel_requirement = travel_requirement or {}
        # 其他字段初始化...

    # 自定义合并规则：合并整个状态对象
    @classmethod
    def merge(cls, old: "CustomTravelState", new: "CustomTravelState") -> "CustomTravelState":
        merged_messages = add_messages(old.messages, new.messages)
        merged_travel_requirement = {**old.travel_requirement, **new.travel_requirement}
        # 合并其他字段...
        return cls(
            messages=merged_messages,
            travel_requirement=merged_travel_requirement,
            # 其他字段...
        )

    # 自定义方法：计算总预算估算
    def calculate_total_budget(self) -> float:
        # 这里写计算总预算的逻辑
        return 0.0

优点：

  • 完全自由，可以添加任何你想要的自定义方法和合并规则；

缺点：

  • 需要自己实现序列化和反序列化；
  • 需要自己实现合并规则；
  • LangGraph的兼容性不如Pydantic BaseModel；
  • 调试起来比较麻烦。
概念对比表格

为了帮助你更好地选择状态对象的定义方式，我们把三种方式的优缺点整理成了一个Markdown表格：

定义方式 简单性 类型验证 默认值验证 取值范围验证 自定义方法 自定义合并规则 序列化支持 LangGraph兼容性 推荐指数
TypedDict ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ ❌ ❌ ❌ ❌ ⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐
Pydantic BaseModel（推荐） ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐
自定义Python类 ⭐⭐ ❌ ❌ ❌ ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐⭐ ⭐⭐ ⭐⭐⭐ ⭐⭐

结论：在绝大多数情况下，我们都应该选择Pydantic BaseModel作为LangGraph的状态对象！

（由于篇幅限制，本文的剩余内容——包括节点、边、图的核心概念，旅行规划系统的具体实现，最佳实践，行业发展趋势等——将在后续的「下篇」文章中发布。不过，为了满足你当前的需求，我可以先给你提供一个完整的、可直接运行的旅行规划系统的代码框架，以及一个简略的后续内容大纲。）

# 人工智能
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

大模型学习及就业路线

大模型学习及就业路线简介

avatar AtomGit开源社区

AI本地部署快速步骤

本文详细介绍了AI本地知识库的快速搭建方法，分为模型运行环境和知识库管理两部分。首先通过Ollama部署大语言模型和嵌入模型，配置环境变量并下载所需模型。然后使用Anything LLM作为知识库管理工具，完成模型对接、工作区创建和文档上传。最后通过优化检索设置提升查询准确率，实现本地化、隐私安全的文档问答功能。整个流程提供详细操作截图，适合新手按步骤完成部署。

avatar AtomGit开源社区

基于深度学习的可回收垃圾材质识别与分类研究毕业设计--整套 C/S 架构完整方案

架构组成：• Server 服务端：Flask + 你训练好的 ResNet18 五分类模型（后端接口）• Client 客户端：PyQt5 桌面界面（不本地加载模型，只调接口）• 流程：PyQt 选图 → 传给 Flask 后端 → 后端 AI 识别 → 返回类别 + 置信度 → 界面展示安装依赖。

avatar AtomGit开源社区