一 LangChain 概述与生态系统

LangChain 是一个开源的代理工程平台，提供预构建的代理架构和与任何模型 / 工具的集成，使开发者能够快速构建能够适应生态系统变化的 AI 代理。截至 2026 年 5 月，最新稳定版本为v1.1.0，于 2025 年 12 月发布，标志着框架进入成熟稳定阶段，承诺直到 2.0 版本前不会有破坏性变更LangChain。

1.1 生态系统分层

LangChain 生态系统由三个主要部分组成，满足不同复杂度的开发需求：

层级	产品	适用场景	特点
最高级	Deep Agents	快速构建复杂代理	"开箱即用"，包含自动上下文压缩、虚拟文件系统、子代理生成等功能
高级	LangChain	标准代理开发	简单易用，高度灵活，基于 LangGraph 运行时
低级	LangGraph	高度自定义代理	细粒度控制，支持确定性和代理工作流的组合，生产级长运行代理

1.2 包结构

LangChain 采用模块化包设计，只安装你需要的部分：

• langchain-core：基础抽象接口、LCEL 原语和基类
• langchain：核心代理实现、预构建链和工具
• langchain-openai、langchain-anthropic等：官方维护的模型提供商集成
• langchain-community：社区维护的第三方集成
• langchain-legacy：v0.x 版本的遗留功能（用于迁移）

pip install --pre -U langchain
pip install --pre langchain-openai
pip install langchain-community
pip install python-dotenv pypdf

二 核心架构与组件

LangChain 的核心组件按照从底层到上层的顺序组织，形成一个完整的 AI 应用开发栈：

2.1 组件生态系统

输入处理 → 嵌入与存储 → 检索 → 生成 → 编排

2.2 核心组件分类

类别	目的	关键组件	主要用途
模型	AI 推理与生成	聊天模型、LLMs、嵌入模型	文本生成、推理、语义理解
工具	外部能力扩展	APIs、数据库、文件系统	Web 搜索、数据访问、计算任务
代理	编排与决策	ReAct 代理、工具调用代理	非确定性工作流、自主决策
记忆	上下文保存	消息历史、自定义状态	多轮对话、有状态交互
检索器	信息获取	向量检索器、Web 检索器	RAG、知识库搜索
文档处理	数据摄入	加载器、分割器、转换器	PDF 处理、网页抓取、数据清洗
向量存储	语义搜索	Chroma、Pinecone、FAISS	相似性搜索、嵌入存储

三 快速入门

# 安装依赖
# pip install -qU langchain "langchain[openai]"
from deepseek_test import deep_llm as llm
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage


# 定义工具
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    return f"{city}今天天气晴朗，气温25摄氏度。"

def test1():
    # 创建代理
    agent = create_agent(
        model=llm,  # 支持任何模型提供商
        tools=[get_weather],
        system_prompt="你是一个乐于助人的助手。"
    )

    # 调用代理
    result = agent.invoke({
        "messages": [HumanMessage("上海今天天气怎么样？")]
    })

    # 输出结果
    print(result["messages"][-1].content)
    # 上海今天天气晴朗，气温25摄氏度。


if __name__ == '__main__':
    test1()

四 核心功能详解

4.1 模型集成

LangChain 提供了统一的模型接口，支持几乎所有主流大语言模型提供商：

from langchain.chat_models import init_chat_model

# 初始化不同提供商的模型
    openai_model = init_chat_model("openai:gpt-4o")
    anthropic_model = init_chat_model("anthropic:claude-3-opus-20240229")
    google_model = init_chat_model("google:gemini-1.5-pro")
    ollama_model = init_chat_model("ollama:llama3")
    deepseek_model = init_chat_model("deepseek:deepseek-chat", temperature=0, api_key=dp_api_key)

    # 统一调用方式
    response = deepseek_model.invoke("你好，世界！")
    print(response.content)

模型配置文件（v1.1 + 新特性）:

# 查看模型支持的能力
print(openai_model.profile)
# {
#   "max_input_tokens": 128000,
#   "image_inputs": True,
#   "reasoning_output": True,
#   "tool_calling": True,
#   "structured_output": True
# }

4.2 工具系统

工具是代理与外部世界交互的桥梁。LangChain 支持多种工具定义方式

from langchain.tools import tool
from pydantic import BaseModel, Field

# 方式1：使用@tool装饰器
@tool
def calculate(expression: str) -> float:
    """计算数学表达式的值。
    
    Args:
        expression: 要计算的数学表达式，如"2+3*4"
    """
    return eval(expression)

# 方式2：使用Pydantic模型定义
class SearchInput(BaseModel):
    query: str = Field(description="搜索查询词")
    num_results: int = Field(description="返回结果数量", default=5)

@tool(args_schema=SearchInput)
def web_search(query: str, num_results: int = 5) -> list[str]:
    """执行网络搜索并返回结果。"""
    # 实际实现...
    return [f"结果{i}: ..." for i in range(num_results)]

# 方式3：返回内容和工件（用于RAG）
@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """检索上下文信息以帮助回答问题。"""
    retrieved_docs = vector_store.similarity_search(query, k=2)
    serialized = "\n\n".join(
        f"来源: {doc.metadata}\n内容: {doc.page_content}"
        for doc in retrieved_docs
    )
    return serialized, retrieved_docs

4.3 结构化输出

LangChain 1.0 将结构化输出直接集成到主代理循环中，消除了额外的 LLM 调用，降低了延迟和成本：

from pydantic import BaseModel
from langchain.agents import create_agent

# 定义输出结构
class WeatherInfo(BaseModel):
    city: str
    temperature: float
    condition: str
    humidity: int


def test3():
    # 创建具有结构化输出的代理
    agent = create_agent(
        model=deepseek_model,
        tools=[get_weather],
        response_format=WeatherInfo
    )

    # 调用代理
    result = agent.invoke({
        "messages": [HumanMessage("北京今天的天气怎么样？")]
    })

    # 直接获取结构化结果
    weather_info = result["structured_response"]
    print(f"城市: {weather_info.city}")
    print(f"温度: {weather_info.temperature}°C")
    print(f"天气: {weather_info.condition}")
    print(f"湿度: {weather_info.humidity}%")

4.4 中间件系统

中间件是 LangChain 1.0 引入的强大功能，允许你在代理执行的每个步骤进行细粒度控制：

from langchain.agents.middleware import before_model, after_model, wrap_model_call
from langchain.agents import AgentState
from langchain.agents.middleware.types import ModelRequest, ModelResponse
from langgraph.runtime import Runtime

# 日志中间件
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> dict | None:
    print(f"即将调用模型，消息数量: {len(state['messages'])}")
    return None

# 输出验证中间件
@after_model(can_jump_to=["end"])
def validate_output(state: AgentState, runtime: Runtime) -> dict | None:
    last_message = state["messages"][-1]
    if "敏感词" in last_message.content:
        return {
            "messages": "抱歉，我无法回答这个问题。",
            "jump_to": "end"
        }
    return None

# 重试中间件
@wrap_model_call
def retry_model(
    request: ModelRequest,
    handler: callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as e:
            if attempt == 2:
                raise
            print(f"重试 {attempt + 1}/3，错误: {e}")

# 动态模型选择中间件
@wrap_model_call
def dynamic_model_selection(
    request: ModelRequest,
    handler: callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    if len(request.messages) > 10:
        # 长对话使用更便宜的模型
        request.model = "openai:gpt-4o-mini"
    else:
        # 短对话使用更强大的模型
        request.model = "openai:gpt-4o"
    return handler(request)

# 使用中间件创建代理
def test4():
    # 使用中间件创建代理
    agent = create_agent(
        model=deepseek_model,
        tools=[get_weather, calculate],
        middleware=[log_before_model, validate_output, retry_model, dynamic_model_selection]
    )

4.5 记忆系统

LangChain 提供了多种记忆实现，用于保存对话历史和代理状态：

from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

def test5():
    # 方式1：使用MemorySaver（推荐，基于LangGraph）
    memory = MemorySaver()
    agent = create_agent(
        model=deepseek_model,
        tools=[get_weather],
        checkpointer=memory
    )

    # 多轮对话
    config = {"configurable": {"thread_id": "1"}}

    # 第一轮
    result1 = agent.invoke(
        {"messages": [HumanMessage("你好，我叫张三")]},
        config=config
    )
    print(result1["messages"][-1].content)  # 你好张三！有什么我可以帮助你的吗？

    # 第二轮（代理记得用户名字）
    result2 = agent.invoke(
        {"messages": [HumanMessage("我叫什么名字？")]},
        config=config
    )
    print(result2["messages"][-1].content)  # 你叫张三。

4.6 RAG 实现

检索增强生成（RAG）是 LangChain 最常用的应用场景之一。以下是一个完整的 RAG 代理实现：

import bs4
from langchain.tools import tool
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_core.messages import HumanMessage
from langchain_core.vectorstores import InMemoryVectorStore
from langchain.agents import create_agent
from deepseek_test import deepseek_model

# 1. 加载和处理文档
loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
docs = loader.load()

# 2. 文本分块
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
all_splits = text_splitter.split_documents(docs)

# 3. 创建向量存储
embeddings = OpenAIEmbeddings()
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents=all_splits)

# 4. 创建检索工具
@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """检索关于LLM代理的信息以帮助回答问题。"""
    retrieved_docs = vector_store.similarity_search(query, k=3)
    serialized = "\n\n".join(
        f"来源: {doc.metadata.get('title', '未知')}\n内容: {doc.page_content}"
        for doc in retrieved_docs
    )
    return serialized, retrieved_docs

# 5. 创建RAG代理
agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[retrieve_context],
    system_prompt="""你是一个关于LLM代理的专家。
    请使用提供的检索工具来回答用户的问题。
    如果检索到的信息不足以回答问题，请诚实地说你不知道。"""
)

# 6. 提问
result = agent.invoke({
    "messages": [HumanMessage("什么是任务分解？LLM代理如何进行任务分解？")]
})

print(result["messages"][-1].content)

五 LangChain Expression Language (LCEL)

LCEL 是 LangChain 的声明式管道语言，用于将组件组合成复杂的工作流。它支持流式处理、并行执行、错误处理等高级特性。

5.1 基本原语

from langchain_core.runnables import (
    RunnableLambda, RunnablePassthrough, RunnableParallel, RunnableBranch
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

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

# 1. RunnableLambda：将函数转换为可运行对象
runnable = RunnableLambda(lambda x: x.upper())
print(runnable.invoke("hello"))  # HELLO

# 2. RunnablePassthrough：传递输入不变
passthrough = RunnablePassthrough()
print(passthrough.invoke("hello"))  # hello

# 3. RunnableParallel：并行执行多个可运行对象
parallel = RunnableParallel(
    original=RunnablePassthrough(),
    uppercase=lambda x: x.upper(),
    lowercase=lambda x: x.lower()
)
print(parallel.invoke("Hello"))
# {'original': 'Hello', 'uppercase': 'HELLO', 'lowercase': 'hello'}

# 4. RunnableBranch：条件分支
branch = RunnableBranch(
    (lambda x: x > 0, lambda x: f"正数: {x}"),
    (lambda x: x < 0, lambda x: f"负数: {x}"),
    lambda x: "零"
)
print(branch.invoke(5))  # 正数: 5
print(branch.invoke(-3))  # 负数: -3
print(branch.invoke(0))  # 零

5.2 构建复杂管道

# 简单的问答链
prompt = ChatPromptTemplate.from_template("回答以下问题：{question}")
chain = prompt | model | parser
print(chain.invoke({"question": "什么是人工智能？"}))

# 带上下文的RAG链
rag_prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题：
上下文：{context}
问题：{question}
""")

retriever = vector_store.as_retriever()

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | rag_prompt
    | model
    | parser
)

print(rag_chain.invoke("什么是ReAct代理？"))

六、生产级开发：LangSmith 集成

LangSmith 是 LangChain 官方提供的生产级开发平台，提供可观测性、评估、提示工程和部署功能。

6.1 基本集成

import os

# 设置环境变量
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "你的API密钥"
os.environ["LANGSMITH_PROJECT"] = "我的代理项目"

# 现在所有的LangChain调用都会自动被追踪
result = agent.invoke({
    "messages": [HumanMessage("上海今天天气怎么样？")]
})

6.2 高级追踪

from langsmith import traceable
import langsmith as ls

# 自定义追踪
@traceable(
    name="我的自定义函数",
    tags=["production", "email-assistant"],
    metadata={"version": "1.0.0"}
)
def my_custom_function(input_data):
    # 调用LangChain组件
    return chain.invoke(input_data)

# 上下文管理器控制追踪
with ls.tracing_context(
    project_name="测试项目",
    tags=["test"],
    metadata={"user_id": "user_123"}
):
    result = agent.invoke({"messages": [HumanMessage("测试问题")]})

七、最佳实践与常见模式

7.1 代理开发最佳实践

1. 从简单开始：先使用create_agent构建基础代理，再逐步添加功能
2. 使用中间件：将横切关注点（日志、验证、重试）提取到中间件中
3. 结构化输出：尽可能使用结构化输出，避免解析自由文本
4. 添加错误处理：为工具调用和模型调用添加适当的错误处理
5. 使用 LangSmith：从开发初期就集成 LangSmith 进行调试和评估

7.2 常见设计模式

1. RAG 模式：检索增强生成，解决知识过时和领域限制问题
2. 工具调用模式：让代理能够使用外部工具完成复杂任务
3. Map-Reduce 模式：处理大量文档或长文本
4. 路由模式：根据查询类型将请求路由到不同的处理链
5. 多代理模式：多个专业代理协作完成复杂任务

八、迁移指南：从 v0.x 到 v1.0

8.1 主要变更

• Python 版本要求：Python 3.10+（不再支持 Python 3.9）
• 核心 API：create_agent取代了create_react_agent
• 导入路径：许多组件从langgraph.prebuilts移动到了langchain.agents
• 消息 API：.text()方法改为.text属性
• 包结构：遗留功能移动到了langchain-legacy包

8.2 迁移步骤

1. 更新依赖：pip install -U langchain langchain-core
2. 替换导入：

# 旧
from langgraph.prebuilts import create_agent, ToolNode, AgentState

# 新
from langchain.agents import create_agent, ToolNode, AgentState

3. 更新消息访问：

# 旧
text = response.text()

# 新
text = response.text

4. 对于遗留功能，安装langchain-legacy并更新导入：

# 旧
from langchain import SomeLegacyClass

# 新
from langchain_legacy import SomeLegacyClass

九、未来发展方向

LangChain 团队正在以下几个方向持续投入：

1. Deep Agents：进一步增强开箱即用的代理能力
2. 多模态支持：更好地支持图像、音频和视频输入输出
3. 企业级功能：增强安全性、合规性和可扩展性
4. 模型优化：自动选择最佳模型和提示策略
5. 低代码工具：让非技术人员也能构建 AI 代理

总结

LangChain 1.0 是一个成熟稳定的代理工程平台，通过create_agent抽象和中间件系统，极大地简化了 AI 代理的开发过程。它提供了统一的模型接口、强大的工具系统、灵活的记忆管理和完整的 RAG 支持，同时通过 LangSmith 提供了生产级的可观测性和评估能力。

无论你是想快速构建一个简单的聊天机器人，还是开发复杂的企业级 AI 应用，LangChain 都能提供合适的工具和抽象层次。随着生态系统的不断发展，LangChain 将继续引领 AI 代理开发的潮流。