1. 项目概述为什么我们需要重新认识LangChain如果你在过去一年里接触过AI应用开发尤其是基于大语言模型LLM的智能体Agent或聊天机器人那么“LangChain”这个名字你一定不陌生。它几乎成了连接业务逻辑与大模型的“标准中间件”。然而随着LangChain v1.x版本的全面发布整个框架经历了从“概念验证工具集”到“企业级应用框架”的深刻蜕变。很多开发者还停留在v0.x时代“链式调用”的简单认知里用着过时的API写着脆弱的代码然后抱怨LangChain“抽象过度”、“性能低下”。这其实是一个巨大的误解。我最近在将一个v0.1xx版本的旧项目迁移到v1.x时感触颇深。新版本不仅仅是API改名那么简单它重构了核心抽象明确了六大组件的职责边界并提供了大量生产级的最佳实践。简单来说v0.x像是给你一堆乐高零件和一张模糊的示意图而v1.x则提供了一套完整的建筑图纸、标准化的结构件甚至告诉你哪里该用钢筋加固。本教程的目的就是带你彻底吃透LangChain v1.x的这六大核心组件并通过可直接用于生产环境的代码示例让你避开我踩过的那些坑真正掌握构建稳健、可维护的AI应用的能力。2. 核心组件全解析从“玩具”到“工具”的思维转变LangChain v1.x 将构建LLM应用的要素清晰地归纳为六大核心组件Schema、Models、Prompts、Indexes、Chains和Agents。理解它们的关系和设计哲学是高效使用框架的关键。2.1 Schema数据交换的“普通话”在v0.x中我们经常需要手动拼接字符串来构造提示词Prompt或者费力地解析模型返回的文本。v1.x的Schema组件首先解决了这个问题它定义了与LLM交互的标准化数据结构。BaseMessage及其子类这是最核心的抽象。一条消息不再是一个简单的字符串而是一个具有明确角色的对象。HumanMessage: 代表用户输入。AIMessage: 代表AI模型的回复。SystemMessage: 代表系统指令用于设定AI的角色和行为边界。FunctionMessage/ToolMessage: 代表函数或工具调用的输入和输出。这种设计让多轮对话的管理变得清晰无比。一个对话历史ChatMessageHistory本质上就是一个BaseMessage的列表。Document对象处理外部知识如从PDF、网页爬取的内容时我们不再使用原始的文本字符串。Document对象封装了一段文本及其元数据如来源、页码、作者。这使得在检索增强生成RAG流程中能够精准地追溯答案的来源。实操心得养成使用SystemMessage的习惯。在v0.x中系统提示常被混在用户提示里。现在明确地用SystemMessage设置角色如“你是一个专业的翻译助手”能让模型更好地遵循指令也使得提示词模板更易维护。2.2 Models不仅仅是“换模型改API Key”Models组件提供了与各种LLM交互的统一接口。v1.x最大的改进在于“可预测性”和“生态集成”。ChatModelsvsLLMs现在严格区分了聊天模型如GPT-4, Claude和补全模型如text-davinci-003。对于绝大多数应用你应该使用ChatOpenAI,ChatAnthropic等ChatModels因为它们天然支持上述的BaseMessage序列。统一的输入输出无论底层是OpenAI、Anthropic、Cohere还是本地部署的Ollama、vLLM你都可以通过相同的invoke()、batch()、stream()方法来调用。输出也被标准化为AIMessage对象从中可以方便地提取内容 (message.content)、函数调用信息 (message.additional_kwargs) 等。生产级特性内置重试逻辑、速率限制、失败回退fallback等生产环境必备功能现在可以通过model ChatOpenAI(max_retries2, ...)或ChatOpenAI(fallback_to[another_model])轻松配置无需自己造轮子。# 生产级模型调用示例 from langchain_openai import ChatOpenAI from langchain_anthropic import ChatAnthropic from langchain_core.messages import HumanMessage, SystemMessage # 1. 基础调用 model ChatOpenAI(modelgpt-4-turbo-preview, temperature0) messages [ SystemMessage(content你是一位代码评审专家。), HumanMessage(content请评审这段Python函数def foo(x): return x1) ] response model.invoke(messages) print(response.content) # 2. 配置流式输出用于Web应用 for chunk in model.stream(messages): print(chunk.content, end, flushTrue) # 3. 配置故障转移Fallback primary_model ChatOpenAI(modelgpt-4, max_retries1) fallback_model ChatAnthropic(modelclaude-3-haiku-20240307) # 在实际项目中可以通过LCEL后文会讲将多个模型组合成带fallback的runnable2.3 Prompts告别“字符串地狱”在v0.x中构建复杂的提示模板很容易变成一堆令人头疼的f-string或.format()。v1.x的Prompts组件通过PromptTemplate和ChatPromptTemplate将其系统化。ChatPromptTemplate这是构建聊天提示的推荐方式。它由多个MessagePromptTemplate组成每个对应一种角色。你可以像搭积木一样组合系统提示、少量示例few-shot和用户输入。模板化与部分绑定你可以提前创建模板并在运行时动态注入变量。更强大的是“部分绑定”partial允许你预先填充模板中的部分变量如用户的个人资料在调用链时再传入剩余变量。from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate # 构建一个可复用的RAG提示模板 system_template SystemMessagePromptTemplate.from_template( 你是一个乐于助人的助手。请根据以下上下文回答问题。如果你不知道答案就说不知道。\n\n上下文{context} ) human_template HumanMessagePromptTemplate.from_template(问题{question}) chat_prompt ChatPromptTemplate.from_messages([system_template, human_template]) # 使用模板 formatted_messages chat_prompt.invoke({ context: LangChain是一个用于开发LLM应用的框架。, question: LangChain是什么 }) # formatted_messages 现在是一个准备好发送给ChatModel的BaseMessage列表 # 部分绑定示例假设我们有一个固定的系统角色 from langchain_core.prompts import PromptTemplate prompt PromptTemplate.from_template(以{style}的风格写一篇关于{topic}的文章。) poetic_prompt prompt.partial(style诗歌体) # 现在 poetic_prompt 只需要一个 topic 变量 print(poetic_prompt.invoke({topic: 春天}).to_string())2.4 Indexes构建你的“外部大脑”Indexes组件关乎如何让LLM访问它训练数据之外的信息即检索增强生成RAG的核心。v1.x将其拆解得更加模块化核心是Retriever检索器接口。理解数据流原始文本 - 加载器Loader - 文档Document - 分割器Splitter - 向量化Embedding - 向量存储VectorStore - 检索器Retriever。检索器Retriever这是一个关键抽象。它定义了一个get_relevant_documents(query)方法。任何实现了该接口的对象都可以作为检索器无论是基于向量的、基于关键词的如TF-IDF还是混合检索器。向量存储生态LangChain集成了Chroma、Pinecone、Weaviate、Qdrant等几乎所有主流向量数据库。v1.x的集成更稳定API更一致。# 一个完整的从文本加载到检索的示例 from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.retrievers import BaseRetriever # 1. 加载与分割文档 loader TextLoader(./state_of_the_union.txt) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size1000, chunk_overlap200) splits text_splitter.split_documents(documents) # 2. 创建向量存储并转换为检索器 embeddings OpenAIEmbeddings() vectorstore Chroma.from_documents(documentssplits, embeddingembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 检索最相关的4个片段 # 3. 使用检索器 docs retriever.invoke(总统提到了哪些关于能源的政策) for doc in docs: print(doc.page_content[:200], ...)避坑指南分块Chunking是RAG效果的“隐形杀手”。chunk_size不是越大或越小越好。对于通用文档500-1500是个不错的起点。chunk_overlap设置重叠如200字能防止关键信息被割裂。对于代码、论文等特殊格式需要使用专门的分割器如MarkdownHeaderTextSplitter。2.5 Chains用LCEL构建健壮的工作流这是v1.x革新最大的部分。旧的LLMChain、SequentialChain等被一个更强大、更灵活的概念取代LangChain Expression Language (LCEL)和Runnable 协议。核心理念一切皆可运行Everything is a Runnable。ChatModel、PromptTemplate、Retriever、甚至一个自定义的Python函数只要实现了invoke()、batch()、stream()等方法都是Runnable。你可以用|操作符像连接管道一样将它们组合起来。LCEL的优势自动流式支持用LCEL编写的链天然支持流式输出无需额外代码。并行与批量自动处理组件的并行执行和批量输入。无缝集成轻松添加日志、监控、重试、回退等中间件。易于部署LCEL链可以一键导出为LangServe API或LangSmith跟踪项。from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 使用LCEL构建一个完整的RAG链 # 1. 定义各个组件假设retriever已定义 retriever ... # 来自上一节的检索器 model ChatOpenAI(modelgpt-3.5-turbo) prompt chat_prompt # 来自上一节的提示模板 output_parser StrOutputParser() # 将AIMessage解析为字符串 # 2. 组装链 rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | model | output_parser ) # 3. 调用链 result rag_chain.invoke(LangChain是什么) print(result) # 4. 流式调用 for chunk in rag_chain.stream(LangChain是什么): print(chunk, end, flushTrue)这段代码清晰地展示了数据流用户问题 - 检索器获取上下文 - 组合成提示词 - 发送给模型 - 解析输出。RunnablePassthrough()用于直接传递输入中的某个字段这里是question。2.6 Agents让LLM学会使用工具Agents是让LLM根据目标动态决定调用哪些工具Tools的组件。v1.x重新设计了Agent的执行循环使其更可靠、更易调试。核心概念工具Tool一个可供Agent调用的函数如搜索、计算、查询数据库。使用tool装饰器可以轻松将普通函数转化为Tool。AgentExecutor这是Agent的运行时引擎。它负责管理Agent大脑和工具手脚之间的交互循环处理错误并强制最大迭代次数以防止无限循环。关键选择Agent类型v1.x提供了多种预设的Agent类型对应不同的提示策略和推理逻辑。create_react_agent: 基于ReAct框架强调“思考-行动-观察”的循环适合复杂任务。create_openai_tools_agent: 专为OpenAI的function calling优化是目前最稳定、最推荐的方式。create_structured_chat_agent: 使用结构化消息适合需要复杂参数的工具。from langchain.agents import create_openai_tools_agent, AgentExecutor from langchain.tools import Tool from langchain_openai import ChatOpenAI import datetime # 1. 定义工具 tool def get_current_time(placeholder: str) - str: 获取当前的日期和时间。placeholder参数仅为符合格式要求可忽略。 return f当前时间是{datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S)} tool def search_wikipedia(query: str) - str: 在维基百科中搜索一个主题。 # 这里简化实现实际应调用维基百科API return f关于{query}的搜索结果摘要... # 工具列表 tools [get_current_time, search_wikipedia] # 2. 创建Agent model ChatOpenAI(modelgpt-3.5-turbo, temperature0) prompt ... # 可以使用LangChain内置的OpenAI工具Agent提示模板 agent create_openai_tools_agent(model, tools, prompt) # 3. 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, max_iterations5) # 4. 运行Agent result agent_executor.invoke({input: 现在几点了然后查一下爱因斯坦的生平。}) print(result[output])注意事项verboseTrue在开发时极其有用它会打印出Agent的思考过程和工具调用详情。生产环境中记得关闭。另外务必设置max_iterations通常3-10次这是防止Agent陷入死循环的安全阀。3. 生产级代码示例构建一个可维护的智能客服助手理论讲完了我们动手搭建一个接近生产环境的示例一个具备知识库查询RAG和联网搜索能力的智能客服助手。我们将使用LCEL、自定义工具和结构化输出。3.1 项目结构与配置管理首先建立清晰的项目结构并使用pydantic-settings管理配置避免将API密钥硬编码在代码中。# config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): openai_api_key: str anthropic_api_key: Optional[str] None tavily_api_key: Optional[str] None # 用于搜索的工具 database_url: Optional[str] sqlite:///./chroma.db class Config: env_file .env settings Settings()# .env 文件 (切勿提交到版本库) OPENAI_API_KEYsk-... TAVILY_API_KEYtvly-...3.2 实现核心业务链我们将创建两条链一条用于处理基于内部知识库的问答RAG链另一条用于处理需要最新信息的通用问答搜索链。然后创建一个路由逻辑来决定使用哪条链。# chains.py from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_community.retrievers import TavilySearchAPIRetriever from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableBranch, RunnableLambda from langchain_core.output_parsers import StrOutputParser from config import settings import hashlib # 1. 初始化组件 llm ChatOpenAI(modelgpt-4-turbo-preview, api_keysettings.openai_api_key) embeddings OpenAIEmbeddings(api_keysettings.openai_api_key) # 2. 知识库检索链 (RAG Chain) def load_or_create_vectorstore(): 加载或创建向量存储。生产环境中这部分应独立为数据预处理流水线。 # 这里简化处理。实际应从持久化存储加载。 # 假设我们已经有一个处理好的Chroma集合‘company_kb’ return Chroma( collection_namecompany_kb, embedding_functionembeddings, persist_directory./chroma_db ) vectorstore load_or_create_vectorstore() kb_retriever vectorstore.as_retriever(search_typesimilarity, search_kwargs{k: 4}) rag_prompt ChatPromptTemplate.from_messages([ (system, 你是一家名为“智助科技”的AI公司的客服助手。请严格根据提供的上下文信息回答用户关于公司产品、服务或政策的问题。 上下文 {context} 如果上下文信息不足以回答问题请直接说“根据现有资料我无法回答这个问题”。不要编造信息。), (human, {question}) ]) rag_chain ( {context: kb_retriever, question: RunnablePassthrough()} | rag_prompt | llm | StrOutputParser() ) # 3. 联网搜索链 (Web Search Chain) search_retriever TavilySearchAPIRetriever(api_keysettings.tavily_api_key, k3) search_prompt ChatPromptTemplate.from_messages([ (system, 你是一个有用的助手。请根据以下的网络搜索结果用中文回答用户的问题。 搜索结果 {search_results} 请以清晰、有条理的方式总结答案并注明信息来源。如果搜索结果不相关或不足请如实告知。), (human, {question}) ]) search_chain ( {search_results: search_retriever, question: RunnablePassthrough()} | search_prompt | llm | StrOutputParser() ) # 4. 路由判断链决定用户问题属于哪一类 class RouteQuery: def __init__(self, company_keywordsNone): self.company_keywords company_keywords or [智助科技, 你们公司, 产品价格, 售后服务, 用户协议] def __call__(self, input_dict): question input_dict[question].lower() # 简单规则如果问题包含公司相关关键词走知识库否则走搜索。 # 生产环境可用一个小型分类模型来实现。 if any(keyword in question for keyword in self.company_keywords): return knowledge_base else: return web_search router RunnableLambda(RouteQuery()) # 5. 主链分支路由 main_chain RunnableBranch( (lambda x: x[topic] knowledge_base, rag_chain), (lambda x: x[topic] web_search, search_chain), rag_chain.with_fallbacks([search_chain]) # 默认先尝试知识库失败则回退到搜索 ).with_config(run_nameroute_chain) # 最终暴露的调用接口 final_chain ( {question: RunnablePassthrough()} | { topic: router, question: RunnablePassthrough() } | main_chain )3.3 添加监控、日志与缓存生产级应用必须可观测。我们集成LangSmith进行跟踪并添加缓存提升性能。# monitor.py import os from langsmith import Client from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache, SQLiteCache # 1. 设置LangSmith用于跟踪链的每一步调试和评估 os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_ENDPOINT] https://api.smith.langchain.com os.environ[LANGCHAIN_API_KEY] lsv2_... # 你的LangSmith API Key os.environ[LANGCHAIN_PROJECT] Customer-Support-Agent-Prod # 2. 设置缓存减少对LLM的重复调用和成本 # 开发时用内存缓存生产环境建议用Redis或SQLite set_llm_cache(InMemoryCache()) # 或者使用SQLiteCache持久化缓存 # set_llm_cache(SQLiteCache(database_path.langchain.db)) # 3. 在调用链时所有步骤会自动记录到LangSmith3.4 封装为可部署服务最后我们可以使用FastAPI和LangServe将链部署为HTTP API服务。# app.py (FastAPI LangServe) from fastapi import FastAPI from langserve import add_routes from chains import final_chain from monitor import * # 导入监控配置 app FastAPI( title智助科技客服助手API, version1.0.0, description一个集成了内部知识库和联网搜索的智能客服助手。 ) # 将我们的链添加为API端点 add_routes( app, final_chain, path/chat, input_typestr, # 输入是字符串问题 ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)现在运行python app.py你就拥有了一个运行在http://localhost:8000的AI助手服务。访问http://localhost:8000/docs可以看到自动生成的OpenAPI文档。4. 常见问题与排查技巧实录在实际开发和运维中你会遇到各种各样的问题。这里记录了几个最典型场景的排查思路。4.1 检索效果不佳RAG的典型痛点症状AI回答“根据现有资料无法回答”但你明明知道知识库里有相关内容。排查步骤检查分块用retriever.invoke(“你的关键词”)直接测试检索器看返回的文档片段是否包含答案。如果片段太碎或太大调整chunk_size和chunk_overlap。检查向量化确认使用的嵌入模型Embeddings是否适合你的文本领域中文、代码等。对于中文可以尝试text-embedding-3-small或bge-large-zh。检查检索策略as_retriever()默认使用向量相似度搜索。对于需要精确匹配如产品代号的情况可以尝试search_typemmr最大边际相关性兼顾相关性和多样性或使用混合检索器Hybrid Search结合关键词BM25和向量搜索。检查提示词在提示词中明确指令“必须严格根据上下文回答”并可以增加“如果上下文不相关请直接说不知道”的约束。4.2 Agent陷入循环或调用错误工具症状Agent不停地重复调用同一个工具或者调用了一个完全不相关的工具。排查步骤开启详细日志创建AgentExecutor时务必设置verboseTrue观察模型的“思考”过程。它可能误解了工具的描述。优化工具描述工具的description参数至关重要。用清晰、简洁的语言描述工具的精确用途和输入格式。例如“查询天气”比“获取信息”要好。提供少量示例在给Agent的SystemMessage中提供1-2个正确使用工具的示例Few-shot Learning能极大提升其选择工具的准确性。限制工具范围不要给Agent提供它当前任务用不到的工具。工具越多决策越困难。设置迭代上限max_iterations5是必须的这是最后的防线。4.3 响应速度慢或成本过高症状API调用延迟高或者月度账单激增。优化策略实施缓存如前面所示对LLM调用和嵌入Embedding调用实施缓存。对于频繁出现的相似问题缓存能节省90%以上的成本。使用更轻量模型在链的不同环节使用不同模型。例如用gpt-3.5-turbo做路由判断或初步处理只在最终生成答案时用gpt-4。优化提示词更精确、简短的提示词能减少令牌消耗有时还能提高响应质量。使用ChatPromptTemplate有助于管理和复用提示词。批处理请求如果有多条用户输入需要处理使用chain.batch()而不是循环调用chain.invoke()某些提供商对批处理有优化。异步调用在Web服务中使用chain.ainvoke()进行异步调用避免阻塞事件循环。4.4 依赖版本冲突与兼容性症状ImportError或运行时报错AttributeError: module ‘langchain’ has no attribute ‘...’。解决方案使用命名空间包v1.x后LangChain拆分为多个包。务必使用langchain-community,langchain-openai,langchain-anthropic等而不是直接从langchain导入。检查你的requirements.txt。锁定核心版本在pyproject.toml或requirements.txt中精确指定版本例如langchain-core0.1.0,langchain-openai0.0.5。使用poetry或pip-tools管理依赖。查阅官方迁移指南从v0.x迁移时仔细阅读官方发布的迁移指南大部分旧类都有对应的新位置和新写法。迁移到LangChain v1.x不是一次简单的版本升级而是一次开发范式的升级。它迫使你以更模块化、更声明式LCEL的方式来思考AI应用。初期可能会觉得有些繁琐但一旦适应你会发现构建、调试和维护复杂AI工作流的效率得到了质的提升。记住框架的复杂性是为了应对应用复杂性的必然选择。从理解这六大组件开始一步步构建你的生产级应用那些看似复杂的抽象最终都会成为你手中得心应手的工具。 LangChain v1.x 六大核心组件详解：从概念到生产级AI应用开发
1. 项目概述为什么我们需要重新认识LangChain如果你在过去一年里接触过AI应用开发尤其是基于大语言模型LLM的智能体Agent或聊天机器人那么“LangChain”这个名字你一定不陌生。它几乎成了连接业务逻辑与大模型的“标准中间件”。然而随着LangChain v1.x版本的全面发布整个框架经历了从“概念验证工具集”到“企业级应用框架”的深刻蜕变。很多开发者还停留在v0.x时代“链式调用”的简单认知里用着过时的API写着脆弱的代码然后抱怨LangChain“抽象过度”、“性能低下”。这其实是一个巨大的误解。我最近在将一个v0.1xx版本的旧项目迁移到v1.x时感触颇深。新版本不仅仅是API改名那么简单它重构了核心抽象明确了六大组件的职责边界并提供了大量生产级的最佳实践。简单来说v0.x像是给你一堆乐高零件和一张模糊的示意图而v1.x则提供了一套完整的建筑图纸、标准化的结构件甚至告诉你哪里该用钢筋加固。本教程的目的就是带你彻底吃透LangChain v1.x的这六大核心组件并通过可直接用于生产环境的代码示例让你避开我踩过的那些坑真正掌握构建稳健、可维护的AI应用的能力。2. 核心组件全解析从“玩具”到“工具”的思维转变LangChain v1.x 将构建LLM应用的要素清晰地归纳为六大核心组件Schema、Models、Prompts、Indexes、Chains和Agents。理解它们的关系和设计哲学是高效使用框架的关键。2.1 Schema数据交换的“普通话”在v0.x中我们经常需要手动拼接字符串来构造提示词Prompt或者费力地解析模型返回的文本。v1.x的Schema组件首先解决了这个问题它定义了与LLM交互的标准化数据结构。BaseMessage及其子类这是最核心的抽象。一条消息不再是一个简单的字符串而是一个具有明确角色的对象。HumanMessage: 代表用户输入。AIMessage: 代表AI模型的回复。SystemMessage: 代表系统指令用于设定AI的角色和行为边界。FunctionMessage/ToolMessage: 代表函数或工具调用的输入和输出。这种设计让多轮对话的管理变得清晰无比。一个对话历史ChatMessageHistory本质上就是一个BaseMessage的列表。Document对象处理外部知识如从PDF、网页爬取的内容时我们不再使用原始的文本字符串。Document对象封装了一段文本及其元数据如来源、页码、作者。这使得在检索增强生成RAG流程中能够精准地追溯答案的来源。实操心得养成使用SystemMessage的习惯。在v0.x中系统提示常被混在用户提示里。现在明确地用SystemMessage设置角色如“你是一个专业的翻译助手”能让模型更好地遵循指令也使得提示词模板更易维护。2.2 Models不仅仅是“换模型改API Key”Models组件提供了与各种LLM交互的统一接口。v1.x最大的改进在于“可预测性”和“生态集成”。ChatModelsvsLLMs现在严格区分了聊天模型如GPT-4, Claude和补全模型如text-davinci-003。对于绝大多数应用你应该使用ChatOpenAI,ChatAnthropic等ChatModels因为它们天然支持上述的BaseMessage序列。统一的输入输出无论底层是OpenAI、Anthropic、Cohere还是本地部署的Ollama、vLLM你都可以通过相同的invoke()、batch()、stream()方法来调用。输出也被标准化为AIMessage对象从中可以方便地提取内容 (message.content)、函数调用信息 (message.additional_kwargs) 等。生产级特性内置重试逻辑、速率限制、失败回退fallback等生产环境必备功能现在可以通过model ChatOpenAI(max_retries2, ...)或ChatOpenAI(fallback_to[another_model])轻松配置无需自己造轮子。# 生产级模型调用示例 from langchain_openai import ChatOpenAI from langchain_anthropic import ChatAnthropic from langchain_core.messages import HumanMessage, SystemMessage # 1. 基础调用 model ChatOpenAI(modelgpt-4-turbo-preview, temperature0) messages [ SystemMessage(content你是一位代码评审专家。), HumanMessage(content请评审这段Python函数def foo(x): return x1) ] response model.invoke(messages) print(response.content) # 2. 配置流式输出用于Web应用 for chunk in model.stream(messages): print(chunk.content, end, flushTrue) # 3. 配置故障转移Fallback primary_model ChatOpenAI(modelgpt-4, max_retries1) fallback_model ChatAnthropic(modelclaude-3-haiku-20240307) # 在实际项目中可以通过LCEL后文会讲将多个模型组合成带fallback的runnable2.3 Prompts告别“字符串地狱”在v0.x中构建复杂的提示模板很容易变成一堆令人头疼的f-string或.format()。v1.x的Prompts组件通过PromptTemplate和ChatPromptTemplate将其系统化。ChatPromptTemplate这是构建聊天提示的推荐方式。它由多个MessagePromptTemplate组成每个对应一种角色。你可以像搭积木一样组合系统提示、少量示例few-shot和用户输入。模板化与部分绑定你可以提前创建模板并在运行时动态注入变量。更强大的是“部分绑定”partial允许你预先填充模板中的部分变量如用户的个人资料在调用链时再传入剩余变量。from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate # 构建一个可复用的RAG提示模板 system_template SystemMessagePromptTemplate.from_template( 你是一个乐于助人的助手。请根据以下上下文回答问题。如果你不知道答案就说不知道。\n\n上下文{context} ) human_template HumanMessagePromptTemplate.from_template(问题{question}) chat_prompt ChatPromptTemplate.from_messages([system_template, human_template]) # 使用模板 formatted_messages chat_prompt.invoke({ context: LangChain是一个用于开发LLM应用的框架。, question: LangChain是什么 }) # formatted_messages 现在是一个准备好发送给ChatModel的BaseMessage列表 # 部分绑定示例假设我们有一个固定的系统角色 from langchain_core.prompts import PromptTemplate prompt PromptTemplate.from_template(以{style}的风格写一篇关于{topic}的文章。) poetic_prompt prompt.partial(style诗歌体) # 现在 poetic_prompt 只需要一个 topic 变量 print(poetic_prompt.invoke({topic: 春天}).to_string())2.4 Indexes构建你的“外部大脑”Indexes组件关乎如何让LLM访问它训练数据之外的信息即检索增强生成RAG的核心。v1.x将其拆解得更加模块化核心是Retriever检索器接口。理解数据流原始文本 - 加载器Loader - 文档Document - 分割器Splitter - 向量化Embedding - 向量存储VectorStore - 检索器Retriever。检索器Retriever这是一个关键抽象。它定义了一个get_relevant_documents(query)方法。任何实现了该接口的对象都可以作为检索器无论是基于向量的、基于关键词的如TF-IDF还是混合检索器。向量存储生态LangChain集成了Chroma、Pinecone、Weaviate、Qdrant等几乎所有主流向量数据库。v1.x的集成更稳定API更一致。# 一个完整的从文本加载到检索的示例 from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.retrievers import BaseRetriever # 1. 加载与分割文档 loader TextLoader(./state_of_the_union.txt) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size1000, chunk_overlap200) splits text_splitter.split_documents(documents) # 2. 创建向量存储并转换为检索器 embeddings OpenAIEmbeddings() vectorstore Chroma.from_documents(documentssplits, embeddingembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 4}) # 检索最相关的4个片段 # 3. 使用检索器 docs retriever.invoke(总统提到了哪些关于能源的政策) for doc in docs: print(doc.page_content[:200], ...)避坑指南分块Chunking是RAG效果的“隐形杀手”。chunk_size不是越大或越小越好。对于通用文档500-1500是个不错的起点。chunk_overlap设置重叠如200字能防止关键信息被割裂。对于代码、论文等特殊格式需要使用专门的分割器如MarkdownHeaderTextSplitter。2.5 Chains用LCEL构建健壮的工作流这是v1.x革新最大的部分。旧的LLMChain、SequentialChain等被一个更强大、更灵活的概念取代LangChain Expression Language (LCEL)和Runnable 协议。核心理念一切皆可运行Everything is a Runnable。ChatModel、PromptTemplate、Retriever、甚至一个自定义的Python函数只要实现了invoke()、batch()、stream()等方法都是Runnable。你可以用|操作符像连接管道一样将它们组合起来。LCEL的优势自动流式支持用LCEL编写的链天然支持流式输出无需额外代码。并行与批量自动处理组件的并行执行和批量输入。无缝集成轻松添加日志、监控、重试、回退等中间件。易于部署LCEL链可以一键导出为LangServe API或LangSmith跟踪项。from langchain_core.runnables import RunnablePassthrough from langchain_core.output_parsers import StrOutputParser # 使用LCEL构建一个完整的RAG链 # 1. 定义各个组件假设retriever已定义 retriever ... # 来自上一节的检索器 model ChatOpenAI(modelgpt-3.5-turbo) prompt chat_prompt # 来自上一节的提示模板 output_parser StrOutputParser() # 将AIMessage解析为字符串 # 2. 组装链 rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | model | output_parser ) # 3. 调用链 result rag_chain.invoke(LangChain是什么) print(result) # 4. 流式调用 for chunk in rag_chain.stream(LangChain是什么): print(chunk, end, flushTrue)这段代码清晰地展示了数据流用户问题 - 检索器获取上下文 - 组合成提示词 - 发送给模型 - 解析输出。RunnablePassthrough()用于直接传递输入中的某个字段这里是question。2.6 Agents让LLM学会使用工具Agents是让LLM根据目标动态决定调用哪些工具Tools的组件。v1.x重新设计了Agent的执行循环使其更可靠、更易调试。核心概念工具Tool一个可供Agent调用的函数如搜索、计算、查询数据库。使用tool装饰器可以轻松将普通函数转化为Tool。AgentExecutor这是Agent的运行时引擎。它负责管理Agent大脑和工具手脚之间的交互循环处理错误并强制最大迭代次数以防止无限循环。关键选择Agent类型v1.x提供了多种预设的Agent类型对应不同的提示策略和推理逻辑。create_react_agent: 基于ReAct框架强调“思考-行动-观察”的循环适合复杂任务。create_openai_tools_agent: 专为OpenAI的function calling优化是目前最稳定、最推荐的方式。create_structured_chat_agent: 使用结构化消息适合需要复杂参数的工具。from langchain.agents import create_openai_tools_agent, AgentExecutor from langchain.tools import Tool from langchain_openai import ChatOpenAI import datetime # 1. 定义工具 tool def get_current_time(placeholder: str) - str: 获取当前的日期和时间。placeholder参数仅为符合格式要求可忽略。 return f当前时间是{datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S)} tool def search_wikipedia(query: str) - str: 在维基百科中搜索一个主题。 # 这里简化实现实际应调用维基百科API return f关于{query}的搜索结果摘要... # 工具列表 tools [get_current_time, search_wikipedia] # 2. 创建Agent model ChatOpenAI(modelgpt-3.5-turbo, temperature0) prompt ... # 可以使用LangChain内置的OpenAI工具Agent提示模板 agent create_openai_tools_agent(model, tools, prompt) # 3. 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, max_iterations5) # 4. 运行Agent result agent_executor.invoke({input: 现在几点了然后查一下爱因斯坦的生平。}) print(result[output])注意事项verboseTrue在开发时极其有用它会打印出Agent的思考过程和工具调用详情。生产环境中记得关闭。另外务必设置max_iterations通常3-10次这是防止Agent陷入死循环的安全阀。3. 生产级代码示例构建一个可维护的智能客服助手理论讲完了我们动手搭建一个接近生产环境的示例一个具备知识库查询RAG和联网搜索能力的智能客服助手。我们将使用LCEL、自定义工具和结构化输出。3.1 项目结构与配置管理首先建立清晰的项目结构并使用pydantic-settings管理配置避免将API密钥硬编码在代码中。# config.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): openai_api_key: str anthropic_api_key: Optional[str] None tavily_api_key: Optional[str] None # 用于搜索的工具 database_url: Optional[str] sqlite:///./chroma.db class Config: env_file .env settings Settings()# .env 文件 (切勿提交到版本库) OPENAI_API_KEYsk-... TAVILY_API_KEYtvly-...3.2 实现核心业务链我们将创建两条链一条用于处理基于内部知识库的问答RAG链另一条用于处理需要最新信息的通用问答搜索链。然后创建一个路由逻辑来决定使用哪条链。# chains.py from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_community.retrievers import TavilySearchAPIRetriever from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableBranch, RunnableLambda from langchain_core.output_parsers import StrOutputParser from config import settings import hashlib # 1. 初始化组件 llm ChatOpenAI(modelgpt-4-turbo-preview, api_keysettings.openai_api_key) embeddings OpenAIEmbeddings(api_keysettings.openai_api_key) # 2. 知识库检索链 (RAG Chain) def load_or_create_vectorstore(): 加载或创建向量存储。生产环境中这部分应独立为数据预处理流水线。 # 这里简化处理。实际应从持久化存储加载。 # 假设我们已经有一个处理好的Chroma集合‘company_kb’ return Chroma( collection_namecompany_kb, embedding_functionembeddings, persist_directory./chroma_db ) vectorstore load_or_create_vectorstore() kb_retriever vectorstore.as_retriever(search_typesimilarity, search_kwargs{k: 4}) rag_prompt ChatPromptTemplate.from_messages([ (system, 你是一家名为“智助科技”的AI公司的客服助手。请严格根据提供的上下文信息回答用户关于公司产品、服务或政策的问题。 上下文 {context} 如果上下文信息不足以回答问题请直接说“根据现有资料我无法回答这个问题”。不要编造信息。), (human, {question}) ]) rag_chain ( {context: kb_retriever, question: RunnablePassthrough()} | rag_prompt | llm | StrOutputParser() ) # 3. 联网搜索链 (Web Search Chain) search_retriever TavilySearchAPIRetriever(api_keysettings.tavily_api_key, k3) search_prompt ChatPromptTemplate.from_messages([ (system, 你是一个有用的助手。请根据以下的网络搜索结果用中文回答用户的问题。 搜索结果 {search_results} 请以清晰、有条理的方式总结答案并注明信息来源。如果搜索结果不相关或不足请如实告知。), (human, {question}) ]) search_chain ( {search_results: search_retriever, question: RunnablePassthrough()} | search_prompt | llm | StrOutputParser() ) # 4. 路由判断链决定用户问题属于哪一类 class RouteQuery: def __init__(self, company_keywordsNone): self.company_keywords company_keywords or [智助科技, 你们公司, 产品价格, 售后服务, 用户协议] def __call__(self, input_dict): question input_dict[question].lower() # 简单规则如果问题包含公司相关关键词走知识库否则走搜索。 # 生产环境可用一个小型分类模型来实现。 if any(keyword in question for keyword in self.company_keywords): return knowledge_base else: return web_search router RunnableLambda(RouteQuery()) # 5. 主链分支路由 main_chain RunnableBranch( (lambda x: x[topic] knowledge_base, rag_chain), (lambda x: x[topic] web_search, search_chain), rag_chain.with_fallbacks([search_chain]) # 默认先尝试知识库失败则回退到搜索 ).with_config(run_nameroute_chain) # 最终暴露的调用接口 final_chain ( {question: RunnablePassthrough()} | { topic: router, question: RunnablePassthrough() } | main_chain )3.3 添加监控、日志与缓存生产级应用必须可观测。我们集成LangSmith进行跟踪并添加缓存提升性能。# monitor.py import os from langsmith import Client from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache, SQLiteCache # 1. 设置LangSmith用于跟踪链的每一步调试和评估 os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_ENDPOINT] https://api.smith.langchain.com os.environ[LANGCHAIN_API_KEY] lsv2_... # 你的LangSmith API Key os.environ[LANGCHAIN_PROJECT] Customer-Support-Agent-Prod # 2. 设置缓存减少对LLM的重复调用和成本 # 开发时用内存缓存生产环境建议用Redis或SQLite set_llm_cache(InMemoryCache()) # 或者使用SQLiteCache持久化缓存 # set_llm_cache(SQLiteCache(database_path.langchain.db)) # 3. 在调用链时所有步骤会自动记录到LangSmith3.4 封装为可部署服务最后我们可以使用FastAPI和LangServe将链部署为HTTP API服务。# app.py (FastAPI LangServe) from fastapi import FastAPI from langserve import add_routes from chains import final_chain from monitor import * # 导入监控配置 app FastAPI( title智助科技客服助手API, version1.0.0, description一个集成了内部知识库和联网搜索的智能客服助手。 ) # 将我们的链添加为API端点 add_routes( app, final_chain, path/chat, input_typestr, # 输入是字符串问题 ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)现在运行python app.py你就拥有了一个运行在http://localhost:8000的AI助手服务。访问http://localhost:8000/docs可以看到自动生成的OpenAPI文档。4. 常见问题与排查技巧实录在实际开发和运维中你会遇到各种各样的问题。这里记录了几个最典型场景的排查思路。4.1 检索效果不佳RAG的典型痛点症状AI回答“根据现有资料无法回答”但你明明知道知识库里有相关内容。排查步骤检查分块用retriever.invoke(“你的关键词”)直接测试检索器看返回的文档片段是否包含答案。如果片段太碎或太大调整chunk_size和chunk_overlap。检查向量化确认使用的嵌入模型Embeddings是否适合你的文本领域中文、代码等。对于中文可以尝试text-embedding-3-small或bge-large-zh。检查检索策略as_retriever()默认使用向量相似度搜索。对于需要精确匹配如产品代号的情况可以尝试search_typemmr最大边际相关性兼顾相关性和多样性或使用混合检索器Hybrid Search结合关键词BM25和向量搜索。检查提示词在提示词中明确指令“必须严格根据上下文回答”并可以增加“如果上下文不相关请直接说不知道”的约束。4.2 Agent陷入循环或调用错误工具症状Agent不停地重复调用同一个工具或者调用了一个完全不相关的工具。排查步骤开启详细日志创建AgentExecutor时务必设置verboseTrue观察模型的“思考”过程。它可能误解了工具的描述。优化工具描述工具的description参数至关重要。用清晰、简洁的语言描述工具的精确用途和输入格式。例如“查询天气”比“获取信息”要好。提供少量示例在给Agent的SystemMessage中提供1-2个正确使用工具的示例Few-shot Learning能极大提升其选择工具的准确性。限制工具范围不要给Agent提供它当前任务用不到的工具。工具越多决策越困难。设置迭代上限max_iterations5是必须的这是最后的防线。4.3 响应速度慢或成本过高症状API调用延迟高或者月度账单激增。优化策略实施缓存如前面所示对LLM调用和嵌入Embedding调用实施缓存。对于频繁出现的相似问题缓存能节省90%以上的成本。使用更轻量模型在链的不同环节使用不同模型。例如用gpt-3.5-turbo做路由判断或初步处理只在最终生成答案时用gpt-4。优化提示词更精确、简短的提示词能减少令牌消耗有时还能提高响应质量。使用ChatPromptTemplate有助于管理和复用提示词。批处理请求如果有多条用户输入需要处理使用chain.batch()而不是循环调用chain.invoke()某些提供商对批处理有优化。异步调用在Web服务中使用chain.ainvoke()进行异步调用避免阻塞事件循环。4.4 依赖版本冲突与兼容性症状ImportError或运行时报错AttributeError: module ‘langchain’ has no attribute ‘...’。解决方案使用命名空间包v1.x后LangChain拆分为多个包。务必使用langchain-community,langchain-openai,langchain-anthropic等而不是直接从langchain导入。检查你的requirements.txt。锁定核心版本在pyproject.toml或requirements.txt中精确指定版本例如langchain-core0.1.0,langchain-openai0.0.5。使用poetry或pip-tools管理依赖。查阅官方迁移指南从v0.x迁移时仔细阅读官方发布的迁移指南大部分旧类都有对应的新位置和新写法。迁移到LangChain v1.x不是一次简单的版本升级而是一次开发范式的升级。它迫使你以更模块化、更声明式LCEL的方式来思考AI应用。初期可能会觉得有些繁琐但一旦适应你会发现构建、调试和维护复杂AI工作流的效率得到了质的提升。记住框架的复杂性是为了应对应用复杂性的必然选择。从理解这六大组件开始一步步构建你的生产级应用那些看似复杂的抽象最终都会成为你手中得心应手的工具。