LangChain 是当前最主流的 LLM 应用开发框架。本文不讲基础入门直接聚焦生产级开发中的关键工程实践包括性能优化、错误处理、可观测性和成本控制。一、LangChain 核心抽象体系理解 LangChain 的架构对于高效使用至关重要LangChain 核心抽象├── Models模型│ ├── LLMs文本补全│ └── ChatModels对话├── Prompts提示│ ├── PromptTemplate│ └── ChatPromptTemplate├── Chains链│ ├── LLMChain简单链│ ├── SequentialChain顺序链│ └── LCEL新一代声明式接口├── Retrievers检索器│ └── VectorStoreRetriever 等├── Agents智能体│ ├── ReAct Agent│ └── OpenAI Functions Agent└── Memory记忆 └── 对话历史管理—## 二、LCELLangChain 表达式语言生产推荐写法LCEL 是 LangChain 的现代化接口相比旧版 Chain 有显著优势pythonfrom langchain_openai import ChatOpenAIfrom langchain.prompts import ChatPromptTemplatefrom langchain.schema.output_parser import StrOutputParser# LCEL 声明式链构建用管道符 | 连接prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的{domain}专家), (human, {question})])model ChatOpenAI(modelgpt-4o, temperature0)output_parser StrOutputParser()# 链式组合chain prompt | model | output_parser# 同步调用result chain.invoke({domain: 数据科学, question: 解释什么是梯度下降})# 流式输出for chunk in chain.stream({domain: 数据科学, question: 解释什么是梯度下降}): print(chunk, end, flushTrue)# 并发批处理results chain.batch([ {domain: 法律, question: 合同无效的情形有哪些}, {domain: 医学, question: 高血压的危险因素是什么},])LCEL 的核心优势- 支持流式输出- 支持异步/批量操作- 内置 LangSmith 追踪集成- 声明式语法可读性强—## 三、生产级 RAG 实现### 完整 RAG Pipelinepythonfrom langchain_openai import OpenAIEmbeddings, ChatOpenAIfrom langchain_community.vectorstores import Qdrantfrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain.schema import Documentfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthrough# 1. 文档处理def load_and_split_documents(file_paths: list[str]) - list[Document]: splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, separators[\n\n, \n, 。, , , ] ) docs [] for path in file_paths: # 根据文件类型使用不同 Loader loader get_loader(path) docs.extend(loader.load()) return splitter.split_documents(docs)# 2. 建立向量数据库embeddings OpenAIEmbeddings(modeltext-embedding-3-small)chunks load_and_split_documents([./docs/knowledge_base.pdf])vectorstore Qdrant.from_documents( documentschunks, embeddingembeddings, urlhttp://localhost:6333, collection_nameknowledge_base)# 3. 构建 RAG Chainretriever vectorstore.as_retriever( search_typemmr, # 最大边际相关性减少重复 search_kwargs{k: 5, fetch_k: 20})rag_prompt ChatPromptTemplate.from_template(你是一个专业的知识库助手。请基于以下检索到的文档内容回答用户问题。如果文档内容不足以回答问题请明确说明根据当前知识库无法回答此问题不要猜测。检索文档{context}用户问题{question}回答)llm ChatOpenAI(modelgpt-4o, temperature0)def format_docs(docs): return \n\n.join([ f[来源{doc.metadata.get(source, 未知)}]\n{doc.page_content} for doc in docs ])# LCEL RAG Chainrag_chain ( {context: retriever | format_docs, question: RunnablePassthrough()} | rag_prompt | llm | StrOutputParser())# 调用answer rag_chain.invoke(如何申请公司差旅报销)—## 四、对话历史管理多轮对话是实际应用的标配需要正确管理对话历史pythonfrom langchain.memory import ConversationSummaryBufferMemoryfrom langchain_core.chat_history import BaseChatMessageHistoryfrom langchain_core.runnables.history import RunnableWithMessageHistoryfrom langchain_community.chat_message_histories import RedisChatMessageHistory# 生产环境使用 Redis 持久化对话历史def get_session_history(session_id: str) - BaseChatMessageHistory: return RedisChatMessageHistory( session_idsession_id, urlredis://localhost:6379 )# 创建带历史记录的 Chainprompt_with_history ChatPromptTemplate.from_messages([ (system, 你是一个专业的客服助手), (placeholder, {chat_history}), # 历史记录插入位置 (human, {question})])base_chain prompt_with_history | llm | StrOutputParser()chain_with_history RunnableWithMessageHistory( base_chain, get_session_history, input_messages_keyquestion, history_messages_keychat_history)# 使用 session_id 区分不同用户response chain_with_history.invoke( {question: 你好我想咨询退款政策}, config{configurable: {session_id: user_123}})—## 五、错误处理与容错机制生产环境必须处理各种异常pythonfrom langchain_core.runnables import RunnableWithFallbacksfrom langchain_openai import ChatOpenAIfrom tenacity import retry, stop_after_attempt, wait_exponential# 方案一Fallback 降级链primary_llm ChatOpenAI(modelgpt-4o)fallback_llm ChatOpenAI(modelgpt-4o-mini) # 降级到更便宜的模型chain_with_fallback (prompt | primary_llm).with_fallbacks( [prompt | fallback_llm])# 方案二自定义重试逻辑retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max30))async def invoke_with_retry(chain, inputs): try: return await chain.ainvoke(inputs) except Exception as e: logger.warning(f调用失败准备重试: {e}) raise# 方案三超时控制import asyncioasync def invoke_with_timeout(chain, inputs, timeout30): try: return await asyncio.wait_for( chain.ainvoke(inputs), timeouttimeout ) except asyncio.TimeoutError: return 处理超时请稍后重试—## 六、可观测性LangSmith 集成生产环境必须能追踪每次 LLM 调用pythonimport os# 开启 LangSmith 追踪仅需设置环境变量os.environ[LANGCHAIN_TRACING_V2] trueos.environ[LANGCHAIN_API_KEY] your-api-keyos.environ[LANGCHAIN_PROJECT] production-rag# 之后所有 LangChain 调用都会自动追踪包括# - 输入/输出# - 延迟时间# - Token 消耗# - 错误信息# - 子链调用关系自定义追踪标签pythonfrom langchain.callbacks.tracers import LangChainTracertracer LangChainTracer(project_namemy-project)result chain.invoke( {question: user query}, config{ callbacks: [tracer], metadata: { user_id: user_123, feature: customer_service } })—## 七、成本控制LLM API 成本管理是生产级应用的必要考量pythonfrom langchain_community.callbacks import get_openai_callback# 精确统计每次调用的 Token 消耗with get_openai_callback() as cb: result chain.invoke({question: your query}) print(fPrompt Tokens: {cb.prompt_tokens}) print(fCompletion Tokens: {cb.completion_tokens}) print(fTotal Cost: ${cb.total_cost:.4f})成本优化策略| 策略 | 实现方式 | 节省幅度 ||------|----------|----------|| 模型降级 | 简单任务用 GPT-4o-mini | 60-90% || 输出缓存 | 相同查询复用结果 | 20-40% || Prompt 压缩 | 精简系统提示词 | 10-30% || 批处理 | 合并多个小请求 | 15-25% || 流量分层 | 高价值用户用高端模型 | 整体 30-50% |—## 八、总结LangChain 生产级开发的核心要点1.用 LCEL 代替旧版 Chain更简洁更灵活流式原生支持2.会话历史用 Redis 持久化不要用内存存储3.必须有 Fallback 机制LLM 服务不是100%可靠的4.接入 LangSmith没有可观测性无法优化5.成本意识Token 消耗要定期审计和优化—标签LangChain | LCEL | RAG | LangSmith | 生产部署 | LLM应用开发