LangChain 应用开发实战LCEL 声明式编排与完整问答系统搭建LangChain 早期版本被很多人诟病抽象太多、心智负担重但 2024 年之后框架经历了关键转型以 LCELLangChain Expression Language为核心重写了组件组合方式把链从黑盒对象变成了透明、可追踪、可并行的数据管道。这篇文章用实战方式讲清楚 LCEL 的底层逻辑并在此基础上搭一个带向量检索的企业知识问答系统覆盖从环境准备到流式输出的完整过程。一、先理解 LCEL 到底解决了什么问题传统的链式写法是把组件塞进一个 Chain 对象内部逻辑不可见调试靠猜。LCEL 的做法完全不同它用管道运算符|把提示词模板 → 模型 → 输出解析器串成一条数据流上一个组件的输出自动成为下一个组件的输入。fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAI promptChatPromptTemplate.from_messages([(system,你是一个严谨的技术写作助手擅长用通俗语言解释复杂概念。),(user,请解释{topic}),])llmChatOpenAI(modelgpt-4o-mini,temperature0.3)chainprompt|llm resultchain.invoke({topic:什么是 LCEL 声明式管道})print(result.content)这条链的本质是一个可调用对象输入一个字典输出一个模型消息。|符号背后是一套统一的 Runnable 协议——任何实现了该协议的对象提示词模板、模型、解析器、自定义函数都能互相串联这给组合带来了前所未有的自由度。二、Runnable 协议LCEL 的基石LCEL 的核心不是管道符号而是它背后统一的 Runnable 接口。每个 Runnable 都实现了invoke同步调用、batch批量调用、stream流式调用、ainvoke异步调用等方法。这意味着你写出的任何一段链天然就支持批处理和流式输出不需要额外代码。# 流式输出用户打字机效果forchunkinchain.stream({topic:Runnable 协议的价值}):print(chunk.content,end,flushTrue) python# 批量调用批量处理自动获得并发resultschain.batch([{topic:LCEL},{topic:RAG},{topic:Agent}])流式输出对用户体验的提升是立竿见影的——首字延迟从等完整回答变成秒回。批量调用则直接提升了吞吐尤其适合离线批处理场景。除了组合LCEL 还支持条件路由。用RunnableBranch可以根据输入内容把请求分发给不同处理路径比如技术问题走技术链非技术问题走通用链fromlangchain_core.runnablesimportRunnableBranch classifyChatPromptTemplate.from_template(判断下面的问题是否与技术相关只回答 yes 或 no\n{question})|llm|(lambdam:m.content.strip().lower())branchRunnableBranch((lambdax:yesinx[judge],tech_chain),(lambdax:True,general_chain),)router{question:lambdax:x[question],judge:classify,}|branch 条件路由的价值在于不同质量要求、不同成本模型的场景可以共用一套入口由路由决定走向这让整体架构的扩展性大大提升。## 三、实战企业知识库问答系统理论讲完进入正题。我们要搭建的系统解决一个经典问题让员工用自然语言查询企业内部文档模型只基于检索到的资料回答杜绝凭空编造。系统分四步文档加载、切片入库、向量检索、生成回答。### 3.1 文档加载与切片先把文档读进来切成适合检索的片段。切片的粒度直接决定召回质量——太粗会混入无关信息太细会丢失上下文。 pythonfromlangchain_community.document_loadersimportPyPDFLoader,TextLoaderfromlangchain_text_splittersimportRecursiveCharacterTextSplitter loaderPyPDFLoader(employee_handbook.pdf)docsloader.load()splitterRecursiveCharacterTextSplitter(chunk_size800,# 每片约 800 字符chunk_overlap120,# 相邻片保留 120 字符重叠separators[\n\n,\n,。,,,, ],)chunkssplitter.split_documents(docs)print(f共切出{len(chunks)}个片段) RecursiveCharacterTextSplitter 的聪明之处在于它按优先级尝试分隔符先按段落切切不动再按句子切再按词切。chunk_overlap 保证跨片边界的信息不丢失。这两个参数的调优没有银弹需要在真实语料上反复验证后面我们会讲评估方法。### 3.2 向量化与入库把每个片段转成向量存入向量数据库。这里以 FAISS 为例本地轻量生产环境可以换成 Milvus、Qdrant 等分布式方案。 pythonfromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportFAISS embeddingsOpenAIEmbeddings(modeltext-embedding-3-small)vectorstoreFAISS.from_documents(chunks,embeddings)# 保存与加载避免重复向量化vectorstore.save_local(./faiss_index)vectorstoreFAISS.load_local(./faiss_index,embeddings,allow_dangerous_deserializationTrue)向量化的细节决定了检索下限嵌入模型要选支持中文的如 text-embedding-3、bge 系列生产环境建议把向量库和文档库解耦文档更新时增量重建向量而不是每次全量跑。3.3 检索增强生成检索环节用as_retriever把向量库包装成检索器配合一个文档压缩步骤——只把与问题最相关的片段拼进上下文避免塞入噪音。fromlangchain.chains.combine_documentsimportcreate_retrieval_chainfromlangchain.chainsimportcreate_history_aware_retrieverfromlangchain_core.chat_historyimportBaseChatMessageHistory retrievervectorstore.as_retriever(search_kwargs{k:4})qa_promptChatPromptTemplate.from_messages([(system,你是一个企业知识库问答助手。只能根据下面提供的资料回答 如果资料中没有相关信息明确回答资料库中未找到相关内容不要自行编造。 回答时注明信息来自哪些资料片段。),(human,资料\n{context}\n\n问题{input}),])defformat_docs(docs):return\n\n---\n\n.join(f【片段{i1}】{d.page_content}fori,dinenumerate(docs))rag_chain({context:retriever|format_docs,input:lambdax:x[question]}|qa_prompt|llm)answerrag_chain.invoke({question:年假申请需要提前几天})print(answer.content)这里有几个值得展开的设计点。第一retriever | format_docs让检索结果在进入提示词之前先被格式化逻辑清晰且可单独测试。第二system prompt 里明确写了没有就直说这是 RAG 系统对抗幻觉的第一道防线。第三回答要求注明片段来源既方便用户核验也为后续的评测留了抓手。3.4 多轮对话的记忆处理真实场景里用户会连续追问“那病假呢”“最多能请几天”。这要求系统把历史对话转成独立于历史的检索问题——先让模型把用户当前问题结合上下文改写成一条完整查询再拿去检索检索结果和完整对话历史一起交给模型生成回答。这就是create_history_aware_retriever在做的事。fromlangchain_core.messagesimportHumanMessagedefcondense_question(history,question):ifnothistory:returnquestion condense_promptChatPromptTemplate.from_template(根据对话历史把用户的最新问题改写成一条可独立检索的完整问题。\n历史{history}\n最新问题{question}\n只输出改写后的问题。)return(condense_prompt|llm).invoke({history:history,question:question}).content history[HumanMessage(content年假申请需要提前几天)]qcondense_question(history,那最多能请几天呢)# 输出类似企业年假单次最多可申请的天数是多少改写这一步质量差后续检索必然跟着差所以值得单独建评测用例去验证。四、评测与调优让检索质量可度量知识问答系统上线前必须回答一个问题检索准不准回答对不对用人工逐条看太慢推荐两段式自动评测检索质量准备一批问题 → 应命中片段的标注数据计算 RecallK——前 K 个检索结果里包含标准片段的比例。回答质量用 LLM-as-Judge 打分重点检查回答是否基于给定资料和是否遗漏关键信息。defrecall_at_k(retriever,questions,golden,k4):hit0forq,goldinzip(questions,golden):docsretriever.invoke(q)[:k]texts[d.page_contentfordindocs]ifany(gintfortintexts):hit1returnhit/len(questions) 有了度量调参就不再靠感觉chunk_size 从400试到1200overlap 从50试到200k 从3试到6每次改完跑一遍 Recall 和 Judge 分数留下最优组合。这套流程才是 RAG 工程真正的核心工作量所在。## 五、进阶RunnableParallel 与流式组合最后补充一个生产高频用法。检索、重写等耗时操作可以用 RunnableParallel 并行执行让整体延迟从串行之和变成最长路径 pythonfromlangchain_core.runnablesimportRunnableParallel parallelRunnableParallel(retrievedretriever,condensedcondense_chain,)resultparallel.invoke({question:年假能拆成半天申请吗}) 配合 stream你甚至可以在检索还没完成时就先把正在检索资料…的占位符流给用户体验层面的优化空间非常大。## 六、小结LCEL 的真正价值在于把编排变成了数据流的自然表达组件可组合、链路可追踪、并行和流式开箱即用。基于它搭建 RAG 系统从文档加载到检索增强生成的每一环都可以独立测试、独立优化。记住两条主线**检索质量靠切片与召回参数调优回答质量靠提示词约束与评测闭环**。把这两条主线跑通你手上就是一个真正能上生产的知识问答系统。