LangChain详解

一 LangChain 概述与生态系统

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

1.1 生态系统分层

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

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 核心组件分类

三 快速入门

# 安装依赖 # 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 代理开发的潮流。