LangChain 多任务应用实战，从客服到自动化全覆盖

一、前言

2024年以来，大语言模型的能力已经毋庸置疑，但真正让开发者头疼的问题从"模型行不行"变成了"怎么把它用起来"——单次问答谁都会，可一旦涉及多轮对话、多个数据源、多步骤决策，代码复杂度就指数级上升。

LangChain 正是为解决这个痛点而生的。它的核心价值不是让你写更少的代码，而是让你用更清晰的结构去组织复杂的 AI 应用逻辑。本文不讲"Hello World"，只讲真实可落地的多任务实战案例，每个案例都附上核心代码片段，读完就能直接上手改造。

二、五分钟回顾：LangChain 的"乐高积木"

在进入案例之前，先用一张表快速回顾 LangChain 的核心组件，这些组件会在后续案例中反复出现：

这些组件就像乐高积木，单个看功能有限，组合起来却能搭建出复杂系统。下面的案例，就是不同积木组合的实战演示。

三、案例一：电商智能客服——多轮对话 + 知识检索 + 实时查询

3.1 业务场景

某电商平台每天产生 5000 条客服咨询，其中 60% 是重复性问题（退货政策、物流查询、优惠券使用），客服团队 30 人，人均日处理 160 条，压力巨大。

3.2 技术架构

这个案例的核心难点在于：客服机器人需要同时具备三种能力——记住上下文（用户刚才说了什么）、检索知识库（退货政策是什么）、查询实时数据（订单现在到哪了）。单一能力都能实现，但三合一才是难点。

LangChain 的解决方案：用 Memory 管上下文，用 Retriever 管知识检索，用 Agent 管工具调用，三者协同工作。

3.3 核心代码实现

fromlangchain.memoryimportConversationBufferMemoryfromlangchain.chainsimportConversationalRetrievalChainfromlangchain_community.vectorstoresimportChromafromlangchain_community.embeddingsimportOpenAIEmbeddingsfromlangchain.chat_modelsimportChatOpenAIfromlangchain.toolsimportToolfromlangchain.agentsimportinitialize_agent,AgentType# ===== 第一步：构建知识库（一次性操作） =====# 将退货政策、保修条款等文档向量化存入 Chromavectorstore=Chroma.from_documents(documents=docs,# 已加载的文档列表embedding=OpenAIEmbeddings())retriever=vectorstore.as_retriever(search_kwargs={"k":3})# ===== 第二步：创建带记忆的对话链 =====llm=ChatOpenAI(model="gpt-4o-mini",temperature=0.3)memory=ConversationBufferMemory(memory_key="chat_history",return_messages=True)qa_chain=ConversationalRetrievalChain.from_llm(llm=llm,retriever=retriever,memory=memory,verbose=True)# ===== 第三步：定义实时查询工具 =====defquery_order_status(order_id:str)->str:"""根据订单号查询物流状态，实际项目中对接ERP接口"""# 这里模拟返回数据returnf"订单{order_id}当前状态：已发货，预计7月10日送达"defcheck_coupon_validity(coupon_code:str)->str:"""验证优惠券是否有效"""returnf"优惠券{coupon_code}有效期至2026年12月31日，满200减30"tools=[Tool(name="query_order_status",func=query_order_status,description="根据订单号查询物流状态，输入订单号字符串"),Tool(name="check_coupon_validity",func=check_coupon_validity,description="验证优惠券是否有效，输入优惠券代码")]# ===== 第四步：创建 Agent，自主决策调用哪个工具 =====agent=initialize_agent(tools=tools,llm=llm,agent=AgentType.OPENAI_FUNCTIONS,verbose=True)# ===== 第五步：对话示例 =====# 第一轮：知识库检索response1=qa_chain.invoke({"question":"买了东西不满意可以退货吗？"})print(response1["answer"])# 输出：根据我们的退货政策，自签收之日起7天内可以申请无理由退货...# 第二轮：继承上下文response2=qa_chain.invoke({"question":"那退货之后多久能收到退款？"})print(response2["answer"])# 输出：退款将在我们收到退货商品后的3-5个工作日内原路返回...# 第三轮：需要实时数据时，转交 Agent 处理response3=agent.invoke("帮我查一下订单 ORD20260701 的物流状态")print(response3["output"])# 输出：订单 ORD20260701 当前状态：已发货，预计7月10日送达

3.4 多任务协同流程

这个案例中，一个用户对话实际上触发了三种任务类型的协作：

用户提问 │ ├── 是知识类问题（退货政策）？ │ └── RetrievalQA Chain → 向量检索 → 生成回答 │ ├── 是实时数据问题（订单状态）？ │ └── Agent → 调用工具 → 返回结果 │ └── 需要上下文记忆（多轮对话）？ └── Memory → 拼接历史 → 生成连贯回答

3.5 落地效果

上线三个月后，该客服系统替代了 62% 的人工咨询量，问题一次解决率从规则引擎时代的 41% 提升至 79%，客服团队从 30 人缩减至 12 人，剩余人员专注于处理复杂纠纷和 VIP 客户服务。

四、案例二：RAG 文档问答——从 1000 份产品手册中秒级检索

4.1 业务场景

某家电企业积累了 1000 多份产品手册（PDF + Word），客服人员每天需要翻找数十份手册来回答客户关于产品参数、使用说明、故障排查的咨询。平均每次查询耗时 8 分钟，且经常遗漏关键信息。

4.2 技术方案

用 LangChain 的 LCEL 表达式语言串联 RAG 流程，实现"提问 → 检索 → 生成"的端到端问答。

4.3 核心代码实现

fromlangchain_community.document_loadersimportPyPDFLoader,Docx2txtLoaderfromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain_community.vectorstoresimportChromafromlangchain_community.embeddingsimportOpenAIEmbeddingsfromlangchain.chat_modelsimportChatOpenAIfromlangchain.promptsimportChatPromptTemplatefromlangchain_core.runnablesimportRunnablePassthroughfromlangchain_core.output_parsersimportStrOutputParserimportosimportglob# ===== 第一步：批量加载文档 =====documents=[]# 加载所有 PDFforpdf_fileinglob.glob("manuals/*.pdf"):loader=PyPDFLoader(pdf_file)documents.extend(loader.load())# 加载所有 Word 文档fordocx_fileinglob.glob("manuals/*.docx"):loader=Docx2txtLoader(docx_file)documents.extend(loader.load())print(f"共加载{len(documents)}个文档片段")# ===== 第二步：文本切割 =====text_splitter=RecursiveCharacterTextSplitter(chunk_size=500,# 每个文本块 500 字符chunk_overlap=50,# 相邻块重叠 50 字符，避免语义断裂separators=["\n\n","\n","。","."," ",""]# 优先按段落分割)docs=text_splitter.split_documents(documents)print(f"切割后共{len(docs)}个文本块")# ===== 第三步：向量化存储 =====vectorstore=Chroma.from_documents(documents=docs,embedding=OpenAIEmbeddings(),persist_directory="./chroma_db"# 持久化存储，下次直接加载)retriever=vectorstore.as_retriever(search_type="similarity",# 相似度检索search_kwargs={"k":4}# 返回最相关的 4 个文本块)# ===== 第四步：构建 RAG Prompt 模板 =====template=""" 你是一个专业的产品客服助手，请根据以下产品手册内容回答用户问题。 要求： 1. 如果手册中有明确答案，请直接引用并标注来源 2. 如果手册中没有相关信息，请诚实告知用户并建议联系人工客服 3. 回答要简洁明了，避免冗长 产品手册内容： {context} 用户问题：{question} 你的回答： """prompt=ChatPromptTemplate.from_template(template)# ===== 第五步：用 LCEL 串联整个 RAG 流程 =====llm=ChatOpenAI(model="gpt-4o-mini",temperature=0.1)rag_chain=({"context":retriever,# 检索相关文档"question":RunnablePassthrough()# 原样传递用户问题}|prompt# 注入 Prompt 模板|llm# 调用 LLM 生成回答|StrOutputParser()# 解析输出为纯文本)# ===== 第六步：测试问答 =====question="XH-200 型空气炸锅的最高温度是多少？怎么设置？"response=rag_chain.invoke(question)print(response)# 输出：# 根据 XH-200 产品手册第12页，该型号空气炸锅的最高温度为200℃。# 设置方法：旋转温控旋钮至所需温度，长按3秒确认。建议首次使用前先空烧10分钟去除异味。# 如果问手册中没有的内容question2="这个空气炸锅支持语音控制吗？"response2=rag_chain.invoke(question2)print(response2)# 输出：根据现有产品手册，未找到关于语音控制功能的说明。# 建议您联系人工客服（400-xxx-xxxx）获取更详细的产品信息。

4.4 搭配 LangSmith 监控

生产环境部署时，5 行代码接入 LangSmith 实现全链路追踪：

importos os.environ["LANGCHAIN_TRACING_V2"]="true"os.environ["LANGCHAIN_API_KEY"]="ls__你的API密钥"os.environ["LANGCHAIN_PROJECT"]="product-manual-qa"

接入后，每次问答的完整链路都会被记录——LLM 调用耗时、Token 消耗、检索命中率、回答忠实度——哪里慢了、哪里检索不准，一目了然。

4.5 落地效果

1000 份手册的检索时间从人工平均 8 分钟降至 3 秒以内，回答准确率达到 92%，日均处理咨询量从 200 条提升至 1500 条，且支持 7×24 小时不间断服务。

五、案例三：AI 代码助手——生成、解释、调试三合一

5.1 业务场景

开发团队在日常工作中需要频繁切换于代码编写、Bug 排查、代码审查和技术文档编写之间。每个任务都需要不同的工具和思路，但本质上都围绕"代码"这个核心对象。

5.2 技术方案

用 LangChain Agent 将代码生成、代码解释、错误调试三个能力封装为独立工具，由 LLM 自主判断用户意图并调用对应工具。

5.3 核心代码实现

fromlangchain.chat_modelsimportChatOpenAIfromlangchain.agentsimportinitialize_agent,AgentType,Toolfromlangchain.promptsimportPromptTemplate llm=ChatOpenAI(model="gpt-4o-mini",temperature=0)# ===== 工具1：代码生成 =====defgenerate_code(requirement:str)->str:"""根据需求描述生成代码"""prompt=PromptTemplate(input_variables=["requirement"],template=""" 请根据以下需求生成 Python 代码，要求： 1. 代码完整可运行，包含必要的 import 语句 2. 添加适当的注释 3. 遵循 PEP 8 规范 4. 包含一个简单的使用示例 需求：{requirement} """)chain=prompt|llmreturnchain.invoke({"requirement":requirement}).content# ===== 工具2：代码解释 =====defexplain_code(code:str)->str:"""逐行解释代码逻辑"""prompt=PromptTemplate(input_variables=["code"],template=""" 请用通俗易懂的语言，逐段解释以下代码的功能和逻辑。 对关键函数和复杂逻辑要详细说明。 代码： {code} """)chain=prompt|llmreturnchain.invoke({"code":code}).content# ===== 工具3：错误调试 =====defdebug_error(error_info:str)->str:"""分析错误信息，给出修复建议"""prompt=PromptTemplate(input_variables=["error_info"],template=""" 请分析以下错误信息，给出： 1. 错误原因分析 2. 具体的修复方案 3. 修复后的代码示例 错误信息： {error_info} """)chain=prompt|llmreturnchain.invoke({"error_info":error_info}).content# ===== 组装 Agent =====tools=[Tool(name="generate_code",func=generate_code,description="根据需求描述生成 Python 代码。当用户要求写代码、实现功能时使用此工具。"),Tool(name="explain_code",func=explain_code,description="解释代码的逻辑和功能。当用户要求解释代码、分析代码时使用此工具。"),Tool(name="debug_error",func=debug_error,description="分析错误信息并给出修复建议。当用户提供报错信息、询问如何修复Bug时使用此工具。")]agent=initialize_agent(tools=tools,llm=llm,agent=AgentType.OPENAI_FUNCTIONS,verbose=True)# ===== 测试：Agent 自主判断调用哪个工具 =====# 场景1：生成代码agent.invoke("帮我写一个函数，实现斐波那契数列的生成，要求能指定生成前N项")# Agent 自动调用 generate_code 工具# 场景2：解释代码（用户粘贴一段代码）agent.invoke(""" 帮我解释这段代码在做什么： def process_data(data): result = {} for item in data: key = item.get('category', 'other') if key not in result: result[key] = [] result[key].append(item.get('value', 0)) return {k: sum(v)/len(v) for k, v in result.items()} """)# Agent 自动调用 explain_code 工具# 场景3：调试错误agent.invoke(""" 代码报错了，帮我看看怎么修： TypeError: unsupported operand type(s) for +: 'int' and 'str' 代码第15行：total = count + price """)# Agent 自动调用 debug_error 工具

5.4 多任务协同

这个案例的精妙之处在于：Agent 就像一个"技术主管"，用户提需求后，它自己判断该找谁干活——是让写代码的同事上，还是让做代码审查的同事上，还是让排查 Bug 的同事上。用户不需要关心内部的分工逻辑。

六、案例四：多步骤业务流程自动化——LangGraph 编排

6.1 业务场景

电商售后流程涉及"查询订单 → 验证库存 → 判断退货条件 → 生成退款单 → 调用支付接口 → 发送通知"等多个步骤，每个步骤有分支判断和循环依赖。用传统 if-else 代码实现，逻辑稍复杂就变成难以维护的"意大利面代码"。

6.2 技术方案

LangChain 生态中的 LangGraph 专门解决这类问题，用图结构编排步骤间的逻辑分支和循环。这里用 LangGraph 构建一个简化版的售后处理流程。

6.3 核心代码实现

fromtypingimportTypedDict,Optionalfromlanggraph.graphimportStateGraph,ENDfromlangchain.chat_modelsimportChatOpenAI llm=ChatOpenAI(model="gpt-4o-mini",temperature=0)# ===== 定义状态结构 =====classOrderState(TypedDict):order_id:strstatus:str# 订单状态：delivered / returned / cancelledis_returnable:bool# 是否可退货refund_amount:floatmessage:str# 最终返回给用户的消息# ===== 定义各处理节点 =====defcheck_order_status(state:OrderState)->OrderState:"""节点1：查询订单状态"""# 实际项目对接 ERP 系统state["status"]="delivered"# 模拟查询结果returnstatedefcheck_return_eligibility(state:OrderState)->OrderState:"""节点2：判断是否满足退货条件"""# 已签收7天内可退货ifstate["status"]=="delivered":state["is_returnable"]=Truestate["message"]="订单符合退货条件"else:state["is_returnable"]=Falsestate["message"]="订单状态不符合退货要求"returnstatedefcalculate_refund(state:OrderState)->OrderState:"""节点3：计算退款金额"""state["refund_amount"]=299.0# 模拟计算returnstatedefgenerate_refund_order(state:OrderState)->OrderState:"""节点4：生成退款单"""state["message"]=f"退款单已生成，金额：{state['refund_amount']}元，预计3-5个工作日到账"returnstatedefreject_return(state:OrderState)->OrderState:"""节点5：拒绝退货"""state["message"]=f"订单{state['order_id']}不符合退货条件，请联系客服"returnstate# ===== 定义路由逻辑 =====defshould_refund(state:OrderState)->str:"""判断是否执行退款流程"""ifstate["is_returnable"]:return"calculate_refund"else:return"reject_return"# ===== 构建状态图 =====workflow=StateGraph(OrderState)# 添加节点workflow.add_node("check_order",check_order_status)workflow.add_node("check_eligibility",check_return_eligibility)workflow.add_node("calculate_refund",calculate_refund)workflow.add_node("generate_refund",generate_refund_order)workflow.add_node("reject_return",reject_return)# 添加边（定义执行顺序）workflow.set_entry_point("check_order")workflow.add_edge("check_order","check_eligibility")workflow.add_conditional_edges("check_eligibility",should_refund,{"calculate_refund":"calculate_refund","reject_return":"reject_return"})workflow.add_edge("calculate_refund","generate_refund")workflow.add_edge("generate_refund",END)workflow.add_edge("reject_return",END)# 编译并执行app=workflow.compile()# 测试：正常退货流程result=app.invoke({"order_id":"ORD20260701"})print(result["message"])# 输出：退款单已生成，金额：299.0元，预计3-5个工作日到账

6.4 流程可视化

LangGraph 的执行流程可以用图表示：

开始 → 查询订单状态 → 判断退货条件 ├── 符合条件 → 计算退款金额 → 生成退款单 → 结束 └── 不符合 → 拒绝退货 → 结束

每个节点是一个独立的处理单元，节点间的流转逻辑由路由函数控制。当业务规则变化时（比如退货条件从7天改为15天），只需修改对应节点的逻辑，不影响整体流程结构。

七、技巧总结：三个让你的 LangChain 应用更稳的实践

7.1 错误处理与降级策略

生产环境中最怕 LLM 返回格式不对或者超时，导致整个流程中断。建议为每个 Chain 加上错误处理：

fromlangchain_core.runnablesimportRunnableLambdadefsafe_invoke(chain,input_data,default="抱歉，服务暂时不可用，请稍后重试"):try:returnchain.invoke(input_data)exceptExceptionase:print(f"调用失败:{e}")returndefault

7.2 缓存策略

对于高频重复的问题（如"退货政策是什么"），可以用 LangChain 的缓存机制避免重复调用 LLM，节省 Token 成本：

fromlangchain.cacheimportInMemoryCachefromlangchain.globalsimportset_llm_cache set_llm_cache(InMemoryCache())# 开启后，相同问题第二次调用直接从缓存返回，不消耗 Token

7.3 流式输出

对于长文本生成场景，使用流式输出避免用户长时间等待：

# 流式输出，逐字返回forchunkinchain.stream({"question":"解释一下量子计算"}):print(chunk,end="",flush=True)

八、总结

LangChain 多任务应用的核心思路可以浓缩为一句话：用组件拆解业务，用编排串联流程。

如果你正在规划一个 LangChain 项目，建议从数据量最大、规则最清晰、重复性最高的那个任务开始——比如客服知识库问答，用一个 RAG 应用跑通最小闭环，验证效果后再逐步扩展为多任务系统。永远不要一上来就追求"大而全"，先让一个场景跑起来，比什么都重要。