LangChain
中文文档

LangChain 支持 Python 3.9 及以上版本。官方推荐使用虚拟环境隔离项目依赖， 避免不同 LangChain 版本之间的冲突。以下是从零开始配置完整 LangChain 开发环境的标准流程。

核心包安装

LangChain 1.0 将框架拆分为多个独立的发布包。langchain-core 包含核心抽象， langchain 是主框架，各模型集成通过独立的 langchain-openai、 langchain-anthropic 等包安装，互不干扰。

# 安装 LangChain 核心框架
pip install langchain

# 安装 OpenAI 模型集成（最常用）
pip install langchain-openai

# 安装社区集成包（包含大量第三方工具）
pip install langchain-community

# 安装 LangGraph（多智能体编排，推荐一并安装）
pip install langgraph

环境变量配置

LangChain 通过环境变量读取 API Key 和各种配置。推荐在项目根目录创建 .env 文件， 使用 python-dotenv 自动加载，不要把 Key 硬编码在代码里。

# .env 文件示例
OPENAI_API_KEY=sk-your-openai-key

# 启用 LangSmith 调用链追踪（可选但强烈推荐）
LANGSMITH_API_KEY=ls_your-langsmith-key
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=my-langchain-project

# 如果使用免费的 Groq 推理 API
GROQ_API_KEY=gsk_your-groq-key

# 在 Python 文件最开头加载 .env
from dotenv import load_dotenv
load_dotenv()  # 必须在 import langchain 之前调用

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke("你好，LangChain！")
print(response.content)

使用 Groq 免费 API

OpenAI API 需要付费，刚开始学习 LangChain 建议使用 Groq 的免费推理 API， 速度极快（Llama 3 每秒 800+ tokens），免费额度足够学习使用。

pip install langchain-groq

from langchain_groq import ChatGroq

# 模型名称参考 Groq 官网：console.groq.com
llm = ChatGroq(model="llama-3.1-70b-versatile")
response = llm.invoke("用 LangChain 能做什么？")
print(response.content)

常用 LangChain 包速查

包名	用途	备注
langchain	LangChain 主框架	必装
langchain-core	核心抽象层	随 langchain 自动安装
langchain-openai	OpenAI / Azure 模型	最常用
langchain-anthropic	Claude 系列模型	—
langchain-groq	Groq 免费推理	学习首选
langchain-community	海量第三方集成	按需安装
langgraph	多智能体编排	生产必装
langsmith	调试 / 监控 / 评估	推荐开发期启用

LCEL（LangChain Expression Language）是 LangChain 1.0 的核心编排语法。 它用 Python 的 | 管道符将"提示词模板 → 语言模型 → 输出解析器"串成一条链， 原生支持流式输出、批处理并发与异步调用，不需要写任何样板代码。

最简单的 LCEL 链

一条 LCEL 链由三部分组成：提示词模板（PromptTemplate）负责把用户变量填入固定格式， 语言模型（LLM）负责生成，输出解析器负责把 AIMessage 转成干净的字符串。

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

# 1. 定义提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的 {role}，请用中文简洁地回答。"),
    ("human",  "{question}"),
])

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

# 3. 用 LCEL 管道符 | 组合成链
chain = prompt | llm | StrOutputParser()

# 4. 调用
result = chain.invoke({
    "role":     "Python 工程师",
    "question": "什么是 LangChain LCEL？",
})
print(result)

流式输出

流式输出对用户体验影响巨大——让用户看到文字逐步出现，而不是等待几秒后一次性出现。 LangChain 的 LCEL 链只需用 .stream() 替换 .invoke()。

# 流式输出：实时打印每个 token
for chunk in chain.stream({"role": "LangChain 专家", "question": "解释 LCEL"}):
    print(chunk, end="", flush=True)

批处理并发

如果需要同时处理多个输入，.batch() 会自动并发执行， 比用 for 循环逐个调用效率高很多，特别适合批量离线处理场景。

inputs = [
    {"role": "工程师", "question": "解释 LangChain Agent"},
    {"role": "工程师", "question": "解释 LangChain RAG"},
    {"role": "工程师", "question": "解释 LangGraph"},
]

# 并发处理所有输入，config 控制最大并发数
results = chain.batch(inputs, config={"max_concurrency": 3})

Runnable 接口

LCEL 的底层是 Runnable 接口。LangChain 的每个组件——提示词、模型、解析器、检索器—— 都实现了 Runnable，因此都能用 | 串联。你也可以用 RunnableLambda 把任意 Python 函数接入 LCEL 链。

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

# 用 RunnableLambda 把普通函数封装进 LCEL 链
def add_context(inputs):
    inputs["question"] = f"[LangChain 教程] " + inputs["question"]
    return inputs

chain_with_context = (
    RunnableLambda(add_context)   # 步骤1：预处理输入
    | prompt                        # 步骤2：填充提示词模板
    | llm                           # 步骤3：LangChain LLM 生成
    | StrOutputParser()            # 步骤4：解析输出
)

LangChain Agent 是框架最核心的能力之一。Agent 让语言模型不再只是"回答问题"， 而是能够主动决策、调用工具、查看结果、再继续推进——直到完成任务。 LangChain 1.0 使用 create_agent（通过 LangGraph 实现）替代了旧版的 create_react_agent。

工具的定义方式

LangChain 工具（Tool）是 Agent 可以调用的函数。用 @tool 装饰器 标记一个 Python 函数，LangChain 会自动从函数签名和 docstring 中提取工具名称和描述， 传递给语言模型用于工具选择。

from langchain_core.tools import tool
import requests

@tool
def search_web(query: str) -> str:
    """在网络上搜索实时信息。适用于需要最新资讯的问题。"""
    # 实际使用时接入 Tavily/Serper 等搜索 API
    return f"搜索结果：关于 '{query}' 的最新信息..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式，如 '2 ** 10' 或 '(3 + 5) * 2'。"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算失败：{e}"

# 注册到工具列表
tools = [search_web, calculate]

创建 LangChain Agent

LangChain 1.0 推荐通过 LangGraph 的 create_react_agent 创建 Agent（注意：这是 langgraph 包里的新版函数，不是旧版 langchain 包里的同名函数，二者不能混用）。

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

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

# 创建具备工具调用能力的 LangChain Agent
agent = create_react_agent(
    model=llm,
    tools=tools,
    # 可选：添加系统提示词定制 Agent 行为
    state_modifier="你是一个专业的 LangChain 助手，使用工具时要说明理由。"
)

# 运行 Agent
result = agent.invoke({
    "messages": [{"role": "user", "content": "LangChain 最新版本是什么，2 的 10 次方是多少？"}]
})
print(result["messages"][-1].content)

Agent 的推理循环（ReAct）

LangChain Agent 内部运行 ReAct 推理模式（Reasoning + Acting）： 模型思考需要做什么，选择并调用工具，观察结果，再决定是否继续调用工具还是直接回答。 这个循环会一直执行，直到模型认为任务完成。

1. 思考（Reasoning）：模型分析用户问题，判断需要调用哪个工具以及传入什么参数
2. 行动（Acting）：LangChain 执行工具调用，获取真实世界的信息或执行操作
3. 观察（Observation）：工具返回结果，模型观察并更新自己的理解
4. 迭代或完成：若信息不足继续下一轮工具调用，否则生成最终回答

流式输出 Agent 的思考过程

# 流式查看 LangChain Agent 的每一步决策
for step in agent.stream(
    {"messages": [{"role": "user", "content": "搜索 LangChain 最新动态"}]},
    stream_mode="updates"
):
    for node, value in step.items():
        print(f"\n[{node}]")
        print(value["messages"][-1])

RAG（Retrieval-Augmented Generation，检索增强生成）是 LangChain 最主流的应用模式之一。 核心思路：把私有文档建立向量索引，用户提问时先检索最相关的段落， 再把这些段落作为上下文塞给语言模型，让 LangChain 模型基于真实文档而非训练记忆来回答。

RAG 全流程概览

1. 加载文档：使用 LangChain DocumentLoader 从 PDF、网页、数据库等来源读取原始文本
2. 文本分块：RecursiveCharacterTextSplitter 将长文档切成有重叠的小块
3. 生成 Embedding：OpenAIEmbeddings 将每个文本块转换为高维向量
4. 存入向量库：Pinecone 或 Chroma 建立可检索的向量索引
5. 检索：用户问题同样转为向量，检索最相似的文档块
6. 生成回答：LangChain 将检索结果注入提示词，模型据此生成有依据的回答

完整 RAG 代码示例

from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# ── 步骤 1：加载文档 ──
loader = WebBaseLoader("https://docs.langchain.com/docs/")
docs = loader.load()

# ── 步骤 2：文本分块（chunk_overlap 保证上下文连贯）──
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
)
chunks = splitter.split_documents(docs)

# ── 步骤 3 + 4：Embedding + 存入 Chroma 本地向量库 ──
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db",
)

# ── 步骤 5：创建检索器 ──
retriever = vectorstore.as_retriever(
    search_type="mmr",           # mmr 兼顾相关性和多样性
    search_kwargs={"k": 4},      # 返回最相关的 4 个文档块
)

# ── 步骤 6：构建 LangChain RAG 链 ──
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 LangChain 问答助手。仅根据以下检索到的文档回答，如果文档中没有答案就说不知道。\n\n文档：{context}"),
    ("human",  "{question}"),
])

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

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

answer = rag_chain.invoke("LangChain 的 LCEL 是什么？")
print(answer)

使用 Pinecone 构建生产级 RAG

本地 Chroma 适合开发测试，生产环境推荐使用 Pinecone 云向量数据库， 支持亿级向量规模、低延迟检索和多租户隔离。 LangChain 内置 Pinecone 集成，切换成本极低。

pip install langchain-pinecone

from langchain_pinecone import PineconeVectorStore
import os

# PINECONE_API_KEY 放在 .env 文件中
vectorstore = PineconeVectorStore.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    index_name="langchain-rag-index",  # Pinecone 控制台创建的 index 名
)

# 之后的用法和 Chroma 完全一样
retriever = vectorstore.as_retriever()

大语言模型本身是无状态的——每次调用都是独立的，不会记住上一轮对话。 LangChain 通过 Memory 组件在应用层手动维护对话历史， 在 LangGraph 中则通过 Checkpointer 实现持久化状态，支持跨会话恢复， 这是构建真实多轮对话 LangChain 应用的核心机制。

ChatMessageHistory — 基础会话历史

最基础的 Memory 组件是 ChatMessageHistory， 它在内存中维护一份消息列表，每轮对话都把 Human 和 AI 消息追加进去， 再作为历史上下文传给 LangChain 模型。

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 用字典模拟多用户的会话存储
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 LangChain 中文助手，帮助开发者学习 LangChain。"),
    MessagesPlaceholder(variable_name="history"),   # 历史消息插入位置
    ("human", "{input}"),
])

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

# 包装成带历史记录的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 同一个 session_id 的调用会自动积累历史
config = {"configurable": {"session_id": "user_001"}}
r1 = chain_with_history.invoke({"input": "LangChain 的 LCEL 是什么？"}, config=config)
r2 = chain_with_history.invoke({"input": "上面说的管道符怎么用？"},       config=config)

LangGraph Checkpointer — 持久化状态

对于需要跨进程、跨重启恢复对话的场景，InMemoryChatMessageHistory 不够用。 LangGraph 的 Checkpointer 可以把每个 Agent 步骤的完整状态序列化到 SQLite 或 PostgreSQL，程序重启后用 thread_id 恢复完整上下文。

from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

# MemorySaver：内存版 Checkpointer（开发期使用）
memory = MemorySaver()

agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o-mini"),
    tools=tools,
    checkpointer=memory,   # 传入 checkpointer 即启用持久化
)

# thread_id 代表一个对话会话，相同 thread_id 共享历史
config = {"configurable": {"thread_id": "session_42"}}

r1 = agent.invoke({"messages": [("user", "我叫小明，帮我搜索 LangChain 新版本")]}, config)
# 第二轮对话，Agent 仍记得"我叫小明"
r2 = agent.invoke({"messages": [("user", "你还记得我叫什么吗？")]}, config)

生产环境推荐：SQLite Checkpointer

from langgraph.checkpoint.sqlite import SqliteSaver

# 数据持久化到本地 SQLite 文件
with SqliteSaver.from_conn_string("./langchain_sessions.db") as memory:
    agent = create_react_agent(model=llm, tools=tools, checkpointer=memory)
    result = agent.invoke({...}, config)

LangGraph 是 LangChain 官方的多智能体编排框架， 将 Agent 的执行流程建模为一个有向图（Directed Graph）： 节点（Node）是各个处理步骤，边（Edge）是步骤间的流转逻辑。 这种设计让复杂的多步骤、多 Agent 协作系统变得可视化、可控、可调试。

StateGraph — LangGraph 核心概念

LangGraph 图的核心是 State（状态），它定义了图中所有节点共享的数据结构。 StateGraph 基于这个 State 构建图，每个节点接收当前 State，处理后返回状态更新。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# 1. 定义 LangGraph 全局 State 结构
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # add_messages 实现消息追加而非覆盖
    task_done: bool

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

# 2. 定义节点函数（接收 State，返回 State 更新）
def call_llm(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def check_completion(state: AgentState):
    # 判断是否完成任务
    return {"task_done": "完成" in state["messages"][-1].content}

# 3. 构建 LangGraph
graph_builder = StateGraph(AgentState)
graph_builder.add_node("llm", call_llm)
graph_builder.add_node("check", check_completion)

# 4. 添加边（定义流转逻辑）
graph_builder.add_edge(START, "llm")
graph_builder.add_edge("llm", "check")

# 条件边：根据状态决定去哪个节点
graph_builder.add_conditional_edges(
    "check",
    lambda s: END if s["task_done"] else "llm",
)

graph = graph_builder.compile()
result = graph.invoke({"messages": [("user", "用 LangGraph 写一个简单的 LangChain 应用")], "task_done": False})

Supervisor 多智能体模式

当任务需要多个专业 Agent 协作时，Supervisor 模式是 LangGraph 最常用的架构： 一个 Supervisor Agent 负责任务分解和路由，把子任务分别派给专门的 Worker Agent， 最后汇总结果。

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

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

# 创建两个专业 Worker Agent
research_agent = create_react_agent(
    model=llm, tools=[search_web],
    name="researcher",
    prompt="你是专业的 LangChain 信息搜集 Agent，负责查找准确信息。"
)

writer_agent = create_react_agent(
    model=llm, tools=[],
    name="writer",
    prompt="你是专业的中文技术写作 Agent，负责把信息整理成清晰的技术文档。"
)

# 创建 Supervisor，它会自动调度上面两个 Agent
supervisor = create_supervisor(
    agents=[research_agent, writer_agent],
    model=llm,
    prompt="你是 LangChain 多智能体系统的 Supervisor。先让 researcher 搜集信息，再让 writer 整理成文档。"
).compile()

result = supervisor.invoke({
    "messages": [("user", "写一篇关于 LangChain 1.0 新特性的技术文档")]
})

LangGraph 与 LangChain 的关系

对比维度	LangChain	LangGraph
定位	模型集成、链式编排、工具调用	多智能体工作流、复杂状态机
执行模型	线性链（LCEL 管道）	有向图（支持循环、分支）
状态管理	每步传递中间结果	全局 State，跨节点共享
持久化	手动维护 History	内置 Checkpointer（SQLite/Postgres）
适用场景	RAG 流水线、简单问答链	多轮对话 Agent、复杂任务自动化