LangChain核心组件最新文档及示例代码（2025年更新）

一、模型（Models）

LLM初始化示例（支持GPT-4 Turbo及本地模型）：

python

from langchain_openai import ChatOpenAI
from langchain_ollama import OllamaLLM# OpenAI模型
openai_llm = ChatOpenAI(model_name="gpt-4-turbo-2025",temperature=0.7,max_tokens=2048
)# 本地模型（Ollama部署）
local_llm = OllamaLLM(base_url="http://localhost:11434",model="gemma2"
)

嵌入模型（支持动态维度调整）：

python

from langchain_openai import OpenAIEmbeddingsembeddings = OpenAIEmbeddings(model="text-embedding-3-large",dimensions=3072  # 支持维度压缩至1536维保持90%准确率
)

二、提示词（Prompts）

聊天模板示例（带角色定义）：

python

from langchain_core.prompts import ChatPromptTemplatechat_template = ChatPromptTemplate.from_messages([("system", "你是{role}，专门帮助用户解决{domain}相关问题"),("human", "我的问题是：{question}"),("ai", "我理解您的问题，让我详细解答")
])# 调用示例
messages = chat_template.invoke({"role": "技术专家","domain": "人工智能","question": "什么是深度学习？"
})

条件逻辑模板：

python

conditional_template = PromptTemplate.from_template("""
{%- if user_type == "premium" -%}
欢迎尊贵用户！您可享受高级功能
{%- else -%}
欢迎普通用户！
{%- endif -%}
请问需要帮助吗？关于：{topic}
""")

三、链（Chains）

顺序链示例（多步骤处理）：

python

from langchain.chains import LLMChain, SequentialChain# 步骤1：生成问题
question_chain = LLMChain(llm=openai_llm,prompt=question_prompt,output_key="question"
)# 步骤2：验证答案
validation_chain = LLMChain(llm=openai_llm,prompt=validation_prompt,output_key="final_answer"
)# 组合顺序链
full_chain = SequentialChain(chains=[question_chain, validation_chain],input_variables=["topic"],output_variables=["final_answer"]
)

LCEL表达式（更简洁的链定义）：

python

from langchain_core.runnables import RunnablePassthrough# 文档问答链
qa_chain = ({"context": retriever, "question": RunnablePassthrough()}| prompt_template| openai_llm| StrOutputParser()
)

四、记忆（Memory）

会话记忆实现（滑动窗口优化）：

python

from langchain.memory import ConversationBufferMemorymemory = ConversationBufferMemory(memory_key="chat_history",return_messages=True,k=5  # 保留最近5轮对话
)# 带记忆的对话链
conversation_chain = ConversationChain(llm=openai_llm,memory=memory,verbose=True
)

五、代理（Agents）

工具集成示例（计算器+天气API）：

python

from langchain.agents import Tool# 计算器工具
def math_processor(expression):return eval(expression)  # 生产环境需添加安全沙箱calculator_tool = Tool(name="Calculator",func=math_processor,description="用于执行数学计算"
)# 天气工具
weather_tool = Tool(name="WeatherAPI",func=get_weather_data,description="查询实时天气数据"
)# 工具列表
tools = [calculator_tool, weather_tool]

LangGraph代理（支持循环逻辑）：

python

from langgraph.graph import StateGraph, END# 定义状态
class AgentState(TypedDict):input: strplan: List[str]completed_tasks: List[Tuple[str, str]]# 定义节点
def execute_task(state):task = state["plan"][0]# 执行任务逻辑return {"completed_tasks": state["completed_tasks"] + [(task, result)],"plan": state["plan"][1:]}# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("execute", execute_task)
workflow.set_entry_point("execute")
workflow.add_edge("execute", END)
app = workflow.compile()

六、数据连接（Data Connection）

文档处理流程：

python

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import FAISS# 1. 加载文档
loader = PyPDFLoader("technical_manual.pdf")
documents = loader.load()# 2. 文本分割（按1024 tokens分块）
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=200
)
docs = text_splitter.split_documents(documents)# 3. 创建向量存储
vectorstore = FAISS.from_documents(documents=docs,embedding=OpenAIEmbeddings()
)# 4. 检索器（支持MMR算法）
retriever = vectorstore.as_retriever(search_type="mmr",search_kwargs={"k": 5}
)

七、注意事项与常见问题

1. 版本兼容性：

   • LangChain v0.3+ 需显式安装集成包（如langchain-openai）
   • 移除langchain-community依赖，需单独安装

2. 性能优化：

   • 使用merge_message_runs方法取消深拷贝提升性能
   • 长对话采用ConversationSummaryMemory压缩历史

3. 错误处理：

   python

   from langchain_core.callbacks import FileCallbackHandler# 全链路日志
   handler = FileCallbackHandler('llm_logs.json')
   chain.run(input, callbacks=[handler])
   

4. 安全最佳实践：

   • API密钥使用环境变量管理
   • 数据库操作限制最小权限
   • 工具调用添加输入验证
     """)

七、v0.3新特性与迁移指南

核心变更

1. Pydantic 2支持

python

# 旧版本
from pydantic import BaseModel as BaseModelV1# v0.3+
from pydantic.v1 import BaseModel as BaseModelV1  # 兼容旧代码
from pydantic import BaseModel  # Pydantic 2原生支持

1. 独立集成包

bash

# 安装独立集成包
pip install langchain-openai  # OpenAI集成
pip install langchain-anthropic  # Anthropic集成

1. 流式处理优化

python

# 流式工具调用
async for event in agent.astream_events(inputs, version="v1"):if event["event"] == "on_tool_start":print(f"调用工具: {event['name']}")elif event["event"] == "on_tool_end":print(f"工具返回: {event['data']['output']}")

迁移步骤

1. 升级核心包

bash

pip install -U langchain-core langchain

1. 更新导入路径

python

# 旧版本
from langchain.llms import OpenAI# v0.3+
from langchain_openai import ChatOpenAI

八、实战案例：构建RAG智能问答系统

完整流程

python

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain_openai import OpenAIEmbeddings, ChatOpenAI# 1. 加载文档
loader = PyPDFLoader("enterprise_manual.pdf")
documents = loader.load()# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=200
)
splits = text_splitter.split_documents(documents)# 3. 创建向量库
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embeddings)# 4. 构建问答链
qa_chain = RetrievalQA.from_chain_type(llm=ChatOpenAI(model="gpt-4-turbo"),chain_type="stuff",retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),return_source_documents=True
)# 5. 执行查询
result = qa_chain.invoke({"query": "如何配置API密钥？"})
print(result["result"])
# 输出引用源
for doc in result["source_documents"]:print(f"来源: {doc.metadata['source']}, 页码: {doc.metadata['page']}")

九、常见问题与解决方案

1. 版本兼容性问题

错误：ModuleNotFoundError: No module named 'langchain.llms'

解决：安装独立集成包

bash

pip install langchain-openai

2. 工具调用参数错误

错误：TypeError: complex_tool() missing 1 required argument

解决：使用参数验证

python

from pydantic import BaseModelclass ToolInput(BaseModel):int_arg: intfloat_arg: floatdict_arg: dict@tool(args_schema=ToolInput)
def complex_tool(int_arg: int, float_arg: float, dict_arg: dict) -> int:return int_arg * float_arg

3. 内存溢出

错误：MemoryError when processing large documents

解决：增量处理

python

# 批量处理文档
for i in range(0, len(documents), 10):batch = documents[i:i+10]vectorstore.add_documents(batch)

4. 检索结果相关性低

优化方案：混合检索策略

python

from langchain.retrievers import BM25Retriever, EnsembleRetriever# 向量检索 + BM25检索
vector_retriever = vectorstore.as_retriever()
bm25_retriever = BM25Retriever.from_documents(splits)ensemble_retriever = EnsembleRetriever(retrievers=[bm25_retriever, vector_retriever],weights=[0.3, 0.7]
)

十、最佳实践与性能优化

1. 缓存策略

python

from langchain.cache import InMemoryCache# 启用缓存
llm.cache = InMemoryCache()

2. 异步处理

python

# 异步调用示例
async def async_qa(query):return await qa_chain.ainvoke({"query": query})

3. 监控与调试

python

# 启用LangSmith跟踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your_api_key"

十一、项目实战：智能数据分析助手

核心功能

python

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import PythonAstREPLTool
import pandas as pd# 1. 加载数据
df = pd.read_csv("sales_data.csv")# 2. 定义工具
tool = PythonAstREPLTool(locals={"df": df})
tools = [tool]# 3. 创建代理
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)# 4. 执行分析
result = agent_executor.invoke({"input": "分析2024年各季度销售额增长率，生成折线图"
})

可视化输出

python

# 生成图表
import matplotlib.pyplot as plt
df.groupby("季度")["销售额"].sum().plot(kind="line")
plt.savefig("sales_trend.png")

十二、总结与进阶学习

推荐学习路径

1. 基础阶段：掌握Models/Prompts/Chains核心用法
2. 进阶阶段：实现RAG系统与工具调用
3. 高级阶段：使用LangGraph构建多智能体协作

资源推荐

• 官方文档：python.langchain.com/docs
• 实战项目：github.com/langchain-ai/langchain
• 视频课程：LangChain Academy (LangSmith账号免费访问)

        通过本文档，你已掌握LangChain从基础到高级的全部核心技能，可独立构建企业级LLM应用。建议重点关注RAG优化与多智能体协作方向，这是当前工业界的主流需求。