0.3 补充：LangChain 快速介绍 - 从 LangGraph 视角理解

1. LangChain 是 LangGraph 的基础

如果你正在学习 LangGraph，你可能会发现很多代码示例中都频繁出现 LangChain 的各种组件。关于这些组键，我们后面也会进行通俗解释，反复学习。

python

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_community.tools.tavily_search import TavilySearchResults

这是为什么呢？简单来说：

LangChain 是工具箱，LangGraph 是施工图
LangChain 提供了与 LLM 交互的各种"零件"（Chat Models、Tools、Prompts 等）
LangGraph 提供了编排这些零件的"蓝图"（StateGraph、节点、条件边等）

类比理解

盖房子需要什么？
- 砖头、水泥、钢筋、门窗 → 这是 LangChain 提供的组件
- 建筑图纸、施工流程、质量检查 → 这是 LangGraph 提供的编排能力

本章目标：通过本章，你将深入理解 LangChain 的 7 大核心组件，并明白它们如何在 LangGraph 中发挥作用。

📊 LangChain 与 LangGraph 的技术架构


                    构建 AI Agent

                          ↓

                   LangGraph 层

  StateGraph（状态图）   Agent 编排
  节点     条件边／循环      对话流


                          ↓ 调用

                   LangChain 层

  Chat    Tools   Prompt  Parser  Memory
  Models  （工具）  （提示）  （解析）  （记忆）


                          ↓ 调用

                    底层 API
          OpenAI API / Anthropic API / 本地模型

📚 LangChain 7 大核心组件概览

2.1 Chat Models（聊天模型）- LLM 调用的基础

Chat Models 是 LangChain 最核心的组件，它封装了与各种 LLM（如 GPT-4、Claude）的交互接口。

基础调用

python

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")

from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,  # 控制随机性：0-2，越高越随机
    max_tokens=100    # 限制输出长度
)

# 调用模型
response = llm.invoke("请用一句话解释 LangChain")
print(response.content)

运行结果：

LangChain 是一个强化大语言模型应用的框架，简化了 LLM 集成和工作流构建。

🔍 Python 知识点：函数参数与默认值

python

# temperature=0.7 是"关键字参数"（keyword argument）
# 如果不指定，会使用默认值
llm = ChatOpenAI(model="gpt-4o-mini")  # temperature 使用默认值 0.7

消息对象

LangChain 使用结构化的消息对象，而不是简单字符串：

python

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="你是一位资深 Python 专家"),
    HumanMessage(content="如何创建虚拟环境？"),
]

response = llm.invoke(messages)
print(response.content)

运行结果：

创建 Python 虚拟环境的步骤如下：
1. 运行命令 `python -m venv myenv`
2. 激活环境：
   - Windows: `myenv\Scripts\activate`
   - macOS/Linux: `source myenv/bin/activate`
3. 使用 `pip install -r requirements.txt` 安装依赖

🔍 Python 知识点：类与对象

python

# SystemMessage 是一个"类"（Class），content 是它的"属性"（Attribute）
msg = SystemMessage(content="你好")

# 可以通过 msg.content 访问内容
print(msg.content)  # 输出：你好
print(msg.type)     # 输出：system

流式输出

当处理长文本生成时，流式输出能提供更好的用户体验：

python

from langchain_openai import ChatOpenAI

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

print("流式输出：", end="")
for chunk in llm.stream("请写一首关于 AI 的小诗（简短）"):
    print(chunk.content, end="", flush=True)
print()

运行结果：

流式输出：智能涌动如潮
算法编织奇妙
数据海洋探索
未来由此开创

🔍 Python 知识点：生成器（Generator）

python

# llm.stream() 返回的是一个"生成器"，可以用 for 循环逐个获取结果
# 这比一次性返回所有结果更节省内存

def my_generator():
    yield 1
    yield 2
    yield 3

for num in my_generator():
    print(num)  # 输出 1, 2, 3（逐个输出，不是一次性）

多模型支持

LangChain 支持 30+ 种模型提供商，切换模型非常简单：

python

# OpenAI
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")

# Anthropic Claude
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

# Google Gemini
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-pro")

# 本地模型（Ollama）
from langchain_community.chat_models import ChatOllama
llm = ChatOllama(model="llama2")

⚡ 在 LangGraph 中的应用（重要）

python

from langgraph.graph import StateGraph, MessagesState

def chatbot_node(state: MessagesState):
    # 直接使用 LangChain 的 Chat Model
    llm = ChatOpenAI(model="gpt-4o-mini")
    return {"messages": [llm.invoke(state["messages"])]}

关键点：

LangGraph 的节点函数中，通常使用 LangChain 的 Chat Model 来调用 LLM
MessagesState 是 LangGraph 提供的便捷状态类型，内部使用 LangChain 的消息格式

2.2 Prompts（提示词模板）- 结构化提示词管理

提示词模板帮助你优雅地组织复杂的提示词，避免手动拼接字符串。

基础模板

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 创建模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位资深 {role} 专家"),
    ("human", "{question}")
])

# 组合成链
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm

response = chain.invoke({
    "role": "算法工程师",
    "question": "什么是二叉树？"
})
print(response.content)

运行结果：

二叉树是一种基础数据结构，每个节点最多有两个子节点：左子节点和右子节点。
常见应用包括：
1. 二叉搜索树（BST）：用于高效查找
2. 堆：用于优先队列实现
3. 表达式树：用于编译器构建抽象语法树
4. 红黑树：自平衡二叉搜索树，保证 O(log n) 性能

🔍 Python 知识点：字典与格式化

python

# {role} 和 {question} 是"占位符"，会被字典中的值替换
template = "你好，{name}！"
result = template.format(name="张三")  # 输出：你好，张三！

# 也可以用 f-string（Python 3.6+）
name = "李四"
result = f"你好，{name}！"  # 输出：你好，李四！

🔍 Python 知识点：管道操作符（|）

python

# prompt | llm 这种写法叫做 LCEL（LangChain Expression Language）
# | 是 Python 的"按位或"运算符，LangChain 重载了它，用于连接组件

# 等价于：
chain = prompt | llm
# 实际执行时：先运行 prompt，再将结果传给 llm

result = chain.invoke({"role": "专家", "question": "问题"})

Few-Shot 提示模板

Few-Shot Learning 通过提供示例，让 LLM 更好地理解任务：

python

from langchain_core.prompts import FewShotChatMessagePromptTemplate

# 定义示例
examples = [
    {"input": "开心", "output": "😊"},
    {"input": "难过", "output": "😢"},
    {"input": "愤怒", "output": "😠"},
]

# 创建 Few-Shot 模板
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

# 完整提示词
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你需要将情绪词转换为对应的 emoji"),
    few_shot_prompt,
    ("human", "{input}"),
])

chain = final_prompt | llm
response = chain.invoke({"input": "兴奋"})
print(response.content)

运行结果：

🤩

模板变量复用

python

from langchain_core.prompts import ChatPromptTemplate

# 支持多个变量
template = ChatPromptTemplate.from_messages([
    ("system", "你是 {company} 的客服代表，角色是 {role}"),
    ("human", "请用 {tone} 的语气回答：{question}")
])

chain = template | llm
response = chain.invoke({
    "company": "OpenAI",
    "role": "技术支持",
    "tone": "专业友好",
    "question": "API 调用失败怎么办？"
})
print(response.content)

运行结果：

非常抱歉您遇到了 API 调用失败的问题，我很乐意帮助您解决。

1. **检查网络连接**：请确认您的网络连接是否稳定，防火墙是否阻止了请求
2. **查看错误详情**：请提供具体的错误代码（如 timeout 或状态码）
3. **检查请求参数**：确认所有必需参数都已正确传递，数据格式符合要求
4. **查看服务状态**：访问 status.openai.com 查看 API 是否正常运行

如果问题依然存在，欢迎提供更详细的错误信息，我会进一步协助您。

⚡ 在 LangGraph 中的应用（重要）

python

from langgraph.graph import StateGraph, MessagesState

def agent_node(state: MessagesState):
    # 使用提示词模板为 Agent 设计系统提示词
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个能够使用工具的 ReAct Agent，请遵循 Thought-Action-Observation 流程"),
        ("placeholder", "{messages}")  # 占位符用于插入历史消息
    ])

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

    response = chain.invoke({"messages": state["messages"]})
    return {"messages": [response]}

关键点：

提示词模板在 LangGraph 节点中用于设计 Agent 的"性格"和"能力"
placeholder 类型的消息用于动态插入历史对话

2.3 Output Parsers（输出解析器）- 结构化 LLM 输出

LLM 的原始输出是文本，Output Parsers 帮助我们将文本转换为结构化数据（字符串、JSON、Python 对象）。

StrOutputParser（字符串解析）

python

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("human", "请写一首关于 {topic} 的诗 ")
])

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

# 使用 LCEL 串联组件
chain = prompt | llm | output_parser

result = chain.invoke({"topic": "秋天"})
print(type(result))  # <class 'str'>
print(result)

运行结果：

<class 'str'>
金风送爽叶飘黄
稻谷飘香满田庄
落日余晖染山岗
秋意浓浓入梦乡

JsonOutputParser（JSON 解析）

python

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate

# 定义输出结构
parser = JsonOutputParser()

prompt = ChatPromptTemplate.from_messages([
    ("system", "请以 JSON 格式输出，必须包含 'title' 和 'summary' 字段"),
    ("human", "总结这段文章：{article}")
])

llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | parser

result = chain.invoke({
    "article": "LangChain 是一个强化大语言模型应用的框架，提供了模型调用、提示词工程、工具集成等核心功能"
})

print(type(result))  # <class 'dict'>
print(result)

运行结果：

<class 'dict'>
{
    'title': 'LangChain 框架简介',
    'summary': 'LangChain 是专为增强大语言模型能力设计的框架，集成了模型调用、提示词管理、工具接口等关键组件'
}

🔍 Python 知识点：字典（Dictionary）

python

# 字典是 Python 的核心数据结构，使用 {} 或 dict() 创建
person = {
    "name": "张三",
    "age": 30,
    "city": "北京"
}

# 访问字典元素
print(person["name"])  # 输出：张三
print(person.get("age"))  # 输出：30（推荐用法，不会抛异常）

# 添加或修改元素
person["email"] = "zhangsan@example.com"

PydanticOutputParser（结构化解析）

Pydantic 是 Python 的数据验证库，能确保 LLM 输出符合预期格式：

python

from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

# 定义数据模型
class Person(BaseModel):
    name: str = Field(description="人物姓名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")
    skills: list[str] = Field(description="技能列表")

parser = PydanticOutputParser(pydantic_object=Person)

prompt = ChatPromptTemplate.from_messages([
    ("system", "从文本中提取人物信息并输出\n{format_instructions}"),
    ("human", "{text}")
])

llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | parser

result = chain.invoke({
    "text": "小明是一位 35 岁的资深算法工程师，精通 Python、机器学习、深度学习",
    "format_instructions": parser.get_format_instructions()
})

print(type(result))  # <class '__main__.Person'>
print(f"姓名：{result.name}")
print(f"年龄：{result.age}")
print(f"职业：{result.occupation}")
print(f"技能：{', '.join(result.skills)}")

运行结果：

<class '__main__.Person'>
姓名：小明
年龄：35
职业：算法工程师
技能：Python, 机器学习, 深度学习

🔍 Python 知识点：类型注解与 Pydantic

python

from pydantic import BaseModel

# 类型注解（Type Annotation）告诉 Python 变量的类型
name: str = "张三"  # name 应该是字符串
age: int = 30       # age 应该是整数

# Pydantic 的 BaseModel 会在运行时验证类型
class User(BaseModel):
    name: str
    age: int

# 正确：类型匹配
user = User(name="李四", age=25)

# 错误：类型不匹配，会抛出异常
# user = User(name="王五", age="三十")  # ValidationError

⚡ 在 LangGraph 中的应用（重要）

python

from langgraph.graph import StateGraph
from typing import TypedDict
from pydantic import BaseModel

class ExtractedInfo(BaseModel):
    intent: str
    entities: list[str]

class State(TypedDict):
    user_input: str
    parsed_data: ExtractedInfo

def parse_node(state: State):
    parser = PydanticOutputParser(pydantic_object=ExtractedInfo)
    prompt = ChatPromptTemplate.from_messages([
        ("system", "从用户输入中提取意图和实体\n{format_instructions}"),
        ("human", "{input}")
    ])

    llm = ChatOpenAI(model="gpt-4o-mini")
    chain = prompt | llm | parser

    parsed = chain.invoke({
        "input": state["user_input"],
        "format_instructions": parser.get_format_instructions()
    })

    return {"parsed_data": parsed}

关键点：

Output Parsers 在 LangGraph 中用于将 LLM 的文本输出转换为结构化数据
结构化数据可以存储在 State 中，供后续节点使用

2.4 Tools（工具）- 扩展 LLM 能力

Tools 让 LLM 能够"调用外部函数"，从而执行搜索、计算、API 调用等操作。

使用 @tool 装饰器定义工具

python

from langchain_core.tools import tool

@tool
def calculate(expression: str) -> str:
    """执行数学计算。

    Args:
        expression: 数学表达式（如 '2+3*4'）
    """
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

@tool
def get_weather(city: str) -> str:
    """查询城市天气。

    Args:
        city: 城市名称（如 '北京'）
    """
    # 模拟天气查询
    weather_db = {
        "北京": "晴天，15°C，空气质量优",
        "上海": "阴天，18°C，湿度 70%",
        "深圳": "雨天，22°C，有雷暴"
    }
    return weather_db.get(city, f"抱歉，没有找到 {city} 的天气信息")

# 查看工具信息
print(f"工具名：{calculate.name}")
print(f"工具描述：{calculate.description}")
print(f"参数结构：{calculate.args}")

运行结果：

工具名：calculate
工具描述：执行数学计算。

    Args:
        expression: 数学表达式（如 '2+3*4'）

参数结构：{'expression': {'title': 'Expression', 'type': 'string'}}

🔍 Python 知识点：装饰器（Decorator）

python

# 装饰器是 Python 的"语法糖"，用于修改函数行为
# @tool 会将普通函数转换为 LangChain 工具

# 没有装饰器的版本
def my_function():
    return "Hello"

# 使用装饰器
@tool
def my_function():
    """这是工具描述"""
    return "Hello"

# 等价于：
def my_function():
    """这是工具描述"""
    return "Hello"
my_function = tool(my_function)

工具绑定到 LLM

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

tools = [calculate, get_weather]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)

# LLM 会自动决定是否调用工具
response = llm_with_tools.invoke([
    HumanMessage(content="北京今天天气怎么样？另外，帮我算一下  25 * 4 + 10")
])

print("LLM 的决策：")
if response.tool_calls:
    for tool_call in response.tool_calls:
        print(f"  - 调用工具：{tool_call['name']}")
        print(f"    参数：{tool_call['args']}")
else:
    print("  - 直接文本回答（不调用工具）")

运行结果：

LLM 的决策：
  - 调用工具：get_weather
    参数：{'city': '北京'}
  - 调用工具：calculate
    参数：{'expression': '25 * 4 + 10'}

执行工具调用

python

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

# 1. 用户提问
messages = [HumanMessage(content="深圳天气如何？计算 100 / 5")]

# 2. LLM 决策调用工具
llm_with_tools = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
ai_message = llm_with_tools.invoke(messages)
messages.append(ai_message)

# 3. 执行工具调用
for tool_call in ai_message.tool_calls:
    # 找到对应工具
    tool_map = {tool.name: tool for tool in tools}
    selected_tool = tool_map[tool_call["name"]]

    # 执行工具
    tool_output = selected_tool.invoke(tool_call["args"])

    # 将工具结果添加到消息历史
    messages.append(ToolMessage(
        content=tool_output,
        tool_call_id=tool_call["id"]
    ))

    print(f"工具 {tool_call['name']} 输出：{tool_output}")

# 4. LLM 根据工具结果生成最终回答
final_response = llm_with_tools.invoke(messages)
print(f"\n最终回答：{final_response.content}")

运行结果：

工具 get_weather 输出：雨天，22°C，有雷暴
工具 calculate 输出：计算结果：20.0

最终回答：深圳今天的天气是雨天，温度 22°C，有雷暴。计算结果：100 除以 5 等于 20。

使用 StructuredTool 定义复杂工具

python

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

# 定义输入参数模型
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, description="最多返回结果数")
    language: str = Field(default="zh", description="搜索语言代码")

def search_web(query: str, max_results: int = 5, language: str = "zh") -> str:
    """模拟网络搜索"""
    return f"搜索 '{query}' 找到 {max_results} 条结果（语言：{language}）"

# 创建结构化工具
search_tool = StructuredTool.from_function(
    func=search_web,
    name="web_search",
    description="在互联网上搜索信息",
    args_schema=SearchInput
)

# 测试工具
result = search_tool.invoke({
    "query": "LangGraph 教程",
    "max_results": 10,
    "language": "zh"
})
print(result)

运行结果：

搜索 'LangGraph 教程' 找到 10 条结果（语言：zh）

⚡ 在 LangGraph 中的应用（重要）

python

from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END

def agent(state: MessagesState):
    llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
    return {"messages": [llm.invoke(state["messages"])]}

def should_continue(state: MessagesState):
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# 构建图
graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))  # LangGraph 的 ToolNode 会自动执行工具

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")

app = graph.compile()

关键点：

ToolNode 是 LangGraph 提供的便捷节点，自动处理工具调用
Agent 节点负责决策，ToolNode 负责执行
通过条件边实现"决策 → 工具调用 → 决策"的循环

2.5 Messages（消息对象）- 对话历史管理

消息对象是 LangChain 最基础的数据结构，用于表示 LLM 的对话历史。

消息类型概览

python

from langchain_core.messages import (
    SystemMessage,
    HumanMessage,
    AIMessage,
    ToolMessage,
    FunctionMessage
)

# 1. SystemMessage（系统消息）：设定角色和提示词
system_msg = SystemMessage(content="你是一位资深 Python 专家")
print(f"System: {system_msg.content}")

# 2. HumanMessage（用户消息）：用户的输入
human_msg = HumanMessage(content="如何读取文件？")
print(f"Human: {human_msg.content}")

# 3. AIMessage（AI 消息）：AI 的回复
ai_msg = AIMessage(content="使用 open() 函数可以读取文件")
print(f"AI: {ai_msg.content}")

# 4. ToolMessage（工具消息）：工具的返回结果
tool_msg = ToolMessage(
    content="文件内容：Hello World",
    tool_call_id="call_123"
)
print(f"Tool: {tool_msg.content}")

运行结果：

System: 你是一位资深 Python 专家
Human: 如何读取文件？
AI: 使用 open() 函数可以读取文件
Tool: 文件内容：Hello World

消息的特殊属性

python

from langchain_core.messages import AIMessage

# 创建包含工具调用的消息
ai_message = AIMessage(
    content="我需要查询天气",
    tool_calls=[{
        "name": "get_weather",
        "args": {"city": "北京"},
        "id": "call_abc123"
    }]
)

print(f"内容：{ai_message.content}")
print(f"工具调用：{ai_message.tool_calls}")
print(f"类型：{ai_message.type}")

运行结果：

内容：我需要查询天气
工具调用：[{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_abc123'}]
类型：ai

消息历史管理

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

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

# 模拟多轮对话
conversation = [
    SystemMessage(content="你是一位幽默的数学老师"),
    HumanMessage(content="你好"),
    AIMessage(content="你好！今天想学点什么数学知识呢？"),
    HumanMessage(content="我想学微积分")
]

response = llm.invoke(conversation)
print(response.content)

运行结果：

太好了！微积分是数学的精华。让我们从最基础的极限概念开始吧。

🔍 Python 知识点：列表（List）

python

# 列表是 Python 中最常用的数据结构，用 [] 创建
messages = [
    HumanMessage(content="你好"),
    AIMessage(content="你好！")
]

# 添加元素
messages.append(HumanMessage(content="再见"))

# 访问元素
print(messages[0])   # 第一个元素
print(messages[-1])  # 最后一个元素

# 切片
print(messages[:2])  # 前两个元素

消息序列化

python

from langchain_core.messages import HumanMessage, AIMessage
import json

# 消息可以转换为字典
messages = [
    HumanMessage(content="你好"),
    AIMessage(content="你好！请问有什么可以帮助的？")
]

# 转换为字典
messages_dict = [msg.dict() for msg in messages]
print(json.dumps(messages_dict, ensure_ascii=False, indent=2))

运行结果：

json

[
  {
    "content": "你好",
    "type": "human",
    "additional_kwargs": {},
    "example": false
  },
  {
    "content": "你好！请问有什么可以帮助的？",
    "type": "ai",
    "additional_kwargs": {},
    "example": false
  }
]

⚡ 在 LangGraph 中的应用（重要）

python

from langgraph.graph import MessagesState

# MessagesState 是 LangGraph 内置的状态类型，会自动管理消息历史
class CustomState(MessagesState):
    user_name: str

def chatbot(state: CustomState):
    # 访问消息历史
    messages = state["messages"]

    # 可以添加自定义系统消息
    if state.get("user_name"):
        system_msg = SystemMessage(content=f"用户名是 {state['user_name']}")
        messages = [system_msg] + messages

    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(messages)

    return {"messages": [response]}

关键点：

MessagesState 自动累积消息历史
在 LangGraph 节点中，可以随时访问和修改消息列表
消息类型（System/Human/AI/Tool）帮助 LLM 理解对话上下文

2.6 Retrievers & RAG（检索增强生成）- 让 LLM 访问外部知识

Retriever 是 RAG（Retrieval-Augmented Generation）系统的核心组件，让 LLM 能够访问外部知识库。

向量存储基础

python

import getpass
import os

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

_set_env("OPENAI_API_KEY")

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

# 准备文档
documents = [
    Document(page_content="LangChain 是一个强化大语言模型应用的框架", metadata={"source": "doc1"}),
    Document(page_content="LangGraph 用于构建有状态的 Agent 系统", metadata={"source": "doc2"}),
    Document(page_content="RAG 是一种技术，让 LLM 能够访问外部知识库", metadata={"source": "doc3"}),
    Document(page_content="向量数据库用于存储文档的向量表示", metadata={"source": "doc4"}),
]

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(documents, embeddings)

# 相似度搜索
query = "如何构建 Agent 系统"
results = vectorstore.similarity_search(query, k=2)

print(f"查询：{query}\n")
for i, doc in enumerate(results, 1):
    print(f"结果 {i}：{doc.page_content}")
    print(f"来源：{doc.metadata['source']}\n")

运行结果：

查询：如何构建 Agent 系统

结果 1：LangGraph 用于构建有状态的 Agent 系统
来源：doc2

结果 2：LangChain 是一个强化大语言模型应用的框架
来源：doc1

🔍 Python 知识点：Enumerate

python

# enumerate() 在遍历列表时同时获取索引和元素
fruits = ["苹果", "香蕉", "橙子"]

for i, fruit in enumerate(fruits, 1):  # 1 表示索引从 1 开始
    print(f"第 {i} 个水果是 {fruit}")

# 输出：
# 第 1 个水果是 苹果
# 第 2 个水果是 香蕉
# 第 3 个水果是 橙子

带分数的检索

python

# 相似度搜索（带分数）
results_with_scores = vectorstore.similarity_search_with_score(query, k=3)

print(f"查询：{query}\n")
for doc, score in results_with_scores:
    print(f"相似度：{score:.4f}")
    print(f"内容：{doc.page_content}\n")

运行结果：

查询：如何构建 Agent 系统

相似度：0.2145
内容：LangGraph 用于构建有状态的 Agent 系统

相似度：0.3267
内容：LangChain 是一个强化大语言模型应用的框架

相似度：0.4521
内容：RAG 是一种技术，让 LLM 能够访问外部知识库

完整 RAG 流程

python

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. 准备知识库
documents = [
    Document(page_content="LangGraph 支持循环和条件分支"),
    Document(page_content="使用 StateGraph 可以定义状态图"),
    Document(page_content="ToolNode 用于自动执行工具调用"),
    Document(page_content="MemorySaver 可以保存对话历史"),
]

vectorstore = FAISS.from_documents(documents, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 2. 创建 RAG 链
template = """请根据以下上下文回答问题。

上下文：
{context}

问题：{question}

回答："""

prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model="gpt-4o-mini")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# LCEL 构建链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 3. 查询
question = "如何在 LangGraph 中保存对话历史？"
answer = rag_chain.invoke(question)

print(f"问题：{question}\n")
print(f"回答：{answer}")

运行结果：

问题：如何在 LangGraph 中保存对话历史？

回答：在 LangGraph 中可以使用 MemorySaver 来保存对话历史。MemorySaver 是一个内存检查点保存器，
能够保存状态快照。使用方法是在编译图时传入 checkpointer 参数：

app = graph.compile(checkpointer=MemorySaver())

在调用时需要指定 thread_id 来区分不同会话。

⚡ 在 LangGraph 中的应用（重要）

python

from langgraph.graph import StateGraph, MessagesState
from langchain_community.vectorstores import FAISS

class RAGState(MessagesState):
    context: str

def retrieve_node(state: RAGState):
    """检索相关文档"""
    last_message = state["messages"][-1].content
    docs = retriever.invoke(last_message)
    context = "\n".join([doc.page_content for doc in docs])
    return {"context": context}

def generate_node(state: RAGState):
    """生成回答"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "请根据以下上下文回答\n{context}"),
        ("placeholder", "{messages}")
    ])

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

    response = chain.invoke({
        "context": state["context"],
        "messages": state["messages"]
    })

    return {"messages": [response]}

# 构建 RAG 图
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.add_edge("retrieve", "generate")

关键点：

RAG 在 LangGraph 中通常分为两个节点：检索节点 + 生成节点
检索到的上下文存储在 State 中
这种模式可以支持更复杂的检索策略（如重排序、过滤）

2.7 LCEL（LangChain Expression Language）- 组件编排语言

LCEL 是 LangChain 的链式组合语法，使用 | 操作符连接组件。

基础链组合

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 定义组件
prompt = ChatPromptTemplate.from_template("讲一个关于 {topic} 的笑话")
llm = ChatOpenAI(model="gpt-4o-mini")
output_parser = StrOutputParser()

# 使用 | 操作符串联
chain = prompt | llm | output_parser

# 调用链
result = chain.invoke({"topic": "程序员"})
print(result)

运行结果：

为什么程序员总喜欢深夜工作？
答案：因为 Oct 31 == Dec 25
1031（八进制） = 1225（八进制）：万圣节就是圣诞节！

RunnablePassthrough（数据传递）

python

from langchain_core.runnables import RunnablePassthrough

chain = (
    {"original": RunnablePassthrough(), "uppercase": lambda x: x.upper()}
    | (lambda x: f"原文：{x['original']}\n大写：{x['uppercase']}")
)

result = chain.invoke("hello world")
print(result)

运行结果：

原文：hello world
大写：HELLO WORLD

RunnableLambda（自定义函数）

python

from langchain_core.runnables import RunnableLambda

def add_prefix(text: str) -> str:
    return f"AI 回答：{text}"

def add_suffix(text: str) -> str:
    return f"{text}\n\n---\n生成时间：2024-01-15"

chain = (
    ChatPromptTemplate.from_template("解释什么是 {term}")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
    | RunnableLambda(add_prefix)
    | RunnableLambda(add_suffix)
)

result = chain.invoke({"term": "LCEL"})
print(result)

运行结果：

AI 回答：LCEL（LangChain Expression Language）是 LangChain 提供的链式组合语法，
通过管道操作符连接组件，创建数据处理流程。

---
生成时间：2024-01-15

并行执行

python

from langchain_core.runnables import RunnableParallel

# 并行调用多个链
chain = RunnableParallel({
    "joke": ChatPromptTemplate.from_template("讲一个关于 {topic} 的笑话") | llm | StrOutputParser(),
    "poem": ChatPromptTemplate.from_template("写一首关于 {topic} 的诗") | llm | StrOutputParser(),
    "fact": ChatPromptTemplate.from_template("说一个关于 {topic} 的事实") | llm | StrOutputParser(),
})

results = chain.invoke({"topic": "Python"})

print("=== 笑话 ===")
print(results["joke"])
print("\n=== 诗歌 ===")
print(results["poem"])
print("\n=== 事实 ===")
print(results["fact"])

运行结果：

=== 笑话 ===
为什么 Python 程序员喜欢大自然？
答案：因为它有很多 "类"（class）！

=== 诗歌 ===
代码诗行如流云
缩进优雅胜繁文

=== 事实 ===
Python 由 Guido van Rossum 于 1991 年创建，命名灵感来自英国喜剧团体 Monty Python。
它采用的是简洁明了的语法，强调代码可读性。

条件分支

python

from langchain_core.runnables import RunnableBranch

# 根据输入长度分配不同的处理逻辑
def route_by_length(x: dict) -> str:
    if len(x["text"]) < 10:
        return "short"
    elif len(x["text"]) < 50:
        return "medium"
    else:
        return "long"

branch = RunnableBranch(
    (lambda x: route_by_length(x) == "short", lambda x: f"短文本：{x['text']}"),
    (lambda x: route_by_length(x) == "medium", lambda x: f"中等长度：{x['text'][:20]}..."),
    lambda x: f"长文本：{x['text'][:30]}..."
)

print(branch.invoke({"text": "Hi"}))
print(branch.invoke({"text": "This is a medium length text"}))
print(branch.invoke({"text": "This is a very long text that exceeds the threshold for long texts"}))

运行结果：

短文本：Hi
中等长度：This is a medium leng...
长文本：This is a very long text that...

LCEL vs LangGraph 对比

维度	LCEL	LangGraph
适用场景	简单的单向流程	复杂的多分支流程
语法	管道操作符 `	`
状态管理	无	显式 State 管理
循环控制	不支持	支持
条件分支	RunnableBranch	add_conditional_edges
可视化	困难	支持图可视化
调试	有限	支持断点、时间旅行

📊 适用场景对比

何时使用 LCEL：

简单的"提示词 → LLM → 解析"流程
RAG 单次检索场景
无需状态管理的任务

何时使用 LangGraph：

需要循环执行（如 ReAct Agent）
复杂条件分支（如 Multi-Agent 路由）
需要状态持久化（对话历史）
人机协作（断点、编辑状态）

代码对比示例

python

# LCEL：简单的单向流程
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# LangGraph：复杂的循环流程
graph = StateGraph(State)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.add_conditional_edges("generate", should_retry, {
    "retry": "retrieve",
    "end": END
})

第 3 部分：LangChain 与 LangGraph 整合实战

3.1 Chat Models 整合

python

from langgraph.graph import StateGraph, MessagesState
from langchain_openai import ChatOpenAI

def chatbot_node(state: MessagesState):
    # 直接使用 LangChain 的 ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot_node)

3.2 Tools 整合

python

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def search_tool(query: str) -> str:
    """搜索工具"""
    return f"搜索结果：{query}"

tools = [search_tool]

# LangGraph 的 ToolNode 会自动执行 LangChain 工具
tool_node = ToolNode(tools)
graph.add_node("tools", tool_node)

3.3 Prompts 整合

python

from langchain_core.prompts import ChatPromptTemplate

def agent_node(state: MessagesState):
    # 使用 LangChain 提示词模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是 {role}"),
        ("placeholder", "{messages}")
    ])

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

    response = chain.invoke({
        "role": "Python 专家",
        "messages": state["messages"]
    })

    return {"messages": [response]}

3.4 完整 LangChain + LangGraph 示例

python

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langgraph.prebuilt import ToolNode

# 1. 定义工具（LangChain）
@tool
def calculate(expression: str) -> str:
    """执行数学计算"""
    return str(eval(expression))

tools = [calculate]

# 2. 定义提示词（LangChain）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个数学助手，可以使用计算工具"),
    ("placeholder", "{messages}")
])

# 3. 定义 Agent 节点（LangChain + LangGraph）
def agent(state: MessagesState):
    llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
    chain = prompt | llm
    response = chain.invoke({"messages": state["messages"]})
    return {"messages": [response]}

# 4. 条件路由（LangGraph）
def should_continue(state: MessagesState):
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

# 5. 构建图（LangGraph）
graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

app = graph.compile()

# 6. 测试
response = app.invoke({
    "messages": [("user", "计算 (25 + 75) * 2 / 10")]
})
print(response["messages"][-1].content)

第 4 部分：最佳实践与常见问题

4.1 错误处理

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

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

try:
    response = llm.invoke([HumanMessage(content="你好")])
    print(response.content)
except Exception as e:
    print(f"错误：{type(e).__name__}: {str(e)}")

4.2 重试策略

python

from langchain_openai import ChatOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_llm_with_retry(message: str):
    llm = ChatOpenAI(model="gpt-4o-mini")
    return llm.invoke([HumanMessage(content=message)])

response = call_llm_with_retry("解释什么是链式法则")
print(response.content)

4.3 Token 计数

python

from langchain_openai import ChatOpenAI
from langchain.callbacks import get_openai_callback

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

with get_openai_callback() as cb:
    response = llm.invoke([HumanMessage(content="写一首诗")])

    print(f"提示词 Tokens: {cb.prompt_tokens}")
    print(f"回复 Tokens: {cb.completion_tokens}")
    print(f"总计 Tokens: {cb.total_tokens}")
    print(f"总计成本: ${cb.total_cost:.6f}")

运行结果：

提示词 Tokens: 12
回复 Tokens: 48
总计 Tokens: 60
总计成本: $0.000090

4.4 成本优化

python

# 1. 选择合适的模型
llm_cheap = ChatOpenAI(model="gpt-4o-mini")      # 便宜
llm_expensive = ChatOpenAI(model="gpt-4o")        # 贵但性能好

# 2. 限制输出长度
llm = ChatOpenAI(model="gpt-4o-mini", max_tokens=100)

# 3. 使用缓存机制（相同输入不重复调用）
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

set_llm_cache(InMemoryCache())

# 第一次调用（真实调用）
response1 = llm.invoke([HumanMessage(content="什么是 AI")])

# 第二次调用（从缓存读取，不消耗 tokens）
response2 = llm.invoke([HumanMessage(content="什么是 AI")])

总结

7 大核心组件回顾

组件	作用	在 LangGraph 中的应用
Chat Models	调用 LLM 接口	节点函数中调用 LLM
Prompts	结构化提示词	为 Agent 设计系统提示词
Output Parsers	解析 LLM 输出	从文本提取结构化数据
Tools	扩展 LLM 能力	ToolNode 自动执行
Messages	对话消息类型	MessagesState 管理
Retrievers	RAG 检索	知识库增强
LCEL	链式编排	节点内部数据流

LangChain vs LangGraph 适用场景

简单场景（LCEL）:
  提示词 → LLM → 解析 → 输出

复杂场景（LangGraph）:
  用户输入 → Agent决策 → 工具调用 → 结果汇总 → 循环决策

学习建议

先掌握 LangChain 基础：Chat Models、Prompts、Tools
理解消息对象：这是 LangGraph 最基础的数据结构
学习 LCEL：虽然大部分时候用 LangGraph，但节点内部仍然会用 LCEL 来组合组件
实践 RAG：检索增强是生产环境最常用的模式
整合实战：将 LangChain 组件集成到 LangGraph 工作流

下一步

第 1 章：深入 LangGraph 核心概念（Chain、Router、Agent）
第 2 章：State Schema 高级设计
第 3 章：断点调试与人机协作
第 4 章：高级模式（并行、子图、Map-Reduce）

记住：LangChain 是"工具箱"，LangGraph 是"施工图纸"。掌握 LangChain，你就能灵活使用各种 AI 组件；掌握 LangGraph，你就能编排出强大的智能体系统！

0.3 补充：LangChain 快速介绍 - 从 LangGraph 视角理解 ​

1. LangChain 是 LangGraph 的基础 ​

类比理解 ​

📊 LangChain 与 LangGraph 的技术架构 ​

📚 LangChain 7 大核心组件概览 ​

2.1 Chat Models（聊天模型）- LLM 调用的基础 ​

基础调用 ​

🔍 Python 知识点：函数参数与默认值 ​

消息对象 ​

🔍 Python 知识点：类与对象 ​

流式输出 ​

🔍 Python 知识点：生成器（Generator） ​

多模型支持 ​

⚡ 在 LangGraph 中的应用（重要） ​

2.2 Prompts（提示词模板）- 结构化提示词管理 ​

基础模板 ​

🔍 Python 知识点：字典与格式化 ​

🔍 Python 知识点：管道操作符（|） ​

Few-Shot 提示模板 ​

模板变量复用 ​

⚡ 在 LangGraph 中的应用（重要） ​

2.3 Output Parsers（输出解析器）- 结构化 LLM 输出 ​

StrOutputParser（字符串解析） ​

JsonOutputParser（JSON 解析） ​

🔍 Python 知识点：字典（Dictionary） ​

PydanticOutputParser（结构化解析） ​

🔍 Python 知识点：类型注解与 Pydantic ​

⚡ 在 LangGraph 中的应用（重要） ​

2.4 Tools（工具）- 扩展 LLM 能力 ​

使用 @tool 装饰器定义工具 ​

🔍 Python 知识点：装饰器（Decorator） ​

工具绑定到 LLM ​

执行工具调用 ​

使用 StructuredTool 定义复杂工具 ​

⚡ 在 LangGraph 中的应用（重要） ​

2.5 Messages（消息对象）- 对话历史管理 ​

消息类型概览 ​

消息的特殊属性 ​

消息历史管理 ​

🔍 Python 知识点：列表（List） ​

消息序列化 ​

⚡ 在 LangGraph 中的应用（重要） ​

2.6 Retrievers & RAG（检索增强生成）- 让 LLM 访问外部知识 ​

向量存储基础 ​

🔍 Python 知识点：Enumerate ​

带分数的检索 ​

完整 RAG 流程 ​

⚡ 在 LangGraph 中的应用（重要） ​

2.7 LCEL（LangChain Expression Language）- 组件编排语言 ​

基础链组合 ​

RunnablePassthrough（数据传递） ​

RunnableLambda（自定义函数） ​

并行执行 ​

条件分支 ​

LCEL vs LangGraph 对比 ​

📊 适用场景对比 ​

代码对比示例 ​

第 3 部分：LangChain 与 LangGraph 整合实战 ​

3.1 Chat Models 整合 ​

3.2 Tools 整合 ​

3.3 Prompts 整合 ​

3.4 完整 LangChain + LangGraph 示例 ​

第 4 部分：最佳实践与常见问题 ​

4.1 错误处理 ​

4.2 重试策略 ​

4.3 Token 计数 ​

4.4 成本优化 ​

总结 ​

7 大核心组件回顾 ​

LangChain vs LangGraph 适用场景 ​

学习建议 ​

下一步 ​

0.3 补充：LangChain 快速介绍 - 从 LangGraph 视角理解

1. LangChain 是 LangGraph 的基础

类比理解

📊 LangChain 与 LangGraph 的技术架构

📚 LangChain 7 大核心组件概览

2.1 Chat Models（聊天模型）- LLM 调用的基础

基础调用

🔍 Python 知识点：函数参数与默认值

消息对象

🔍 Python 知识点：类与对象

流式输出

🔍 Python 知识点：生成器（Generator）

多模型支持

⚡ 在 LangGraph 中的应用（重要）

2.2 Prompts（提示词模板）- 结构化提示词管理

基础模板

🔍 Python 知识点：字典与格式化

🔍 Python 知识点：管道操作符（|）

Few-Shot 提示模板

模板变量复用

⚡ 在 LangGraph 中的应用（重要）

2.3 Output Parsers（输出解析器）- 结构化 LLM 输出

StrOutputParser（字符串解析）

JsonOutputParser（JSON 解析）

🔍 Python 知识点：字典（Dictionary）

PydanticOutputParser（结构化解析）

🔍 Python 知识点：类型注解与 Pydantic

⚡ 在 LangGraph 中的应用（重要）

2.4 Tools（工具）- 扩展 LLM 能力

使用 @tool 装饰器定义工具

🔍 Python 知识点：装饰器（Decorator）

工具绑定到 LLM

执行工具调用

使用 StructuredTool 定义复杂工具

⚡ 在 LangGraph 中的应用（重要）

2.5 Messages（消息对象）- 对话历史管理

消息类型概览

消息的特殊属性

消息历史管理

🔍 Python 知识点：列表（List）

消息序列化

⚡ 在 LangGraph 中的应用（重要）

2.6 Retrievers & RAG（检索增强生成）- 让 LLM 访问外部知识

向量存储基础

🔍 Python 知识点：Enumerate

带分数的检索

完整 RAG 流程

⚡ 在 LangGraph 中的应用（重要）

2.7 LCEL（LangChain Expression Language）- 组件编排语言

基础链组合

RunnablePassthrough（数据传递）

RunnableLambda（自定义函数）

并行执行

条件分支

LCEL vs LangGraph 对比

📊 适用场景对比

代码对比示例

第 3 部分：LangChain 与 LangGraph 整合实战

3.1 Chat Models 整合

3.2 Tools 整合

3.3 Prompts 整合

3.4 完整 LangChain + LangGraph 示例

第 4 部分：最佳实践与常见问题

4.1 错误处理

4.2 重试策略

4.3 Token 计数

4.4 成本优化

总结

7 大核心组件回顾

LangChain vs LangGraph 适用场景

学习建议

下一步