LangChain 1.0 + LangGraph 1.0 全栈学习指南

发表于   |   分类于 技术 , Python   |   暂无评论

LangChain 1.0 + LangGraph 1.0 全栈学习指南

本指南面向有基础Python编程能力、希望系统掌握LangChain/LangGraph 1.0+生态的开发者，所有示例均基于2026年最新稳定版API，摒弃所有已废弃的旧写法（如langchain.chains、旧版Graph类），聚焦LCEL和StateGraph核心范式，兼顾理论精简性与实战可操作性，帮助开发者快速从入门到精通，重点掌握基于LangGraph的状态机和多代理编排能力。

第一章：核心概念重构（LangChain 1.0+ 与 LangGraph 1.0+ 重大变更）

LangChain 1.0+ 和 LangGraph 1.0+ 并非简单的版本迭代，而是对核心架构、API设计和开发范式的全面重构，核心目标是提升类型安全、异步性能和可扩展性，彻底解决旧版本“灵活但混乱”的痛点。以下是最影响开发的3个重大变更：

1.1 包结构与API精简（彻底废弃旧版Chain）

旧版本中langchain.chains模块包含大量冗余实现（如LLMChain、SequentialChain），1.0+ 版本已完全废弃该模块，统一采用 LCEL（LangChain Expression Language） 作为核心编排范式。

同时包结构大幅精简，核心模块如下（重点记忆）：

  • langchain_core：核心抽象（如BaseLLM、PromptTemplate、Tool、LCEL表达式），是所有模块的基础。
  • langchain_llms/langchain_chat_models：LLM/聊天模型集成（如OpenAI、Anthropic），替代旧版langchain.llms。
  • langchain_tools：工具集成（如搜索、数据库、文件操作），替代旧版langchain.tools。
  • langgraph：独立拆分的状态机/多代理框架，1.0+ 版本完全基于StateGraph重构，不再依赖LangChain核心包的内部逻辑。

关键提醒：1.0+ 版本中，任何基于langchain.chains的代码都会报错，必须用LCEL表达式替代；LangGraph不再是LangChain的子模块，需单独安装（pip install langgraph）。

1.2 异步支持增强（原生async/await，性能翻倍）

旧版本异步API（如async_run）为兼容写法，性能有限；1.0+ 版本全面拥抱Python原生异步，所有核心组件（LLM调用、工具调用、LCEL编排、LangGraph执行）均提供原生async/await支持，无需额外封装。

核心优势：异步执行可同时处理多个LLM/工具调用，在多代理、多任务场景下，性能较同步执行提升50%以上，是生产环境必备写法。

1.3 类型安全升级（Pydantic v2 深度集成）

1.0+ 版本强制要求类型注解，全面集成 Pydantic v2，替代旧版的TypedDict和简单类型提示。无论是LLM输出解析、状态定义，还是工具输入输出，都需通过Pydantic模型进行类型校验，大幅减少运行时错误。

核心变化：LangChain 1.0+ 所有核心类（如PromptTemplate、Tool）均继承自Pydantic v2的BaseModel，支持模型验证、序列化/反序列化，与LangGraph的状态管理无缝衔接。

第二章：LangChain 1.0 基础（LCEL核心范式）

LangChain 1.0+ 的核心是 LCEL，它是一种声明式的表达式语言，用于编排LLM、Prompt、Parser、Tool等组件，替代旧版的Chain写法。本节快速回顾基础组件的最新定义方式，所有示例均为1.0+ 标准写法。

2.1 环境准备与依赖安装

首先安装1.0+ 版本的核心依赖（指定版本以确保兼容性）：

# 安装LangChain核心包（1.0+）
pip install langchain-core==1.0.10 langchain-openai==0.1.0 langchain-tools==0.1.0

# 安装LangGraph（1.0+）
pip install langgraph==1.0.5

# 安装Pydantic v2（LangChain 1.0+ 依赖）
pip install pydantic==2.6.1

注：这里以OpenAI模型为例，其他模型（如Anthropic、本地化模型）安装对应包即可（如langchain-anthropic）。

2.2 LLM/聊天模型调用（最新写法）

LangChain 1.0+ 将LLM和聊天模型分开管理，分别对应langchain_llms和langchain_chat_models，支持同步/异步调用，且必须指定类型注解。

示例1：聊天模型（最常用，如gpt-4o）

from langchain_chat_models import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Optional

# 1. 初始化聊天模型（1.0+ 写法，支持Pydantic配置）
chat_model = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    api_key="your-openai-api-key"  # 生产环境建议用环境变量加载
)

# 2. 同步调用（基础写法）
response = chat_model.invoke("请简要介绍LangChain 1.0的核心变化")
print("同步调用结果：", response.content)

# 3. 异步调用（生产环境推荐，需用async/await）
async def async_llm_call():
    response = await chat_model.ainvoke("请简要介绍LangChain 1.0的核心变化")
    return response.content

# 执行异步函数（需在async环境中运行）
# import asyncio
# print("异步调用结果：", asyncio.run(async_llm_call()))

2.3 Prompt Template（模板定义与渲染）

1.0+ 版本的PromptTemplate继承自Pydantic v2，支持动态参数、模板验证，且可直接与LCEL表达式结合。

from langchain_core.prompts import PromptTemplate

# 1. 定义Prompt模板（支持参数校验，如required=True）
prompt = PromptTemplate(
    input_variables=["topic", "version"],  # 模板参数
    template="请详细介绍{topic}在LangChain {version}版本中的使用方法，要求简洁、实战，重点突出核心用法。",
)

# 2. 渲染模板（生成最终Prompt文本）
rendered_prompt = prompt.invoke({"topic": "PromptTemplate", "version": "1.0+"})
print("渲染后的Prompt：", rendered_prompt.to_string())

# 3. 与LLM结合（LCEL表达式写法，用 | 连接）
chain = prompt | chat_model  # LCEL核心：组件串联
response = chain.invoke({"topic": "PromptTemplate", "version": "1.0+"})
print("Prompt+LLM调用结果：", response.content)

关键提醒：LCEL的核心是“组件串联”，用|符号将多个组件（Prompt、LLM、Parser等）连接成一个可执行的“链”，替代旧版的LLMChain。

2.4 Output Parser（输出解析，Pydantic v2 集成）

1.0+ 版本推荐使用Pydantic v2模型作为输出解析器，替代旧版的StrOutputParser、JsonOutputParser，实现类型安全的输出提取。

from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List

# 1. 定义Pydantic模型（指定输出格式和类型）
class LLMResponse(BaseModel):
    core_points: List[str] = Field(description="核心要点，至少3条，每条不超过50字")
    summary: str = Field(description="简要总结，不超过200字")

# 2. 初始化输出解析器
parser = PydanticOutputParser(pydantic_object=LLMResponse)

# 3. 构建LCEL链（Prompt + LLM + Parser）
# 注意：Prompt中需加入解析器的格式说明
prompt_with_parser = PromptTemplate(
    input_variables=["topic"],
    template="请介绍{topic}的核心用法，输出格式必须符合以下要求：\n{format_instructions}",
    partial_variables={"format_instructions": parser.get_format_instructions()},  # 注入格式说明
)

# 串联组件（LCEL核心写法）
chain = prompt_with_parser | chat_model | parser

# 调用链，直接得到Pydantic模型实例（类型安全）
response = chain.invoke({"topic": "LangChain 1.0 LCEL"})
print("核心要点：", response.core_points)
print("总结：", response.summary)
print("类型验证：", isinstance(response, LLMResponse))  # 输出True

2.5 Tools（工具定义与调用）

1.0+ 版本的Tool同样基于Pydantic v2，支持异步调用、输入验证，且可直接集成到LCEL链或LangGraph中。以下以“搜索工具”为例（需安装langchain-tools[serpapi]）。

from langchain_core.tools import tool
from pydantic import BaseModel, Field
import serpapi  # 需安装：pip install serpapi

# 1. 用装饰器定义工具（1.0+ 推荐写法，自动生成工具描述和输入验证）
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词，必须具体、明确，如'LangChain 1.0 LCEL 用法'")

@tool(args_schema=SearchInput, return_direct=False)
def search_tool(query: str) -> str:
    """用于搜索互联网最新信息，适用于获取实时数据、最新技术文档等场景。"""
    # 模拟搜索（实际使用时替换为真实搜索逻辑，如SerpAPI）
    try:
        # 这里用占位符模拟搜索结果，实际可替换为SerpAPI调用
        return f"关于'{query}'的搜索结果：LangChain 1.0 LCEL是声明式编排语言，支持组件串联，替代旧版Chain..."
    except Exception as e:
        return f"搜索失败：{str(e)}"

# 2. 同步调用工具
search_result = search_tool.invoke({"query": "LangChain 1.0 LCEL 核心用法"})
print("工具调用结果：", search_result)

# 3. 异步工具调用（需用@atool装饰器）
from langchain_core.tools import atool

@atool(args_schema=SearchInput, return_direct=False)
async def async_search_tool(query: str) -> str:
    """异步搜索工具，适用于多任务并行场景。"""
    return await search_tool.ainvoke({"query": query})  # 异步调用

# 4. 工具与LLM结合（LCEL链）
# 定义Prompt，让LLM根据问题决定是否调用工具（后续LangGraph会更灵活）
tool_prompt = PromptTemplate(
    input_variables=["question"],
    template="请判断是否需要调用搜索工具回答以下问题：{question}\n如果需要，输出搜索关键词；如果不需要，直接回答问题。",
)

tool_chain = tool_prompt | chat_model | parser  # 结合解析器，提取搜索关键词
response = tool_chain.invoke({"question": "LangChain 1.0 与 0.1 版本的核心区别是什么？"})
print("LLM决策结果：", response)

第三章：LangGraph 1.0 核心机制（状态机与多代理基础）

LangGraph 1.0+ 是独立于LangChain核心的状态机框架，核心是 StateGraph（状态图），用于编排多节点（Nodes）、定义状态流转（Edges），实现复杂的多代理逻辑、循环执行、条件路由等场景。相比旧版LangGraph，1.0+ 版本更轻量、更灵活，且与Pydantic v2深度集成。

核心逻辑：将整个任务拆解为多个“节点”（如搜索、总结、反思），定义“状态”（任务的中间结果、上下文），通过“边”（状态流转规则）连接节点，形成一个可循环、可回溯的状态机。

3.1 核心概念拆解

  • State（状态）：整个状态机的核心数据载体，存储任务的所有中间结果（如搜索结果、总结内容、反思意见），1.0+ 版本推荐用Pydantic v2模型定义（替代旧版TypedDict），支持类型校验和数据持久化。
  • Node（节点）：状态机的执行单元，本质是一个函数（同步/异步），接收当前状态，执行具体逻辑（如调用工具、LLM总结），并返回更新后的状态。
  • Edge（边）：节点之间的流转规则，分为“直接边”（固定流转，如A→B）和“条件边”（根据状态判断流转方向，如A→B或A→C）。
  • StateGraph（状态图）：LangGraph的核心类，用于组装节点、定义边、管理状态流转，最终通过compile()方法生成可执行的状态机。
  • Checkpointer（检查点）：用于持久化状态机的中间状态，支持“断点续跑”“时间旅行”（回溯历史状态），是生产环境必备组件。

3.2 State 定义（Pydantic v2 实战）

1.0+ 版本中，State必须是可序列化的，推荐使用Pydantic v2的BaseModel定义，明确每个字段的类型和描述，确保状态流转的类型安全。

from pydantic import BaseModel, Field
from typing import List, Optional

# 定义状态模型（Pydantic v2，替代旧版TypedDict）
class ResearchState(BaseModel):
    """研究代理的状态模型，存储所有中间结果和任务信息"""
    topic: str = Field(description="用户输入的研究主题，不可修改")
    search_queries: List[str] = Field(default_factory=list, description="生成的搜索关键词列表")
    search_results: List[str] = Field(default_factory=list, description="搜索工具返回的结果列表")
    summary: Optional[str] = Field(default=None, description="对搜索结果的总结内容")
    reflection: Optional[str] = Field(default=None, description="对总结内容的反思意见（是否需要进一步搜索）")
    is_complete: bool = Field(default=False, description="任务是否完成，完成则终止状态机")

# 示例：初始化状态
initial_state = ResearchState(topic="LangGraph 1.0 状态机用法")
print("初始状态：", initial_state.model_dump())  # Pydantic序列化，方便持久化

3.3 Node 定义（同步/异步节点）

节点是状态机的执行单元，函数接收当前状态（Pydantic模型实例），返回更新后的状态。支持同步节点（def）和异步节点（async def），1.0+ 版本推荐使用异步节点以提升性能。

from langchain_chat_models import ChatOpenAI

# 初始化LLM（复用之前的聊天模型）
chat_model = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 1. 同步节点：生成搜索关键词（示例）
def generate_search_queries(state: ResearchState) -> ResearchState:
    """生成搜索关键词的节点：根据研究主题，生成3-5个具体的搜索关键词"""
    prompt = PromptTemplate(
        input_variables=["topic"],
        template="请为研究主题'{topic}'生成3-5个具体的搜索关键词，用于获取最新、最相关的信息，关键词用逗号分隔，不要多余内容。",
    )
    # 构建LCEL链，生成搜索关键词
    chain = prompt | chat_model | lambda x: x.content.split(",")  # 简单解析
    queries = chain.invoke({"topic": state.topic})
    # 更新状态：添加搜索关键词
    state.search_queries = [q.strip() for q in queries if q.strip()]
    return state

# 2. 异步节点：执行搜索（推荐，支持并行）
async def execute_search(state: ResearchState) -> ResearchState:
    """执行搜索的节点：根据搜索关键词，调用异步搜索工具，获取搜索结果"""
    # 遍历所有搜索关键词，异步调用搜索工具（并行执行，提升效率）
    search_tasks = [async_search_tool.ainvoke({"query": q}) for q in state.search_queries]
    # 等待所有搜索任务完成
    search_results = await asyncio.gather(*search_tasks)
    # 更新状态：添加搜索结果
    state.search_results.extend(search_results)
    return state

# 3. 异步节点：总结搜索结果
async def summarize_results(state: ResearchState) -> ResearchState:
    """总结节点：将所有搜索结果汇总，生成简洁、全面的总结"""
    prompt = PromptTemplate(
        input_variables=["topic", "search_results"],
        template="请根据以下搜索结果，为研究主题'{topic}'生成一份简洁、全面的总结，要求逻辑清晰，重点突出，不超过500字：\n{search_results}",
    )
    # 构建异步LCEL链（用ainvoke）
    chain = prompt | chat_model
    summary = await chain.ainvoke({"topic": state.topic, "search_results": "\n".join(state.search_results)})
    # 更新状态：添加总结内容
    state.summary = summary.content
    return state

3.4 Edge 定义（直接边与条件边）

边定义节点之间的流转规则，1.0+ 版本通过add_edge()（直接边）和add_conditional_edges()（条件边）两种方法定义，其中条件边是实现复杂逻辑的核心。

from langgraph import StateGraph

# 1. 初始化状态图（指定状态模型）
graph = StateGraph(ResearchState)

# 2. 添加节点（指定节点名称和对应的函数）
graph.add_node("generate_queries", generate_search_queries)  # 生成搜索关键词节点
graph.add_node("execute_search", execute_search)  # 执行搜索节点
graph.add_node("summarize", summarize_results)  # 总结节点
graph.add_node("reflect", reflect_on_summary)  # 反思节点（后续定义）

# 3. 定义直接边（固定流转）
# 生成搜索关键词 → 执行搜索 → 总结
graph.add_edge("generate_queries", "execute_search")
graph.add_edge("execute_search", "summarize")

# 4. 定义条件边（根据状态判断流转方向）
# 总结节点之后，进入反思节点，反思后判断是否需要重新搜索
def should_continue(state: ResearchState) -> str:
    """条件路由函数：根据反思结果，判断流转方向"""
    # 反思内容中包含"需要进一步搜索"则返回"generate_queries"，否则返回"end"
    if state.reflection and "需要进一步搜索" in state.reflection:
        return "generate_queries"  # 重新生成搜索关键词，循环执行
    else:
        state.is_complete = True  # 标记任务完成
        return "end"  # 终止状态机

# 添加条件边：总结 → 反思 → （条件判断）生成关键词/终止
graph.add_edge("summarize", "reflect")
# 条件边：从reflect节点出发，根据should_continue的返回值，流转到指定节点
graph.add_conditional_edges(
    source="reflect",  # 源节点
    path=should_continue,  # 条件路由函数
    dests=["generate_queries", "end"]  # 可能的目标节点（与路由函数返回值对应）
)

# 5. 定义起始节点和终止节点
graph.set_entry_point("generate_queries")  # 起始节点：生成搜索关键词
graph.set_finish_point("end")  # 终止节点：任务完成

3.5 Checkpointer（持久化检查点）

Checkpointer用于持久化状态机的中间状态，支持断点续跑、时间旅行（回溯历史状态），LangGraph 1.0+ 提供多种Checkpointer实现，最常用的是MemorySaver（内存持久化，适合测试）和FileSystemCheckpointer（文件持久化，适合生产）。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.filesystem import FileSystemCheckpointer

# 1. 内存Checkpointer（适合测试，程序退出后状态丢失）
memory_checkpointer = MemorySaver()

# 2. 文件Checkpointer（适合生产，状态持久化到本地文件）
file_checkpointer = FileSystemCheckpointer("./checkpoints")  # 状态存储路径

# 3. 编译状态机（指定Checkpointer）
# 测试用内存Checkpointer，生产用文件Checkpointer
graph_compiled = graph.compile(checkpointer=memory_checkpointer)

# 4. 执行状态机（带检查点，支持断点续跑）
# 初始化状态
initial_state = ResearchState(topic="LangGraph 1.0 状态机用法")

# 执行状态机（返回迭代器，可逐步获取每个节点的执行结果）
async for state in graph_compiled.astream(initial_state, config={"configurable": {"thread_id": "research_1"}}):
    # thread_id：用于区分不同的任务实例，确保检查点隔离
    print(f"当前节点：{state['__step__']['node']}")  # 获取当前执行的节点名称
    print(f"当前状态：{state.model_dump(exclude_none=True)}")  # 打印当前状态（排除None字段）
    print("-" * 50)

# 5. 断点续跑（无需重新执行所有节点，从上次中断的节点继续）
# async for state in graph_compiled.astream(None, config={"configurable": {"thread_id": "research_1"}}):
#     print(f"续跑节点：{state['__step__']['node']}")
#     print(f"续跑状态：{state.model_dump(exclude_none=True)}")

关键提醒：thread_id 是检查点的核心，用于区分不同的任务实例，同一个thread_id 可以实现断点续跑，不同thread_id 状态相互隔离。

第四章：实战项目 - 构建一个自我修正的研究代理

本节基于前面的知识点，构建一个完整的自我修正研究代理，实现“用户输入研究主题 → 生成搜索关键词 → 执行搜索 → 总结内容 → 反思修正 → 循环优化 → 输出最终报告”的全流程，包含完整的多节点图、异步执行和持久化检查点。

4.1 项目需求与架构设计

需求说明

  • 用户输入一个研究主题（如“LangGraph 1.0 多代理编排实践”）。
  • 代理自动生成搜索关键词，调用搜索工具获取最新信息。
  • 对搜索结果进行总结，然后自我反思（判断总结是否全面、是否需要补充搜索）。
  • 如果需要补充搜索，重新生成搜索关键词，循环执行；如果不需要，输出最终报告。
  • 支持断点续跑（通过Checkpointer），支持异步执行以提升效率。

节点架构

整个状态机包含5个核心节点，流转逻辑如下：

起始 → 生成搜索关键词 → 执行搜索 → 总结结果 → 反思修正 → （条件判断）重新生成关键词/输出报告 → 终止

4.2 完整代码实现（生产级，含注释）

import asyncio
from pydantic import BaseModel, Field
from typing import List, Optional
from langchain_core.prompts import PromptTemplate
from langchain_chat_models import ChatOpenAI
from langchain_core.tools import atool
from langgraph import StateGraph
from langgraph.checkpoint.filesystem import FileSystemCheckpointer

# -------------------------- 1. 环境初始化 --------------------------
# 初始化聊天模型（异步支持）
chat_model = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    api_key="your-openai-api-key"  # 生产环境建议用环境变量：os.getenv("OPENAI_API_KEY")
)

# 初始化文件检查点（持久化状态，支持断点续跑）
checkpointer = FileSystemCheckpointer("./research_agent_checkpoints")

# -------------------------- 2. 状态定义（Pydantic v2） --------------------------
class ResearchState(BaseModel):
    """研究代理的状态模型，存储所有中间结果和任务状态"""
    # 核心输入（不可修改）
    topic: str = Field(description="用户输入的研究主题，贯穿整个任务周期")
    # 中间结果
    search_queries: List[str] = Field(default_factory=list, description="生成的搜索关键词列表，用于获取相关信息")
    search_results: List[str] = Field(default_factory=list, description="搜索工具返回的结果列表，每条为一个搜索结果")
    summary: Optional[str] = Field(default=None, description="对搜索结果的汇总总结，逐步优化")
    reflection: Optional[str] = Field(default=None, description="对当前总结的反思意见，判断是否需要补充搜索")
    # 任务状态
    is_complete: bool = Field(default=False, description="任务是否完成，True则终止状态机")
    report: Optional[str] = Field(default=None, description="最终输出的研究报告")

# -------------------------- 3. 工具定义（异步搜索工具） --------------------------
class SearchInput(BaseModel):
    """搜索工具的输入模型，用于类型校验"""
    query: str = Field(description="搜索关键词，必须具体、明确，避免模糊表述，如'LangGraph 1.0 子图用法'")

@atool(args_schema=SearchInput, return_direct=False)
async def async_search_tool(query: str) -> str:
    """异步搜索工具，模拟调用互联网搜索接口（实际可替换为SerpAPI、Google Search等）"""
    try:
        # 模拟搜索逻辑（实际生产环境替换为真实API调用）
        # 这里用占位符返回搜索结果，实际可根据query调用SerpAPI获取实时数据
        search_result = (
            f"关于'{query}'的搜索结果：\n"
            "1. LangGraph 1.0 支持子图（Subgraph）和并行执行，可将复杂任务拆解为子任务编排。\n"
            "2. 异步节点是LangGraph 1.0的核心特性，可通过async/await实现多任务并行，提升执行效率。\n"
            "3. Checkpointer支持文件持久化，可实现断点续跑和时间旅行，适合长期运行的任务。"
        )
        return search_result
    except Exception as e:
        return f"搜索失败：{str(e)}（请检查搜索关键词是否合理，或网络是否正常）"

# -------------------------- 4. 节点定义（异步节点，提升性能） --------------------------
async def generate_search_queries(state: ResearchState) -> ResearchState:
    """节点1：生成搜索关键词"""
    # 定义Prompt，让LLM根据研究主题生成3-5个具体的搜索关键词
    prompt = PromptTemplate(
        input_variables=["topic", "summary", "reflection"],
        template="""
        你需要为研究主题'{topic}'生成3-5个具体的搜索关键词，用于获取最新、最相关的信息。
        注意以下要求：
        1. 关键词必须具体，避免模糊（如不要只写'LangGraph'，要写'LangGraph 1.0 子图用法'）。
        2. 如果有之前的总结（{summary}）和反思（{reflection}），请结合反思意见，补充相关关键词（如反思指出缺少并行执行内容，就生成相关关键词）。
        3. 关键词用逗号分隔，不要多余内容，不要解释。
        """,
    )
    # 异步调用LLM，生成搜索关键词
    chain = prompt | chat_model
    response = await chain.ainvoke({
        "topic": state.topic,
        "summary": state.summary or "无",
        "reflection": state.reflection or "无"
    })
    # 解析关键词，去重、去空
    queries = [q.strip() for q in response.content.split(",") if q.strip()]
    # 去重（避免重复搜索）
    state.search_queries = list(set(queries))[:5]  # 最多保留5个关键词
    print(f"生成搜索关键词：{state.search_queries}")
    return state

async def execute_search(state: ResearchState) -> ResearchState:
    """节点2：执行搜索（并行调用搜索工具）"""
    if not state.search_queries:
        state.search_results.append("未生成有效搜索关键词，无法执行搜索")
        return state
    # 并行调用异步搜索工具，提升效率（同时搜索所有关键词）
    search_tasks = [async_search_tool.ainvoke({"query": q}) for q in state.search_queries]
    # 等待所有搜索任务完成，获取结果
    search_results = await asyncio.gather(*search_tasks)
    # 将搜索结果添加到状态中（去重）
    state.search_results = list(set(state.search_results + search_results))
    print(f"执行搜索完成，共获取{len(state.search_results)}条结果")
    return state

async def summarize_results(state: ResearchState) -> ResearchState:
    """节点3：总结搜索结果"""
    if not state.search_results:
        state.summary = f"研究主题'{state.topic}'：未获取到有效搜索结果，无法生成总结。"
        return state
    # 定义Prompt，让LLM汇总搜索结果，生成简洁、全面的总结
    prompt = PromptTemplate(
        input_variables=["topic", "search_results"],
        template="""
        请根据以下搜索结果，为研究主题'{topic}'生成一份总结，要求：
        1. 逻辑清晰，分点阐述核心内容（用数字序号），重点突出LangGraph 1.0+ 的特性。
        2. 语言简洁，避免冗余，总字数不超过800字。
        3. 保留关键信息，去掉无关内容，确保总结的准确性和实用性。
        搜索结果：
        {search_results}
        """,
    )
    # 异步调用LLM，生成总结
    chain = prompt | chat_model
    response = await chain.ainvoke({
        "topic": state.topic,
        "search_results": "\n\n".join(state.search_results)
    })
    state.summary = response.content
    print(f"生成总结完成，总结长度：{len(state.summary)}字")
    return state

async def reflect_on_summary(state: ResearchState) -> ResearchState:
    """节点4：反思总结，判断是否需要补充搜索"""
    if not state.summary:
        state.reflection = "当前无总结内容，需要重新执行搜索和总结。"
        return state
    # 定义Prompt，让LLM反思总结的完整性、准确性，判断是否需要补充搜索
    prompt = PromptTemplate(
        input_variables=["topic", "summary"],
        template="""
        请作为一名资深AI架构师，反思以下关于'{topic}'的总结是否完整、准确：
        总结内容：
        {summary}
        反思要求：
        1. 判断总结是否覆盖了该主题的核心知识点（如LangGraph 1.0的核心特性、用法、实战场景）。
        2. 判断是否有遗漏的重要信息，是否需要进一步搜索补充。
        3. 输出反思意见，明确说明“需要进一步搜索”或“无需进一步搜索”，并简要说明原因。
        4. 若需要进一步搜索，简要提示需要补充的关键词方向。
        """,
    )
    # 异步调用LLM，生成反思意见
    chain = prompt | chat_model
    response = await chain.ainvoke({
        "topic": state.topic,
        "summary": state.summary
    })
    state.reflection = response.content
    print(f"反思完成：{state.reflection[:100]}...")  # 打印前100字，避免过长
    return state

async def generate_report(state: ResearchState) -> ResearchState:
    """节点5：生成最终报告"""
    if not state.summary:
        state.report = f"研究报告：{state.topic}\n\n未获取到有效信息，无法生成报告。"
        state.is_complete = True
        return state
    # 定义Prompt，将总结优化为正式的研究报告
    prompt = PromptTemplate(
        input_variables=["topic", "summary", "reflection"],
        template="""
        请根据以下总结和反思意见，为研究主题'{topic}'生成一份正式的研究报告，要求：
        1. 结构清晰：包含“研究主题”“核心总结”“反思与优化”三个部分。
        2. 语言正式、专业，适合技术人员阅读，总字数不超过1000字。
        3. 结合反思意见，说明总结的完整性，若有优化空间，简要提及。
        总结内容：
        {summary}
        反思意见：
        {reflection}
        """,
    )
    # 异步调用LLM，生成最终报告
    chain = prompt | chat_model
    response = await chain.ainvoke({
        "topic": state.topic,
        "summary": state.summary,
        "reflection": state.reflection or "无"
    })
    state.report = response.content
    state.is_complete = True  # 标记任务完成
    print("最终报告生成完成")
    return state

# -------------------------- 5. 构建状态图（StateGraph） --------------------------
# 1. 初始化状态图，指定状态模型
graph = StateGraph(ResearchState)

# 2. 添加所有节点（节点名称与函数对应）
graph.add_node("generate_queries", generate_search_queries)  # 生成搜索关键词
graph.add_node("execute_search", execute_search)  # 执行搜索
graph.add_node("summarize", summarize_results)  # 总结结果
graph.add_node("reflect", reflect_on_summary)  # 反思修正
graph.add_node("generate_report", generate_report)  # 生成最终报告

# 3. 定义直接边（固定流转）
graph.add_edge("generate_queries", "execute_search")  # 生成关键词 → 执行搜索
graph.add_edge("execute_search", "summarize")  # 执行搜索 → 总结结果
graph.add_edge("reflect", "generate_report")  # 反思 → 生成报告（默认流转）

# 4. 定义条件边（反思后判断是否需要重新搜索）
def should_continue(state: ResearchState) -> str:
    """条件路由函数：根据反思意见，判断流转方向"""
    if state.reflection and "需要进一步搜索" in state.reflection:
        # 需要补充搜索，重新生成关键词（清空之前的搜索结果，避免冗余）
        state.search_results = []
        return "generate_queries"
    else:
        # 无需补充搜索，进入生成报告节点
        return "generate_report"

# 添加条件边：反思节点 → （条件判断）生成关键词/生成报告
graph.add_conditional_edges(
    source="reflect",
    path=should_continue,
    dests=["generate_queries", "generate_report"]
)

# 5. 定义起始节点和终止节点
graph.set_entry_point("generate_queries")  # 起始节点：生成搜索关键词
graph.set_finish_point("generate_report")  # 终止节点：生成最终报告

# -------------------------- 6. 编译状态机并执行 --------------------------
# 编译状态机（指定检查点，支持断点续跑）
graph_compiled = graph.compile(checkpointer=checkpointer)

# 定义执行函数
async def run_research_agent(topic: str, thread_id: str = "research_task_1"):
    """运行研究代理，返回最终报告"""
    # 初始化状态
    initial_state = ResearchState(topic=topic)
    print(f"开始研究任务：{topic}")
    print("=" * 60)
    
    # 异步执行状态机（流式获取每个节点的执行结果）
    async for state in graph_compiled.astream(
        initial_state,
        config={"configurable": {"thread_id": thread_id}}  # 线程ID，用于检查点隔离
    ):
        # 打印当前执行的节点和状态摘要
        current_node = state["__step__"]["node"]
        print(f"【当前节点】：{current_node}")
        if current_node == "generate_queries":
            print(f"【关键词】：{state.search_queries}")
        elif current_node == "execute_search":
            print(f"【搜索结果数】：{len(state.search_results)}")
        elif current_node == "summarize":
            print(f"【总结摘要】：{state.summary[:50]}...")
        elif current_node == "reflect":
            print(f"【反思摘要】：{state.reflection[:50]}...")
        elif current_node == "generate_report":
            print(f"【最终报告】：\n{state.report}")
        print("-" * 60)
    
    # 返回最终状态（包含报告）
    final_state = await graph_compiled.get_state(config={"configurable": {"thread_id": thread_id}})
    return final_state.report

# -------------------------- 7. 运行测试 --------------------------
if __name__ == "__main__":
    # 运行研究代理，研究主题：LangGraph 1.0 多代理编排实践
    report = asyncio.run(run_research_agent(topic="LangGraph 1.0 多代理编排实践"))
    # 保存报告到文件
    with open("research_report.txt", "w", encoding="utf-8") as f:
        f.write(report)
    print("报告已保存到 research_report.txt")

4.3 代码说明与测试要点

  • 异步优化：所有核心节点均为异步函数，搜索工具采用异步调用，支持多关键词并行搜索，大幅提升执行效率。
  • 状态管理：通过Pydantic v2模型定义状态，确保类型安全，所有中间结果均存储在状态中，便于调试和回溯。
  • 循环逻辑：通过条件边实现“反思→重新搜索”的循环，直到总结满足要求，实现自我修正。
  • 持久化：使用FileSystemCheckpointer将状态持久化到本地文件，支持断点续跑（如程序中断后，重新运行可从上次中断的节点继续）。
  • 测试方法：替换async_search_tool中的模拟逻辑为真实搜索API（如SerpAPI），即可获取实时数据；修改topic参数，可测试不同研究主题。

第五章：高级特性（LangGraph 1.0+）

LangGraph 1.0+ 提供了多种高级特性，用于解决复杂场景下的多代理编排问题，本节重点讲解最常用的4个高级特性：人类介入、时间旅行、子图、并行执行。

5.1 人类介入（Human-in-the-loop）

在多代理任务中，有时需要人类手动干预（如确认决策、补充信息），LangGraph 1.0+ 提供HumanMessage和HumanNode支持人类介入，暂停状态机执行，等待人类输入后继续。

from langchain_core.messages import HumanMessage
from langgraph.graph import END

# 定义人类介入节点
async def human_in_the_loop(state: ResearchState) -> ResearchState:
    """人类介入节点：暂停状态机，等待人类输入反馈"""
    print("\n【人类介入】请查看当前总结，输入反馈（如'继续'、'需要补充搜索XX内容'）：")
    # 打印当前总结
    print(f"当前总结：{state.summary}")
    # 获取人类输入
    human_feedback = input("请输入反馈：")
    # 将人类反馈添加到状态中
    state.reflection = f"人类反馈：{human_feedback}\n" + (state.reflection or "")
    return state

# 将人类介入节点添加到状态图中
graph.add_node("human_in_loop", human_in_the_loop)

# 修改流转逻辑：总结 → 人类介入 → 反思
graph.add_edge("summarize", "human_in_loop")
graph.add_edge("human_in_loop", "reflect")

# 重新编译状态机
graph_compiled = graph.compile(checkpointer=checkpointer)

核心作用：在关键节点（如总结后）暂停状态机，让人类确认内容准确性，避免代理生成错误信息，适用于对准确性要求高的场景（如学术研究、商业决策）。

5.2 时间旅行（Time Travel）

时间旅行是Checkpointer的高级特性，支持回溯到状态机的任意历史状态，重新执行后续节点，用于调试、优化或修正错误。LangGraph 1.0+ 通过get_state_history()获取历史状态，通过astream()重新执行。

async def time_travel_demo(thread_id: str):
    """时间旅行示例：回溯到历史状态，重新执行"""
    # 1. 获取该线程的所有历史状态（按执行顺序排列）
    history = await graph_compiled.get_state_history(config={"configurable": {"thread_id": thread_id}})
    
    # 2. 打印所有历史状态的节点和时间
    print("历史状态列表：")
    for i, state in enumerate(history):
        node = state["__step__"]["node"]
        timestamp = state["__step__"]["timestamp"]
        print(f"{i+1}. 节点：{node}，时间：{timestamp}")
    
    # 3. 选择一个历史状态（如第2个状态，即执行搜索后的状态）
    if len(history) >= 2:
        historical_state = history[1]  # 索引从0开始，1表示第2个状态
        print(f"\n回溯到状态：{historical_state['__step__']['node']}")
        
        # 4. 从该历史状态重新执行后续节点
        async for state in graph_compiled.astream(
            historical_state,
            config={"configurable": {"thread_id": thread_id + "_time_travel"}}  # 新线程ID，避免覆盖原状态
        ):
            print(f"时间旅行-当前节点：{state['__step__']['node']}")

# 运行时间旅行示例
# asyncio.run(time_travel_demo(thread_id="research_task_1"))

5.3 子图（Subgraphs）

当状态机的节点过多、逻辑复杂时，可将部分节点拆分为“子图”（Subgraph），实现模块化管理，提升代码可读性和可维护性。LangGraph 1.0+ 支持将子图作为一个独立的节点，嵌入到主图中。

from langgraph import StateGraph

# 1. 定义子图（搜索+总结子图）
subgraph = StateGraph(ResearchState)
# 子图节点：生成关键词 → 执行搜索 → 总结
subgraph.add_node("sub_generate_queries", generate_search_queries)
subgraph.add_node("sub_execute_search", execute_search)
subgraph.add_node("sub_summarize", summarize_results)
# 子图流转逻辑
subgraph.add_edge("sub_generate_queries", "sub_execute_search")
subgraph.add_edge("sub_execute_search", "sub_summarize")
# 子图起始和终止节点
subgraph.set_entry_point("sub_generate_queries")
subgraph.set_finish_point("sub_summarize")
# 编译子图
subgraph_compiled = subgraph.compile()

# 2. 将子图作为节点，嵌入主图
main_graph = StateGraph(ResearchState)
# 主图节点：子图 → 反思 → 生成报告
main_graph.add_node("search_summarize_subgraph", subgraph_compiled)  # 子图作为节点
main_graph.add_node("reflect", reflect_on_summary)
main_graph.add_node("generate_report", generate_report)
# 主图流转逻辑
main_graph.add_edge("search_summarize_subgraph", "reflect")
main_graph.add_edge("reflect", "generate_report")
# 主图起始和终止节点
main_graph.set_entry_point("search_summarize_subgraph")
main_graph.set_finish_point("generate_report")
# 编译主图
main_graph_compiled = main_graph.compile(checkpointer=checkpointer)

核心优势：将复杂逻辑拆分为子图，便于单独调试、复用，适合大型多代理系统（如包含10+节点的状态机）。

5.4 并行执行（Parallel Execution）

LangGraph 1.0+ 支持并行执行多个节点（如同时调用多个工具、同时生成多个总结），通过parallel()方法实现，大幅提升任务执行效率。

from langgraph.graph import parallel

# 定义两个并行节点（示例：同时生成两个不同角度的总结）
async def summarize_technical(state: ResearchState) -> ResearchState:
    """技术角度总结节点"""
    prompt = PromptTemplate(
        input_variables=["topic", "search_results"],
        template="请从技术实现角度，总结'{topic}'的核心用法，重点突出代码示例和API细节：\n{search_results}",
    )
    response = await (prompt | chat_model).ainvoke({"topic": state.topic, "search_results": "\n".join(state.search_results)})
    state.summary = f"【技术角度总结】\n{response.content}\n" + (state.summary or "")
    return state

async def summarize_practical(state: ResearchState) -> ResearchState:
    """实战角度总结节点"""
    prompt = PromptTemplate(
        input_variables=["topic", "search_results"],
        template="请从实战应用角度，总结'{topic}'的使用场景和最佳实践：\n{search_results}",
    )
    response = await (prompt | chat_model).ainvoke({"topic": state.topic, "search_results": "\n".join(state.search_results)})
    state.summary += f"【实战角度总结】\n{response.content}"
    return state

# 构建状态图，添加并行节点
graph = StateGraph(ResearchState)
graph.add_node("execute_search", execute_search)
# 并行执行两个总结节点
graph.add_node("parallel_summarize", parallel([summarize_technical, summarize_practical]))
graph.add_node("generate_report", generate_report)

# 流转逻辑：执行搜索 → 并行总结 → 生成报告
graph.add_edge("execute_search", "parallel_summarize")
graph.add_edge("parallel_summarize", "generate_report")

# 编译状态机
graph_compiled = graph.compile(checkpointer=checkpointer)

关键说明：并行节点接收当前状态，同时执行多个节点函数，执行完成后将所有节点返回的状态合并（若多个节点修改同一字段，以最后执行完成的节点结果为准）。并行执行仅适用于“无依赖关系”的节点，如多个独立的总结、多个不相关的工具调用，可最大程度利用资源，缩短任务耗时。

第六章：常见问题与排错指南（LangChain/LangGraph 1.0+）

本节汇总开发者在学习和使用LangChain 1.0+、LangGraph 1.0+ 过程中最常见的问题、报错原因及解决方案，覆盖环境配置、API调用、状态机执行、类型校验等核心场景，帮助开发者快速排错，提升开发效率。

6.1 环境配置类问题

问题1：安装后导入模块报错（如“No module named 'langchain_core'”）

报错原因：1. 未安装对应模块或版本不兼容；2. 虚拟环境未激活，模块安装到了全局环境；3. 安装过程中出现中断，模块未安装完整。

解决方案：

  1. 确认已激活虚拟环境（如conda activate langchain-env），避免全局环境冲突。
  2. 卸载已安装的相关包，重新执行安装命令（指定稳定版本，参考2.1节）：
    `pip uninstall langchain-core langchain-openai langgraph pydantic

pip install langchain-core==1.0.10 langchain-openai==0.1.0 langgraph==1.0.5 pydantic==2.6.1`

  1. 若仍报错，检查Python版本（要求Python 3.8+，推荐3.10+），过低版本不支持LangChain 1.0+ 的部分特性。

问题2：安装LangGraph后，调用graph.compile()报错“Checkpointer must be a Checkpointer instance”

报错原因：LangGraph 1.0+ 对Checkpointer的类型要求严格，传入的Checkpointer不是langgraph.checkpoint模块下的实例，或版本不兼容。

解决方案：

  1. 确认导入的Checkpointer正确，需从langgraph.checkpoint相关模块导入（如MemorySaver、FileSystemCheckpointer），而非自定义类。
  2. 检查LangGraph版本，确保为1.0+，旧版本的Checkpointer不兼容1.0+ 的compile()方法，执行pip install --upgrade langgraph==1.0.5更新。
  3. 若无需持久化，可省略Checkpointer参数（默认不使用），即graph.compile()，但生产环境不推荐。

6.2 API调用类问题

问题1：调用ChatOpenAI.invoke()报错“API key not provided”

报错原因：未配置OpenAI API密钥，或密钥配置错误、过期；未正确加载环境变量。

解决方案：

  1. 获取有效的OpenAI API密钥（登录OpenAI官网创建），确保密钥未过期、未被封禁。
  2. 两种配置方式（推荐第二种，更安全）：
    直接传入api_key参数（测试用）：chat_model = ChatOpenAI(model="gpt-4o", api_key="your-api-key")
  3. 通过环境变量加载（生产用）：
    `import os

from langchain_chat_models import ChatOpenAI

chat_model = ChatOpenAI(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"))`
（配置环境变量：Windows用set OPENAI_API_KEY=your-api-key，Linux/Mac用export OPENAI_API_KEY=your-api-key）

问题2：异步调用ainvoke()报错“cannot be called from a running event loop”

报错原因：在异步环境中重复调用异步函数，或未通过asyncio.run()启动异步事件循环；Jupyter Notebook等环境自带事件循环，易出现此问题。

解决方案：

  1. 普通Python脚本：通过asyncio.run()包裹异步函数，如asyncio.run(run_research_agent(topic="xxx"))。
  2. Jupyter Notebook环境：在异步函数前添加%autoawait魔法命令，或使用async with语句，避免重复启动事件循环：
    `%autoawait

await run_research_agent(topic="LangGraph 1.0 多代理编排实践")`

  1. 避免在异步函数内部直接调用asyncio.run()，需通过await调用其他异步函数。

6.3 状态机与节点类问题

问题1：状态机执行报错“State must be a subclass of BaseModel”

报错原因：LangGraph 1.0+ 要求State必须是Pydantic v2 BaseModel的子类，使用了旧版的TypedDict或普通类定义状态。

解决方案：将状态类修改为继承自pydantic.BaseModel，参考3.2节的ResearchState定义，确保所有字段都有明确的类型注解和Field描述。

问题2：节点函数返回报错“Node must return an instance of the state type”

报错原因：节点函数未返回状态模型实例，或返回的实例类型与StateGraph指定的状态类型不一致；节点函数中修改了状态但未返回。

解决方案：

  1. 确保所有节点函数（同步/异步）的返回值是StateGraph初始化时指定的状态模型实例（如ResearchState）。
  2. 节点函数中修改状态后，必须return state，避免忘记返回（如def node(state: ResearchState) -> ResearchState: state.is_complete = True; return state）。
  3. 检查节点函数的返回类型注解，确保与状态类型一致，避免返回None或其他类型。

问题3：条件边路由函数报错“Path function must return a destination node name”

报错原因：条件路由函数（add_conditional_edges的path参数）返回值不是字符串，或返回的节点名称不在dests列表中；返回值为空。

解决方案：

  1. 确保路由函数的返回值是字符串，且该字符串是add_conditional_edges中dests列表中的节点名称（如return "generate_queries"）。
  2. 避免路由函数返回None、数字等非字符串类型，可添加默认返回值（如else: return "end"）。
  3. 检查dests列表中的节点名称是否与状态图中添加的节点名称一致（大小写敏感）。

6.4 类型校验类问题

问题1：Pydantic模型初始化报错“Field required [type=missing, input_value={}, input_type=dict]”

报错原因：初始化Pydantic状态模型时，未传入必填字段（Field未指定default或default_factory）；传入的字段类型与注解类型不一致。

解决方案：

  1. 检查状态模型的字段定义，必填字段（无默认值）在初始化时必须传入，如ResearchState(topic="xxx")（topic是必填字段）。
  2. 确保传入的字段类型与注解一致，如search_queries是List[str]类型，不可传入字符串或数字。
  3. 若字段可选，可通过default=None或default_factory=list（列表类型）设置默认值，避免必填报错。

问题2：LLM输出解析报错“Could not parse output into Pydantic model”

报错原因：LLM返回的内容不符合Pydantic输出解析器的格式要求；Prompt中未注入解析器的格式说明；LLM输出存在冗余内容（如多余的解释、换行）。

解决方案：

  1. 在Prompt中必须注入解析器的格式说明，即partial_variables={"format_instructions": parser.get_format_instructions()}，参考2.4节。
  2. 优化Prompt，明确要求LLM严格按照格式说明输出，避免多余内容（如“仅输出符合格式的JSON，不要添加任何解释”）。
  3. 若LLM输出存在轻微格式错误，可使用PydanticOutputParser的partial_parse=True参数，允许部分解析（需LangChain 1.0.8+ 版本）。

第七章：学习资源与进阶方向

掌握LangChain 1.0+ 和 LangGraph 1.0+ 的基础用法后，可通过以下资源进一步提升，聚焦实战落地和高级场景，成为多代理系统开发专家。

7.1 官方资源（最权威，优先参考）

  • LangChain 1.0 官方文档：https://python.langchain.com/docs/get_started/introduction，包含核心概念、API详解、实战示例，实时更新。
  • LangGraph 1.0 官方文档：https://langchain.com/docs/langgraph，重点关注StateGraph、Checkpointer、子图等核心特性的详细说明。
  • LangChain GitHub 仓库：https://github.com/langchain-ai/langchain，查看源码、提交记录，了解底层实现逻辑。
  • LangGraph GitHub 仓库：https://github.com/langchain-ai/langgraph，包含官方示例、测试用例，可直接复用代码。

7.2 实战进阶资源

  • LangChain 1.0 实战项目：官方示例仓库中的“agent”“graphs”目录，包含多代理、状态机的完整实战代码，可直接运行和修改。
  • SerpAPI 文档：https://serpapi.com/docs，学习如何将真实搜索接口集成到工具中，替代模拟搜索逻辑。
  • Pydantic v2 官方文档：https://docs.pydantic.dev/latest/，深入学习类型校验、模型序列化、嵌套模型等高级用法，提升状态管理能力。

7.3 进阶学习方向

  1. 多代理系统优化：学习节点调度策略、并行执行优化、资源限制（如LLM调用频率限制），提升系统稳定性和效率。
  2. 复杂状态管理：结合Redis、PostgreSQL等数据库，实现分布式Checkpointer，支持多实例部署和大规模任务调度。
  3. LangGraph 与其他框架集成：如与FastAPI集成，构建多代理API服务；与LangSmith集成，实现任务监控、日志分析和调试。
  4. 本地化部署：将LLM（如Llama 3、Qwen）本地化部署，结合LangChain/LangGraph，构建离线可用的多代理系统。

总结

本指南从LangChain 1.0+ 和 LangGraph 1.0+ 的核心变更出发，逐步讲解了LCEL核心范式、状态机机制、实战项目、高级特性及排错指南，覆盖了从入门到进阶的全流程知识点。LangChain 1.0+ 的核心是“类型安全、异步高效、简洁可扩展”，而LangGraph 1.0+ 则通过StateGraph实现了复杂多代理逻辑的灵活编排，两者结合可快速构建生产级的AI代理系统。

学习过程中，建议重点掌握LCEL组件串联、Pydantic状态定义、节点与边的配置，多动手运行示例代码，结合实战项目巩固知识点。遇到问题时，优先参考官方文档和本指南的排错部分，逐步提升问题解决能力。随着AI技术的发展，LangChain/LangGraph生态会持续更新，建议保持关注官方动态，不断学习新特性，将技术落地到实际项目中。

（注：文档部分内容可能由 AI 生成）

添加新评论