企业官网建设流程全解析

LangChain 应用开发实战：LCEL 声明式编排与完整问答系统搭建

LangChain 早期版本被很多人诟病"抽象太多、心智负担重"，但 2024 年之后框架经历了关键转型：以 LCEL（LangChain Expression Language）为核心重写了组件组合方式，把"链"从黑盒对象变成了透明、可追踪、可并行的数据管道。这篇文章用实战方式讲清楚 LCEL 的底层逻辑，并在此基础上搭一个带向量检索的企业知识问答系统，覆盖从环境准备到流式输出的完整过程。

一、先理解 LCEL 到底解决了什么问题

传统的链式写法是把组件塞进一个 Chain 对象，内部逻辑不可见，调试靠猜。LCEL 的做法完全不同：它用管道运算符|把"提示词模板 → 模型 → 输出解析器"串成一条数据流，上一个组件的输出自动成为下一个组件的输入。

fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAI prompt=ChatPromptTemplate.from_messages([("system","你是一个严谨的技术写作助手，擅长用通俗语言解释复杂概念。"),("user","请解释：{topic}"),])llm=ChatOpenAI(model="gpt-4o-mini",temperature=0.3)chain=prompt|llm result=chain.invoke({"topic":"什么是 LCEL 声明式管道"})print(result.content)

这条链的本质是一个可调用对象：输入一个字典，输出一个模型消息。|符号背后是一套统一的 Runnable 协议——任何实现了该协议的对象（提示词模板、模型、解析器、自定义函数）都能互相串联，这给组合带来了前所未有的自由度。

二、Runnable 协议：LCEL 的基石

LCEL 的核心不是管道符号，而是它背后统一的 Runnable 接口。每个 Runnable 都实现了invoke（同步调用）、batch（批量调用）、stream（流式调用）、ainvoke（异步调用）等方法。这意味着你写出的任何一段链，天然就支持批处理和流式输出，不需要额外代码。

# 流式输出：用户打字机效果forchunkinchain.stream({"topic":"Runnable 协议的价值"}):print(chunk.content,end="",flush=True)``` ```python# 批量调用：批量处理自动获得并发results=chain.batch([{"topic":"LCEL"},{"topic":"RAG"},{"topic":"Agent"}])

流式输出对用户体验的提升是立竿见影的——首字延迟从"等完整回答"变成"秒回"。批量调用则直接提升了吞吐，尤其适合离线批处理场景。

除了组合，LCEL 还支持条件路由。用RunnableBranch可以根据输入内容把请求分发给不同处理路径，比如"技术问题走技术链，非技术问题走通用链"：

fromlangchain_core.runnablesimportRunnableBranch classify=ChatPromptTemplate.from_template("判断下面的问题是否与技术相关，只回答 yes 或 no：\n{question}")|llm|(lambdam:m.content.strip().lower())branch=RunnableBranch((lambdax:"yes"inx["judge"],tech_chain),(lambdax:True,general_chain),)router={"question":lambdax:x["question"],"judge":classify,}|branch ``` 条件路由的价值在于：不同质量要求、不同成本模型的场景可以共用一套入口，由路由决定走向，这让整体架构的扩展性大大提升。## 三、实战：企业知识库问答系统理论讲完，进入正题。我们要搭建的系统解决一个经典问题：让员工用自然语言查询企业内部文档，模型只基于检索到的资料回答，杜绝凭空编造。系统分四步：文档加载、切片入库、向量检索、生成回答。### 3.1 文档加载与切片先把文档读进来，切成适合检索的片段。切片的粒度直接决定召回质量——太粗会混入无关信息，太细会丢失上下文。 ```pythonfromlangchain_community.document_loadersimportPyPDFLoader,TextLoaderfromlangchain_text_splittersimportRecursiveCharacterTextSplitter loader=PyPDFLoader("employee_handbook.pdf")docs=loader.load()splitter=RecursiveCharacterTextSplitter(chunk_size=800,# 每片约 800 字符chunk_overlap=120,# 相邻片保留 120 字符重叠separators=["\n\n","\n","。","！","？","；"," "],)chunks=splitter.split_documents(docs)print(f"共切出{len(chunks)}个片段")``` `RecursiveCharacterTextSplitter` 的聪明之处在于它按优先级尝试分隔符：先按段落切，切不动再按句子切，再按词切。`chunk_overlap` 保证跨片边界的信息不丢失。这两个参数的调优没有银弹，需要在真实语料上反复验证，后面我们会讲评估方法。### 3.2 向量化与入库把每个片段转成向量，存入向量数据库。这里以 FAISS 为例（本地轻量），生产环境可以换成 Milvus、Qdrant 等分布式方案。 ```pythonfromlangchain_openaiimportOpenAIEmbeddingsfromlangchain_community.vectorstoresimportFAISS embeddings=OpenAIEmbeddings(model="text-embedding-3-small")vectorstore=FAISS.from_documents(chunks,embeddings)# 保存与加载，避免重复向量化vectorstore.save_local("./faiss_index")vectorstore=FAISS.load_local("./faiss_index",embeddings,allow_dangerous_deserialization=True)

向量化的细节决定了检索下限：嵌入模型要选支持中文的（如 text-embedding-3、bge 系列）；生产环境建议把向量库和文档库解耦，文档更新时增量重建向量，而不是每次全量跑。

3.3 检索增强生成

检索环节用as_retriever把向量库包装成检索器，配合一个"文档压缩"步骤——只把与问题最相关的片段拼进上下文，避免塞入噪音。

fromlangchain.chains.combine_documentsimportcreate_retrieval_chainfromlangchain.chainsimportcreate_history_aware_retrieverfromlangchain_core.chat_historyimportBaseChatMessageHistory retriever=vectorstore.as_retriever(search_kwargs={"k":4})qa_prompt=ChatPromptTemplate.from_messages([("system","""你是一个企业知识库问答助手。只能根据下面提供的资料回答， 如果资料中没有相关信息，明确回答"资料库中未找到相关内容"，不要自行编造。 回答时注明信息来自哪些资料片段。"""),("human","资料：\n{context}\n\n问题：{input}"),])defformat_docs(docs):return"\n\n---\n\n".join(f"【片段{i+1}】{d.page_content}"fori,dinenumerate(docs))rag_chain=({"context":retriever|format_docs,"input":lambdax:x["question"]}|qa_prompt|llm)answer=rag_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_prompt=ChatPromptTemplate.from_template("根据对话历史，把用户的最新问题改写成一条可独立检索的完整问题。\n""历史：{history}\n最新问题：{question}\n只输出改写后的问题。")return(condense_prompt|llm).invoke({"history":history,"question":question}).content history=[HumanMessage(content="年假申请需要提前几天？")]q=condense_question(history,"那最多能请几天呢？")# 输出类似：企业年假单次最多可申请的天数是多少？

改写这一步质量差，后续检索必然跟着差，所以值得单独建评测用例去验证。

四、评测与调优：让检索质量可度量

知识问答系统上线前必须回答一个问题：检索准不准？回答对不对？用人工逐条看太慢，推荐两段式自动评测：

1. 检索质量：准备一批"问题 → 应命中片段"的标注数据，计算 Recall@K——前 K 个检索结果里包含标准片段的比例。
2. 2. 回答质量：用 LLM-as-Judge 打分，重点检查"回答是否基于给定资料"和"是否遗漏关键信息"。

defrecall_at_k(retriever,questions,golden,k=4):hit=0forq,goldinzip(questions,golden):docs=retriever.invoke(q)[:k]texts=[d.page_contentfordindocs]ifany(gintfortintexts):hit+=1returnhit/len(questions)``` 有了度量，调参就不再靠感觉：chunk_size 从400试到1200，overlap 从50试到200，k 从3试到6，每次改完跑一遍 Recall 和 Judge 分数，留下最优组合。这套流程才是 RAG 工程真正的核心工作量所在。## 五、进阶：RunnableParallel 与流式组合最后补充一个生产高频用法。检索、重写等耗时操作可以用 `RunnableParallel` 并行执行，让整体延迟从"串行之和"变成"最长路径"： ```pythonfromlangchain_core.runnablesimportRunnableParallel parallel=RunnableParallel(retrieved=retriever,condensed=condense_chain,)result=parallel.invoke({"question":"年假能拆成半天申请吗"})``` 配合 `stream`，你甚至可以在检索还没完成时就先把"正在检索资料…"的占位符流给用户，体验层面的优化空间非常大。## 六、小结LCEL 的真正价值在于把"编排"变成了数据流的自然表达：组件可组合、链路可追踪、并行和流式开箱即用。基于它搭建 RAG 系统，从文档加载到检索增强生成的每一环都可以独立测试、独立优化。记住两条主线：**检索质量靠切片与召回参数调优，回答质量靠提示词约束与评测闭环**。把这两条主线跑通，你手上就是一个真正能上生产的知识问答系统。