大概一年多前我接过一个让我印象很深的需求一个做数据运营的朋友每天要从客服后台导出几百条用户反馈然后一条条判断类别、评估情绪、写回复建议最后再汇总成一张日报发给团队。每次两三个小时纯手工机械重复到让人怀疑人生。当时我的第一反应是写个脚本调用大模型接口来处理但真动手之后发现问题远不是调个API这么简单——几百条文本要分类、要情绪判断、要生成回复草稿中间还有可能需要人工介入的流程分支这些逻辑散落在一个脚本里很快就变成一团乱麻。最后我改用LangChain把这些步骤串成一条自动化工作流再用接口和定时任务部署起来效果比预期好了太多维护成本也直线下降。这篇文章就是把我从零搭建这条LangChain自动化工作流的完整过程、代码、部署方式和踩过的坑都整理出来。内容不绕弯子直接给你能落地的东西核心组件的正确组合方式、流程如何分叉与并行、状态怎么保存、完整可运行的代码怎么写、部署时需要注意什么。无论你是刚接触LangChain的新手还是已经写了一堆链但觉得不好维护的开发者这篇都值得你花二十分钟看完。1. 为什么是LangChain而不是自己写一堆函数1.1 一个让我动手的场景先说回开头那个需求。朋友的工单文本长这样App 更新之后每次点进直播间就闪退重启也没用能不能快点修复我最初的方案特别朴素写一个Python脚本把所有文本拼接成一个大提示词丢给模型让模型一次性返回处理结果。听起来简单可一旦数据量上来第一批问题就出现了——单次Token上限根本塞不下几百条文本让模型一次性做分类情绪回复建议三个任务输出格式漂移严重今天返回JSON明天就给你来个Markdown表格更麻烦的是有些NEGATIVE情绪的工单需要转人工有些则不用这种条件分支逻辑写在一次调用里非常别扭。我知道这时候需要一个能把多步、多分支、有状态的流程组织起来的框架LangChain就是干这个的。1.2 它解决的是胶水层的问题LangChain真正解决的不是调用大模型这件事而是调用之外的胶水层提示词模板管理、模型统一封装、输出解析、步骤编排、状态记忆、工具调用。打个比方大模型API就像一套毛坯房你当然可以自己拉电线、铺水管、贴瓷砖但这套工程量不小而且容易出问题。LangChain像一个装修公司不仅给了你标准化的水电管线模型接口还给了你可组合的模块提示词模板、解析器、Runnable组件让你把精力放在房间怎么布局上而不是每块砖怎么砌。在工作流这个场景下LangChain尤其合适的一点是它有一套统一的接口抽象——LCELLangChain Expression Language。整条流程里任意一个步骤都可以用一个Runnable抽象来表示两个Runnable之间用|就能串联用字典就能实现并行用函数就能插入自定义逻辑。这种设计带来的直接好处是你的流程不再是藏在函数调用堆栈里的隐式逻辑而是一个可以被直观描述、被单独测试、被随时替换节点的结构。1.3 什么时候真没必要用LangChain这我必须说句实话。如果你只是单纯想调用一下大模型拿个字符串结果且代码不到五十行那直接写requests调用API反而更轻。LangChain的抽象在简单场景下会让你多学一堆概念这是学习成本。我见过不少人在只有三五个步骤的脚本里硬套LangChain结果反而被各种组件概念绕晕。它真正的价值出现在这类情况流程步骤超过四个包含了并行、条件分支、循环、状态保存输出需要严格解析成结构化数据页面或者服务里需要多处复用同一流程。拿这些条件对照你手头的需求如果命中两个以上LangChain才是一个值得考虑的选择。这也是我在这篇文章里不断强调工作流的原因——我们解决的不是单次调用而是流程编排。2. 环境准备版本号才是第一个坑2.1 安装依赖和三方包的边界LangChain这套生态的版本演进速度很快我落地时用的版本组合拿到今天依然可用但你要注意一个核心原则别装一个巨大的langchain包就算完而是按需安装分体包。LangChain从较早的版本开始就把不同供应商和组件拆成了独立包。我的习惯是这样python -m venv workflow_env source workflow_env/bin/activate pip install langchain-core langchain-openai langchain-community pip install python-dotenv pandas tabulatelangchain-core提供的是最基础的抽象——提示词模板、解析器、Runnable接口。langchain-openai是对具体模型供应商的适配层。如果你用其他模型就装对应的包。这里有个很容易踩的坑在旧教程里你会看到from langchain.llms import OpenAI这种导入方式在现在的新版本里已经废弃了正确写法是from langchain_openai import ChatOpenAI。装包之前先确认代码里的导入路径跟包版本是否对得上不然报错会让你怀疑人生。2.2 API Key、模型接入和可观测性准备第一次跑通之前最容易被忽略的是环境变量管理。我不建议在代码里硬编码API Key哪怕是你自己本地测试用.env文件加上python-dotenv是更稳妥的习惯。文件内容长这样OPENAI_API_KEYsk-你的key代码开头加载from dotenv import load_dotenv load_dotenv()除了Key还有一个很多人不知道的细节LangChain会把完整提示词发给每个模型提供方。如果你处理的是敏感数据务必先确认这套链路是否符合你的数据合规要求。技术上可以在环境变量里开启调试日志把实际发送给模型的Prompt和返回结果都打出来这在你排查为什么输出不对的时候几乎是救命级别的功能export LANGCHAIN_TRACING_V2false export LANGCHAIN_LOG_LEVELDEBUG看到这你可能觉得啰嗦但我可以负责任地说90%的LangChain入门失败都不是代码逻辑错了而是环境和版本不匹配。先把这一步稳住后面的内容才有意义。3. 最小闭环提示词模板、模型与输出解析器怎么组合3.1 用LCEL写出的第一条能跑的链不管未来流程多复杂起点一定是输入文本 - 模型回复 - 输出结果这一条最小链路。LCEL的最大特点是写法像管道命令极其直观from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_template( 你是客户服务专家。请判断以下用户反馈的类别类别只能是技术问题/账单问题/使用咨询/投诉建议。\n反馈内容{content}\n类别 ) model ChatOpenAI(modelgpt-4o-mini, temperature0) parser StrOutputParser() classify_chain prompt | model | parser if __name__ __main__: result classify_chain.invoke({content: App 更新之后每次点进直播间就闪退重启也没用}) print(result)这段代码如果你之前有LangChain基础会觉得平平无奇。但我想借它讲清楚三件事。第一ChatPromptTemplate.from_template里{content}是运行时传入的变量它做的事情是模板渲染保证每次调用模型前提示词按统一格式生成。把提示词集中管理而不是散落在代码各处拼接字符串是后面能大规模扩展的根基。第二ChatOpenAI(modelgpt-4o-mini, temperature0)这里的temperature0不是随便填的。做分类、提取这类确定性任务时温度越高输出越发散0能让模型尽可能稳定地输出。这不是玄学是概率采样机制决定的后面我会专门讲这个坑。第三prompt | model | parser这行管道表达式是LCEL的精髓。|把三个组件串成一个大的Runnable这个Runnable自带invoke、batch、stream方法。你能对单条输入invoke也能对整个列表batch还能流式获取输出。同一个对象生产环境里不同场景都能用这是手写函数很难做到的统一性。3.2 解析器比想象中重要上面那条链返回的是字符串技术问题。但真实工作流里我们通常需要结构化输出。比如同时让模型返回类别、情绪、置信度这时候就不能靠StrOutputParser了需要对输出做严格解析。我强烈推荐使用PydanticOutputParser它结合了提示词引导和数据结构校验两层作用。看这段代码from pydantic import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class TicketAnalysis(BaseModel): category: str Field(description类别只能是技术问题/账单问题/使用咨询/投诉建议) sentiment: str Field(description情绪只能是POSITIVE/NEUTRAL/NEGATIVE) confidence: float Field(description0到1之间的置信度) analysis_parser PydanticOutputParser(pydantic_objectTicketAnalysis) prompt ChatPromptTemplate.from_template( 分析以下用户反馈。 要求{format_instructions} 反馈内容{content} ).partial(format_instructionsanalysis_parser.get_format_instructions()) analysis_chain prompt | model | analysis_parser res analysis_chain.invoke({content: 这个月账单多扣了50块钱怎么回事太气人了}) print(res.category, res.sentiment, res.confidence)这类解析器做的事情是把模型返回的JSON字符串按Pydantic模型定义做类型校验如果缺字段或者格式错误它会抛出异常或者自动修复。实际用下来有两点体会。一是format_instructions会被拼进提示词里相当于你告诉模型你输出必须按这个JSON结构来。模型遵循指令的能力很强配合解析器双保险成功率高很多。二是一旦遇到解析失败建议不要在测试时跳过异常。解析失败往往是模型输出被截断、中途夹杂了多余文字、或者字段值不符合枚举范围。这些信息比任何日志都有价值能直接告诉你是提示词不够明确还是温度太高还是上下文太长截断了。3.3 为什么是管道符强调一下LCEL里的|不是Python语法糖它背后有真实的组合逻辑。左边的Runnable的invoke输出会自动被作为右边Runnable的输入。如果输出是一个字典而下一个组件的输入需要字典它们就正好对接。如果你的步骤之间数据结构不匹配管道传入时会报类型错误这其实是好事——它把数据类型不匹配问题提前暴露在了流程组合阶段而不是让你在某个深夜调一个神秘函数时才发现。这个设计理念贯穿整套工作流每一步都是Runnable每一步的输入输出都有约定。当你需要替换某个步骤比如把文案模型从a换成b你只需要换中间那一节前后接口不动整条链不需要改这是传统函数堆叠很难做到的事情。4. 从一条链升级为一张图并行、路由与记忆4.1 并行分支RunnableParallel让互不依赖的动作同时跑如果我一开始就把上面那条分析链用起来处理朋友的几百条工单每次都只调一次模型倒是能跑但速度一般。更重要的是许多步骤之间其实没有依赖关系——比方说判断类别、判断情绪、判断是否有退费请求这三件事互不依赖完全可以并行调用模型各算各的最后再汇合。LCEL里做并行有两种方式一是RunnableParallel二就是直接传一个字典系统会隐式构造并行节点from langchain_core.runnables import RunnableParallel def merge_results(analysis: TicketAnalysis, urgent_mark: str) - dict: return { category: analysis.category, sentiment: analysis.sentiment, confidence: analysis.confidence, urgent_mark: urgent_mark } analysis_chain RunnableParallel( analysisprompt | model | analysis_parser, urgent_markanother_chain # 这里可以是任意Runnable ) | merge_resultsRunnableParallel会把字典里每个值当作一个独立分支去执行然后在一个节点汇合。实测中两个分支并行通常能比串行省一半时间。在这个例子里分类和情绪判断差不多耗时所以提速很明显。如果某个分支特别慢你还可以用RunnableLambda对它的超时和重试做单独控制这个我后面部署部分会再提。4.2 自定义节点RunnableLambda把普通函数接入流程不是工作流里每一步都需要调用模型。有时候我们需要跑一段Python逻辑比如判断置信度是否低于0.6低于就走人工又比如把模型返回的结果做一次字段重命名。这时候RunnableLambda就派上用场了它能把任意普通函数包装成流程里的一个节点from langchain_core.runnables import RunnableLambda def check_confidence(x: dict) - dict: x[need_review] x[confidence] 0.6 or x[sentiment] NEGATIVE return x process_chain analysis_chain | RunnableLambda(check_confidence)用起来要注意一个细节RunnableLambda包裹的函数最好接收一个参数、返回一个结果并且逻辑要尽量纯——不要在节点里面偷偷改全局状态也不要去打印大段日志否则以后你想并行跑流程时会非常痛苦。把检查和做副作用的事比如发邮件、写数据库分开是工作流设计里的一条好习惯。4.3 条件路由根据中间结果决定下一步如果流程只需要顺序往下走那还谈不上工作流。真正让流程有智能感的是条件路由——根据某一步的结果决定接下来走哪条分支。LCEL里做路由通常用RunnableBranchfrom langchain_core.runnables import RunnableBranch def is_negative(x: dict) - bool: return x[sentiment] NEGATIVE reject_chain prompt_reject | model | parser # 转人工处理链 normal_chain prompt_normal | model | parser # 正常回复链 route_chain RunnableBranch( (is_negative, reject_chain), normal_chain # 默认分支 ) full_chain process_chain | route_chain这段代码的意思是先跑前面的处理拿到带sentiment字段的结果如果情绪是NEGATIVE就走人工介入的reject_chain否则走普通回复链。我一开始特别喜欢把这种路由逻辑写在RunnableLambda里用一堆if else去调不同的链。但后来发现这样很容易让流程变成一团意大利面。RunnableBranch把条件和分支显式地列在一起流程图长什么样代码就长什么样排查和维护都清晰得多。条件函数本身可以是普通函数执行效率上也没有额外开销强烈建议替换掉你那堆手写的if else。4.4 状态管理跨步骤记忆不要全放内存路由做多了之后你自然会发现流程里的中间结果越来越多这条工单的类别、那条工单的情绪、上一步生成的回复草稿……如果这些状态只存在局部变量里流程一旦出错重启前面的计算全部白费。更麻烦的是当你把工作流部署成常驻服务多个人同时调用时局部变量根本撑不住。LangChain本身提供了一些记忆组件但在自动化工作流场景里我一般不太依赖框架自带的内存型记忆。我更推荐的做法是把流程中间状态显式落库哪怕只是一个挂在langchain-core之上的SQLite表。原理很简单——大模型调用是昂贵的、有延迟的而状态读写是廉价的、瞬时的。与其让一次完整的流程从头跑到尾不如在关键节点记录进度这样即使某次调用超时、服务重启流程也能从断点重来。关于具体怎么落库、字段怎么设计我不展开讲因为不同业务差别太大。但你只要记住一个原则流程是会被打断的你的状态存储也必须能承受打断。临时方案可以用本地JSON文件或者SQLite生产环境至少也得是Redis。这个思路会在部署环节帮你省下大量排查时间。5. 实战Demo客户工单自动处理流水线附完整代码5.1 流程设计前面讲了这么多组件这里我把它们全部组装成一个可以跑起来的完整Demo。场景是客户工单自动处理流水线输入一批工单文本系统自动逐条完成分类、情绪判断、回复草稿生成并对高风险工单做拦截标记最后把所有结果汇总成一份当日报告。流程结构如下输入CSV工单列表 - 逐条处理(并行顺序混合) - 分支A分类 情绪判断 风险标记 - 分支B根据分类和情绪生成回复草稿 - 高风险工单过滤 汇总报告 - 结果落盘JSON这里分支A内部并行分类和情绪同时跑分支B依赖分支A的结果必须先知道情绪和分类才能生成合适的回复所以整体是一个并行-汇合-顺序-汇合的结构。5.2 数据准备和模型初始化先准备一份测试数据tickets.csv模拟从客服后台导出的工单id,content,platform,time 1001,App 更新之后每次点进直播间就闪退 重启也没用,Android,2025-01-20 09:00 1002,这个月账单多扣了50块钱 怎么回事,Web,2025-01-20 09:05 1003,请问会员怎么取消自动续费,iOS,2025-01-20 09:12 1004,你们的客服电话打了三次都没人接 太失望了,N/A,2025-01-20 09:30初始化代码import os import csv import json from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser from langchain_core.runnables import RunnableParallel, RunnableLambda, RunnableBranch from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field load_dotenv() model ChatOpenAI(modelgpt-4o-mini, temperature0, max_retries1, timeout60) class TicketAnalysis(BaseModel): category: str Field(description工单类别只能是技术问题/账单问题/使用咨询/投诉建议) sentiment: str Field(description用户情绪只能是POSITIVE/NEUTRAL/NEGATIVE) confidence: float Field(description置信度0到1之间) analysis_parser PydanticOutputParser(pydantic_objectTicketAnalysis) analysis_prompt ChatPromptTemplate.from_template( 你是客户工单质量分析专家。分析下面的用户反馈内容。 {format_instructions} 反馈内容{content} ).partial(format_instructionsanalysis_parser.get_format_instructions()) reply_prompt ChatPromptTemplate.from_template( 你是某公司的客服专员。根据工单的分词结果和情绪写一条给用户的回复草稿。 要求语气专业友好、简洁如果是NEGATIVE情绪先致歉并给出解决方向。 工单内容{content} 类别{category} 情绪{sentiment} 回复草稿 )注意我在初始化里就设置了max_retries1和timeout60。这算是一个实用经验默认值在长流程里可能会导致整体卡死宁可让单次调用快速失败并触发重试也不要在一棵树上吊死。后面部署部分还会围绕它做并发控制。5.3 分类、情绪与回复生成这一节是工作流的核心直接看代码# 创建分析链并行执行分类和情绪判断然后合并结果 analysis_chain ( RunnableParallel( analysisanalysis_prompt | model | analysis_parser, # 这里故意只放一个分支实际可以放多个并行分支 ) | RunnableLambda(lambda x: { content: x[content], category: x[analysis].category, sentiment: x[analysis].sentiment, confidence: x[analysis].confidence, }) ) reply_chain ( reply_prompt | model | StrOutputParser() ) # 定义路由函数NEGATIVE情绪走人工通道其他走普通回复 def is_negative(x: dict) - bool: return x[sentiment] NEGATIVE def manual_route(x: dict) - dict: x[reply] [人工介入] 已标记为高优先级需人工联系用户处理。 x[need_review] True return x def normal_route(x: dict) - dict: x[reply] 您好我们已收到您的反馈正在处理中。 x[need_review] False return x # 这块的逻辑值得细看reply_chain依赖路由决策结果而决策又依赖上面的分析结果 def generate_route(x: dict) - dict: # x里已经带上了content/category/sentiment/confidence if is_negative(x): return manual_route(x) else: result reply_chain.invoke({ content: x[content], category: x[category], sentiment: x[sentiment], }) x[reply] result x[need_review] False return x process_chain analysis_chain | RunnableLambda(generate_route)这段代码我用了一个稍微冗余但更容易理解的路由写法直接在普通函数里判断分支。你也可以换成RunnableBranch但如果分支里要调用另一个链并拿返回值我个人觉得在RunnableLambda里显式调用更直白。注意我这里的注释强调了reply_chain依赖路由决策结果所以它不能放在并行分支里必须先等分析结果出来才能跑。这里再补充一个容易踩的点analysis分支解析出来的是一个TicketAnalysis对象不是字典。所以我要在RunnableLambda里手动把它拆成category、sentiment、confidence三个字段再跟原始content合并成一个平铺的字典。如果不做这步后续所有引用x[category]的地方都会因为取不到字段直接报错。5.4 高风险拦截、汇总与落盘单条流程处理完之后我需要把所有结果收集起来再过滤出高风险工单最后生成一份汇总报告并保存def load_tickets(): tickets [] with open(tickets.csv, moder, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: tickets.append(row) return tickets def run_workflow(tickets): results [] for ticket in tickets: r process_chain.invoke({content: ticket[content]}) r[id] ticket[id] r[platform] ticket[platform] results.append(r) return results def build_report(results): high_risk [r for r in results if r[need_review]] report_parts [当日工单处理报告, f总工单数{len(results)}, f需人工介入{len(high_risk)}, ] for r in results: report_parts.append( f工单{r[id]} [{r[category]} / {r[sentiment]}] 置信度{r[confidence]:.2f} | {r[reply]} ) return \n.join(report_parts) if __name__ __main__: tickets load_tickets() results run_workflow(tickets) with open(output.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) report build_report(results) print(report)运行之后output.json里每条工单会有类似这样的结构{ id: 1001, content: App 更新之后每次点进直播间就闪退 重启也没用, category: 技术问题, sentiment: NEGATIVE, confidence: 0.91, reply: [人工介入] 已标记为高优先级需人工联系用户处理。, need_review: true, platform: Android }我特意把need_review做成布尔字段而不是依赖情绪字段去隐式判断这样下游筛选时写[r for r in results if r[need_review]]非常干净。数据落盘用ensure_asciiFalse保存中文也是调试中文输出时容易忽略的细节。5.5 运行效果与关键输出说明实际跑一遍数据输出报告会是这样当日工单处理报告 总工单数4 需人工介入2 工单1001 [技术问题 / NEGATIVE] 置信度0.91 | [人工介入] 已标记为高优先级需人工联系用户处理。 工单1002 [账单问题 / NEGATIVE] 置信度0.87 | [人工介入] 已标记为高优先级需人工联系用户处理。 工单1003 [使用咨询 / NEUTRAL] 置信度0.78 | 您好我们已收到您的反馈正在处理中。 工单1004 [投诉建议 / NEGATIVE] 置信度0.95 | [人工介入] 已标记为高优先级需人工联系用户处理。测试代码时我的建议是先跑两条典型工单一条NEGATIVE、一条NEUTRAL确认分支逻辑对再跑全量。不要一上来就灌几百条因为你还没验证模型输出质量一次跑太多既费Token又难排查问题。6. 部署到真实环境API化、定时调度与成本控制6.1 用FastAPI把工作流暴露成接口本地跑通只是第一步实际部署时这条工作流通常要被外部系统调用比如客服后台点个按钮就提交一批工单去处理。最省事的方案是包一层FastAPI接口from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class WorkflowRequest(BaseModel): tickets: list[str] class WorkflowResponse(BaseModel): report: str high_risk_count: int app.post(/run_workflow, response_modelWorkflowResponse) def run_workflow_endpoint(req: WorkflowRequest): tickets [{id: fonline_{i}, content: text, platform: API} for i, text in enumerate(req.tickets)] try: results run_workflow(tickets) report build_report(results) high_risk_count len([r for r in results if r[need_review]]) return WorkflowResponse(reportreport, high_risk_counthigh_risk_count) except Exception as e: raise HTTPException(status_code500, detailstr(e))这里有两个部署细节值得注意。第一接口里不要同步跑完整流程尤其当你的工作流单次可能要调好几次模型、耗时十几秒甚至更久时HTTP请求会被拖到超时。更合理的做法是把任务丢进后台队列接口立刻返回task_id前端轮询任务状态。如果你不想引一堆消息队列最轻量的方案是用进程内的任务队列库。第二异常处理要放到最外层。前面说过模型输出解析失败、网络超时、并发冲突这些问题在长流程里不可避免接口层必须能捕获并返回可读的错误信息而不是把堆栈甩给前端。6.2 定时调度与队列如果要每天固定时间跑一次日报汇总我建议直接用轻量的调度库而不是上完整的工作流引擎。我常用的是APSchedulerfrom apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def daily_report_job(): tickets load_tickets() # 从数据库或文件加载当天增量数据 results run_workflow(tickets) report build_report(results) # 发送到企业微信/钉钉/邮件这里省略 save_report(report) scheduler BlockingScheduler() scheduler.add_job(daily_report_job, CronTrigger(hour18, minute30)) scheduler.start()定时任务虽然简单但有个隐藏问题如果某一次调度时工作流没跑完就被杀掉了怎么办所以我在前面强调状态要落库。定时任务加上幂等性处理更稳妥——比如每次生成一个批次号处理完更新状态为已完成下次调度先检查这个批次是否已经处理过避免重复调用模型产生废Token。6.3 并发上限与Token预算工作流一旦部署就一定会遇到并发和成本两个问题。模型API通常有每分钟请求数和Token数限制并发太高直接被限流太低又浪费资源。我的控制套路是三层信号量限流、模型分级、结果缓存。import threading semaphore threading.Semaphore(4) def safe_invoke(chain, payload): with semaphore: return chain.invoke(payload)这个Semaphore(4)配合上面模型的max_retries1能在并发达到上限前让请求排队等待而不是一把梭把API打爆。模型分级指的是像是否NEGATIVE这种二分类问题用便宜的轻量模型像生成回复草稿这种偏创意的任务才用更强的模型。千万别所有步骤都用一个又贵又慢的大模型成本差几倍不是开玩笑。结果缓存就更好理解了同一批工单、同一个处理版本结果应该是一致的做一层哈希缓存重复请求直接命中历史结果。7. 踩坑记录四个让我回头改代码的实战问题7.1 模型输出不稳定温度设0只是第一步我最早跑分类任务时明明给模型规定了只能返回技术问题/账单问题/使用咨询/投诉建议但它偶尔还是会返回技术类这种变体导致解析器直接报错。排查了半天发现根因是temperature默认值偏高——我之前从别的项目复制来配置忘了改。把temperature0设置好之后变体问题改善了八成但还没根治。真正有效的做法是双重校验既在提示词里枚举所有合法值又在解析器里做字段校验。如果枚举值有限更保险的方式是直接在解析完之后做一个Python层面的校正映射把技术类映射回技术问题。不要指望模型一次回答完美你的流程里总得有一层兜底逻辑。7.2 Prompt是需要版本管理的配置这个坑是我改Prompt踩出来的。某一天我发现同一批工单的输出质量突然下降翻聊天记录才想起来前一天顺手改过提示词里的一个措辞把请判断反馈类别改成了请对反馈进行分类。看起来差不多但模型输出风格确实变了。后来我把所有提示词模板挪到单独的配置文件里并且用语义化的方式做了版本管理。每次改动Prompt都要记录一句这次改了什么、为什么改。你可能觉得小题大做但Prompt直接影响模型行为它跟代码一样重要。一个完善的自动化工作流里Prompt绝不能被当成一次性字符串塞在代码里。7.3 状态持久化——内存里的记忆一旦重启就消失这个前面已经反复提醒过。我第一个版本的处理结果都放在进程内存里后来部署成常驻服务某次发布更新后服务重启当天还没处理完的批次数据全部清零只能重新跑一遍。那种明明已经处理了三十条重启后全部从头再来的心情体验过一次你就再也不想经历了。我的教训是批次处理类工作流状态必须持久化。哪怕是往数据库里写一行批次号、工单ID、处理状态、结果摘要成本忽略不计收益却巨大。这个状态表同时还能帮你做断点续跑和排查问题。7.4 全自动是理想人工闸门才是现实做自动化流程时容易有一种冲动把所有环节都交给模型追求零人工介入。但实践下来真正稳妥的设计是在关键位置留一道人工闸门。比如我上面Demo里的need_review标记NEGATIVE情绪或者低置信度的工单默认不直接发回复而是进入人工复核列表。有朋友问我这样不是增加人工成本了吗我的回答是模型回复错了后果是用户投诉升级那个成本远高于人工看一眼。自动化工作流的价值在于把人的精力从80%的常规工作中解放出来让人集中处理真正需要判断力的20%而不是试图消灭人的作用。这道闸门既是产品质量的保护也是对用户负责。我自己在这条工作流上线后又迭代了好几版加过自动化邮件通知、接过分时段限流、也给高风险工单单独开过加急通道。但核心的这套结构一直没变提示词模板、模型调用、输出解析、并行分叉、条件路由、状态落库、人工兜底。你把这几个点吃透了拿去套到任何批量文本处理决策输出的业务场景里基本都不会跑偏。这也是我最想让你从这篇文章里拿走的东西——LangChain的组件会更新API会变但流程编排的思路和工程实践是长期有效的。 
长河君 · 2026/10/19 10:42:02 · 阅读 1.2k 从零搭建LangChain自动化工作流：核心组件、并行路由与部署实践
大概一年多前我接过一个让我印象很深的需求一个做数据运营的朋友每天要从客服后台导出几百条用户反馈然后一条条判断类别、评估情绪、写回复建议最后再汇总成一张日报发给团队。每次两三个小时纯手工机械重复到让人怀疑人生。当时我的第一反应是写个脚本调用大模型接口来处理但真动手之后发现问题远不是调个API这么简单——几百条文本要分类、要情绪判断、要生成回复草稿中间还有可能需要人工介入的流程分支这些逻辑散落在一个脚本里很快就变成一团乱麻。最后我改用LangChain把这些步骤串成一条自动化工作流再用接口和定时任务部署起来效果比预期好了太多维护成本也直线下降。这篇文章就是把我从零搭建这条LangChain自动化工作流的完整过程、代码、部署方式和踩过的坑都整理出来。内容不绕弯子直接给你能落地的东西核心组件的正确组合方式、流程如何分叉与并行、状态怎么保存、完整可运行的代码怎么写、部署时需要注意什么。无论你是刚接触LangChain的新手还是已经写了一堆链但觉得不好维护的开发者这篇都值得你花二十分钟看完。1. 为什么是LangChain而不是自己写一堆函数1.1 一个让我动手的场景先说回开头那个需求。朋友的工单文本长这样App 更新之后每次点进直播间就闪退重启也没用能不能快点修复我最初的方案特别朴素写一个Python脚本把所有文本拼接成一个大提示词丢给模型让模型一次性返回处理结果。听起来简单可一旦数据量上来第一批问题就出现了——单次Token上限根本塞不下几百条文本让模型一次性做分类情绪回复建议三个任务输出格式漂移严重今天返回JSON明天就给你来个Markdown表格更麻烦的是有些NEGATIVE情绪的工单需要转人工有些则不用这种条件分支逻辑写在一次调用里非常别扭。我知道这时候需要一个能把多步、多分支、有状态的流程组织起来的框架LangChain就是干这个的。1.2 它解决的是胶水层的问题LangChain真正解决的不是调用大模型这件事而是调用之外的胶水层提示词模板管理、模型统一封装、输出解析、步骤编排、状态记忆、工具调用。打个比方大模型API就像一套毛坯房你当然可以自己拉电线、铺水管、贴瓷砖但这套工程量不小而且容易出问题。LangChain像一个装修公司不仅给了你标准化的水电管线模型接口还给了你可组合的模块提示词模板、解析器、Runnable组件让你把精力放在房间怎么布局上而不是每块砖怎么砌。在工作流这个场景下LangChain尤其合适的一点是它有一套统一的接口抽象——LCELLangChain Expression Language。整条流程里任意一个步骤都可以用一个Runnable抽象来表示两个Runnable之间用|就能串联用字典就能实现并行用函数就能插入自定义逻辑。这种设计带来的直接好处是你的流程不再是藏在函数调用堆栈里的隐式逻辑而是一个可以被直观描述、被单独测试、被随时替换节点的结构。1.3 什么时候真没必要用LangChain这我必须说句实话。如果你只是单纯想调用一下大模型拿个字符串结果且代码不到五十行那直接写requests调用API反而更轻。LangChain的抽象在简单场景下会让你多学一堆概念这是学习成本。我见过不少人在只有三五个步骤的脚本里硬套LangChain结果反而被各种组件概念绕晕。它真正的价值出现在这类情况流程步骤超过四个包含了并行、条件分支、循环、状态保存输出需要严格解析成结构化数据页面或者服务里需要多处复用同一流程。拿这些条件对照你手头的需求如果命中两个以上LangChain才是一个值得考虑的选择。这也是我在这篇文章里不断强调工作流的原因——我们解决的不是单次调用而是流程编排。2. 环境准备版本号才是第一个坑2.1 安装依赖和三方包的边界LangChain这套生态的版本演进速度很快我落地时用的版本组合拿到今天依然可用但你要注意一个核心原则别装一个巨大的langchain包就算完而是按需安装分体包。LangChain从较早的版本开始就把不同供应商和组件拆成了独立包。我的习惯是这样python -m venv workflow_env source workflow_env/bin/activate pip install langchain-core langchain-openai langchain-community pip install python-dotenv pandas tabulatelangchain-core提供的是最基础的抽象——提示词模板、解析器、Runnable接口。langchain-openai是对具体模型供应商的适配层。如果你用其他模型就装对应的包。这里有个很容易踩的坑在旧教程里你会看到from langchain.llms import OpenAI这种导入方式在现在的新版本里已经废弃了正确写法是from langchain_openai import ChatOpenAI。装包之前先确认代码里的导入路径跟包版本是否对得上不然报错会让你怀疑人生。2.2 API Key、模型接入和可观测性准备第一次跑通之前最容易被忽略的是环境变量管理。我不建议在代码里硬编码API Key哪怕是你自己本地测试用.env文件加上python-dotenv是更稳妥的习惯。文件内容长这样OPENAI_API_KEYsk-你的key代码开头加载from dotenv import load_dotenv load_dotenv()除了Key还有一个很多人不知道的细节LangChain会把完整提示词发给每个模型提供方。如果你处理的是敏感数据务必先确认这套链路是否符合你的数据合规要求。技术上可以在环境变量里开启调试日志把实际发送给模型的Prompt和返回结果都打出来这在你排查为什么输出不对的时候几乎是救命级别的功能export LANGCHAIN_TRACING_V2false export LANGCHAIN_LOG_LEVELDEBUG看到这你可能觉得啰嗦但我可以负责任地说90%的LangChain入门失败都不是代码逻辑错了而是环境和版本不匹配。先把这一步稳住后面的内容才有意义。3. 最小闭环提示词模板、模型与输出解析器怎么组合3.1 用LCEL写出的第一条能跑的链不管未来流程多复杂起点一定是输入文本 - 模型回复 - 输出结果这一条最小链路。LCEL的最大特点是写法像管道命令极其直观from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_template( 你是客户服务专家。请判断以下用户反馈的类别类别只能是技术问题/账单问题/使用咨询/投诉建议。\n反馈内容{content}\n类别 ) model ChatOpenAI(modelgpt-4o-mini, temperature0) parser StrOutputParser() classify_chain prompt | model | parser if __name__ __main__: result classify_chain.invoke({content: App 更新之后每次点进直播间就闪退重启也没用}) print(result)这段代码如果你之前有LangChain基础会觉得平平无奇。但我想借它讲清楚三件事。第一ChatPromptTemplate.from_template里{content}是运行时传入的变量它做的事情是模板渲染保证每次调用模型前提示词按统一格式生成。把提示词集中管理而不是散落在代码各处拼接字符串是后面能大规模扩展的根基。第二ChatOpenAI(modelgpt-4o-mini, temperature0)这里的temperature0不是随便填的。做分类、提取这类确定性任务时温度越高输出越发散0能让模型尽可能稳定地输出。这不是玄学是概率采样机制决定的后面我会专门讲这个坑。第三prompt | model | parser这行管道表达式是LCEL的精髓。|把三个组件串成一个大的Runnable这个Runnable自带invoke、batch、stream方法。你能对单条输入invoke也能对整个列表batch还能流式获取输出。同一个对象生产环境里不同场景都能用这是手写函数很难做到的统一性。3.2 解析器比想象中重要上面那条链返回的是字符串技术问题。但真实工作流里我们通常需要结构化输出。比如同时让模型返回类别、情绪、置信度这时候就不能靠StrOutputParser了需要对输出做严格解析。我强烈推荐使用PydanticOutputParser它结合了提示词引导和数据结构校验两层作用。看这段代码from pydantic import BaseModel, Field from langchain_core.output_parsers import PydanticOutputParser class TicketAnalysis(BaseModel): category: str Field(description类别只能是技术问题/账单问题/使用咨询/投诉建议) sentiment: str Field(description情绪只能是POSITIVE/NEUTRAL/NEGATIVE) confidence: float Field(description0到1之间的置信度) analysis_parser PydanticOutputParser(pydantic_objectTicketAnalysis) prompt ChatPromptTemplate.from_template( 分析以下用户反馈。 要求{format_instructions} 反馈内容{content} ).partial(format_instructionsanalysis_parser.get_format_instructions()) analysis_chain prompt | model | analysis_parser res analysis_chain.invoke({content: 这个月账单多扣了50块钱怎么回事太气人了}) print(res.category, res.sentiment, res.confidence)这类解析器做的事情是把模型返回的JSON字符串按Pydantic模型定义做类型校验如果缺字段或者格式错误它会抛出异常或者自动修复。实际用下来有两点体会。一是format_instructions会被拼进提示词里相当于你告诉模型你输出必须按这个JSON结构来。模型遵循指令的能力很强配合解析器双保险成功率高很多。二是一旦遇到解析失败建议不要在测试时跳过异常。解析失败往往是模型输出被截断、中途夹杂了多余文字、或者字段值不符合枚举范围。这些信息比任何日志都有价值能直接告诉你是提示词不够明确还是温度太高还是上下文太长截断了。3.3 为什么是管道符强调一下LCEL里的|不是Python语法糖它背后有真实的组合逻辑。左边的Runnable的invoke输出会自动被作为右边Runnable的输入。如果输出是一个字典而下一个组件的输入需要字典它们就正好对接。如果你的步骤之间数据结构不匹配管道传入时会报类型错误这其实是好事——它把数据类型不匹配问题提前暴露在了流程组合阶段而不是让你在某个深夜调一个神秘函数时才发现。这个设计理念贯穿整套工作流每一步都是Runnable每一步的输入输出都有约定。当你需要替换某个步骤比如把文案模型从a换成b你只需要换中间那一节前后接口不动整条链不需要改这是传统函数堆叠很难做到的事情。4. 从一条链升级为一张图并行、路由与记忆4.1 并行分支RunnableParallel让互不依赖的动作同时跑如果我一开始就把上面那条分析链用起来处理朋友的几百条工单每次都只调一次模型倒是能跑但速度一般。更重要的是许多步骤之间其实没有依赖关系——比方说判断类别、判断情绪、判断是否有退费请求这三件事互不依赖完全可以并行调用模型各算各的最后再汇合。LCEL里做并行有两种方式一是RunnableParallel二就是直接传一个字典系统会隐式构造并行节点from langchain_core.runnables import RunnableParallel def merge_results(analysis: TicketAnalysis, urgent_mark: str) - dict: return { category: analysis.category, sentiment: analysis.sentiment, confidence: analysis.confidence, urgent_mark: urgent_mark } analysis_chain RunnableParallel( analysisprompt | model | analysis_parser, urgent_markanother_chain # 这里可以是任意Runnable ) | merge_resultsRunnableParallel会把字典里每个值当作一个独立分支去执行然后在一个节点汇合。实测中两个分支并行通常能比串行省一半时间。在这个例子里分类和情绪判断差不多耗时所以提速很明显。如果某个分支特别慢你还可以用RunnableLambda对它的超时和重试做单独控制这个我后面部署部分会再提。4.2 自定义节点RunnableLambda把普通函数接入流程不是工作流里每一步都需要调用模型。有时候我们需要跑一段Python逻辑比如判断置信度是否低于0.6低于就走人工又比如把模型返回的结果做一次字段重命名。这时候RunnableLambda就派上用场了它能把任意普通函数包装成流程里的一个节点from langchain_core.runnables import RunnableLambda def check_confidence(x: dict) - dict: x[need_review] x[confidence] 0.6 or x[sentiment] NEGATIVE return x process_chain analysis_chain | RunnableLambda(check_confidence)用起来要注意一个细节RunnableLambda包裹的函数最好接收一个参数、返回一个结果并且逻辑要尽量纯——不要在节点里面偷偷改全局状态也不要去打印大段日志否则以后你想并行跑流程时会非常痛苦。把检查和做副作用的事比如发邮件、写数据库分开是工作流设计里的一条好习惯。4.3 条件路由根据中间结果决定下一步如果流程只需要顺序往下走那还谈不上工作流。真正让流程有智能感的是条件路由——根据某一步的结果决定接下来走哪条分支。LCEL里做路由通常用RunnableBranchfrom langchain_core.runnables import RunnableBranch def is_negative(x: dict) - bool: return x[sentiment] NEGATIVE reject_chain prompt_reject | model | parser # 转人工处理链 normal_chain prompt_normal | model | parser # 正常回复链 route_chain RunnableBranch( (is_negative, reject_chain), normal_chain # 默认分支 ) full_chain process_chain | route_chain这段代码的意思是先跑前面的处理拿到带sentiment字段的结果如果情绪是NEGATIVE就走人工介入的reject_chain否则走普通回复链。我一开始特别喜欢把这种路由逻辑写在RunnableLambda里用一堆if else去调不同的链。但后来发现这样很容易让流程变成一团意大利面。RunnableBranch把条件和分支显式地列在一起流程图长什么样代码就长什么样排查和维护都清晰得多。条件函数本身可以是普通函数执行效率上也没有额外开销强烈建议替换掉你那堆手写的if else。4.4 状态管理跨步骤记忆不要全放内存路由做多了之后你自然会发现流程里的中间结果越来越多这条工单的类别、那条工单的情绪、上一步生成的回复草稿……如果这些状态只存在局部变量里流程一旦出错重启前面的计算全部白费。更麻烦的是当你把工作流部署成常驻服务多个人同时调用时局部变量根本撑不住。LangChain本身提供了一些记忆组件但在自动化工作流场景里我一般不太依赖框架自带的内存型记忆。我更推荐的做法是把流程中间状态显式落库哪怕只是一个挂在langchain-core之上的SQLite表。原理很简单——大模型调用是昂贵的、有延迟的而状态读写是廉价的、瞬时的。与其让一次完整的流程从头跑到尾不如在关键节点记录进度这样即使某次调用超时、服务重启流程也能从断点重来。关于具体怎么落库、字段怎么设计我不展开讲因为不同业务差别太大。但你只要记住一个原则流程是会被打断的你的状态存储也必须能承受打断。临时方案可以用本地JSON文件或者SQLite生产环境至少也得是Redis。这个思路会在部署环节帮你省下大量排查时间。5. 实战Demo客户工单自动处理流水线附完整代码5.1 流程设计前面讲了这么多组件这里我把它们全部组装成一个可以跑起来的完整Demo。场景是客户工单自动处理流水线输入一批工单文本系统自动逐条完成分类、情绪判断、回复草稿生成并对高风险工单做拦截标记最后把所有结果汇总成一份当日报告。流程结构如下输入CSV工单列表 - 逐条处理(并行顺序混合) - 分支A分类 情绪判断 风险标记 - 分支B根据分类和情绪生成回复草稿 - 高风险工单过滤 汇总报告 - 结果落盘JSON这里分支A内部并行分类和情绪同时跑分支B依赖分支A的结果必须先知道情绪和分类才能生成合适的回复所以整体是一个并行-汇合-顺序-汇合的结构。5.2 数据准备和模型初始化先准备一份测试数据tickets.csv模拟从客服后台导出的工单id,content,platform,time 1001,App 更新之后每次点进直播间就闪退 重启也没用,Android,2025-01-20 09:00 1002,这个月账单多扣了50块钱 怎么回事,Web,2025-01-20 09:05 1003,请问会员怎么取消自动续费,iOS,2025-01-20 09:12 1004,你们的客服电话打了三次都没人接 太失望了,N/A,2025-01-20 09:30初始化代码import os import csv import json from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser from langchain_core.runnables import RunnableParallel, RunnableLambda, RunnableBranch from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field load_dotenv() model ChatOpenAI(modelgpt-4o-mini, temperature0, max_retries1, timeout60) class TicketAnalysis(BaseModel): category: str Field(description工单类别只能是技术问题/账单问题/使用咨询/投诉建议) sentiment: str Field(description用户情绪只能是POSITIVE/NEUTRAL/NEGATIVE) confidence: float Field(description置信度0到1之间) analysis_parser PydanticOutputParser(pydantic_objectTicketAnalysis) analysis_prompt ChatPromptTemplate.from_template( 你是客户工单质量分析专家。分析下面的用户反馈内容。 {format_instructions} 反馈内容{content} ).partial(format_instructionsanalysis_parser.get_format_instructions()) reply_prompt ChatPromptTemplate.from_template( 你是某公司的客服专员。根据工单的分词结果和情绪写一条给用户的回复草稿。 要求语气专业友好、简洁如果是NEGATIVE情绪先致歉并给出解决方向。 工单内容{content} 类别{category} 情绪{sentiment} 回复草稿 )注意我在初始化里就设置了max_retries1和timeout60。这算是一个实用经验默认值在长流程里可能会导致整体卡死宁可让单次调用快速失败并触发重试也不要在一棵树上吊死。后面部署部分还会围绕它做并发控制。5.3 分类、情绪与回复生成这一节是工作流的核心直接看代码# 创建分析链并行执行分类和情绪判断然后合并结果 analysis_chain ( RunnableParallel( analysisanalysis_prompt | model | analysis_parser, # 这里故意只放一个分支实际可以放多个并行分支 ) | RunnableLambda(lambda x: { content: x[content], category: x[analysis].category, sentiment: x[analysis].sentiment, confidence: x[analysis].confidence, }) ) reply_chain ( reply_prompt | model | StrOutputParser() ) # 定义路由函数NEGATIVE情绪走人工通道其他走普通回复 def is_negative(x: dict) - bool: return x[sentiment] NEGATIVE def manual_route(x: dict) - dict: x[reply] [人工介入] 已标记为高优先级需人工联系用户处理。 x[need_review] True return x def normal_route(x: dict) - dict: x[reply] 您好我们已收到您的反馈正在处理中。 x[need_review] False return x # 这块的逻辑值得细看reply_chain依赖路由决策结果而决策又依赖上面的分析结果 def generate_route(x: dict) - dict: # x里已经带上了content/category/sentiment/confidence if is_negative(x): return manual_route(x) else: result reply_chain.invoke({ content: x[content], category: x[category], sentiment: x[sentiment], }) x[reply] result x[need_review] False return x process_chain analysis_chain | RunnableLambda(generate_route)这段代码我用了一个稍微冗余但更容易理解的路由写法直接在普通函数里判断分支。你也可以换成RunnableBranch但如果分支里要调用另一个链并拿返回值我个人觉得在RunnableLambda里显式调用更直白。注意我这里的注释强调了reply_chain依赖路由决策结果所以它不能放在并行分支里必须先等分析结果出来才能跑。这里再补充一个容易踩的点analysis分支解析出来的是一个TicketAnalysis对象不是字典。所以我要在RunnableLambda里手动把它拆成category、sentiment、confidence三个字段再跟原始content合并成一个平铺的字典。如果不做这步后续所有引用x[category]的地方都会因为取不到字段直接报错。5.4 高风险拦截、汇总与落盘单条流程处理完之后我需要把所有结果收集起来再过滤出高风险工单最后生成一份汇总报告并保存def load_tickets(): tickets [] with open(tickets.csv, moder, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: tickets.append(row) return tickets def run_workflow(tickets): results [] for ticket in tickets: r process_chain.invoke({content: ticket[content]}) r[id] ticket[id] r[platform] ticket[platform] results.append(r) return results def build_report(results): high_risk [r for r in results if r[need_review]] report_parts [当日工单处理报告, f总工单数{len(results)}, f需人工介入{len(high_risk)}, ] for r in results: report_parts.append( f工单{r[id]} [{r[category]} / {r[sentiment]}] 置信度{r[confidence]:.2f} | {r[reply]} ) return \n.join(report_parts) if __name__ __main__: tickets load_tickets() results run_workflow(tickets) with open(output.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) report build_report(results) print(report)运行之后output.json里每条工单会有类似这样的结构{ id: 1001, content: App 更新之后每次点进直播间就闪退 重启也没用, category: 技术问题, sentiment: NEGATIVE, confidence: 0.91, reply: [人工介入] 已标记为高优先级需人工联系用户处理。, need_review: true, platform: Android }我特意把need_review做成布尔字段而不是依赖情绪字段去隐式判断这样下游筛选时写[r for r in results if r[need_review]]非常干净。数据落盘用ensure_asciiFalse保存中文也是调试中文输出时容易忽略的细节。5.5 运行效果与关键输出说明实际跑一遍数据输出报告会是这样当日工单处理报告 总工单数4 需人工介入2 工单1001 [技术问题 / NEGATIVE] 置信度0.91 | [人工介入] 已标记为高优先级需人工联系用户处理。 工单1002 [账单问题 / NEGATIVE] 置信度0.87 | [人工介入] 已标记为高优先级需人工联系用户处理。 工单1003 [使用咨询 / NEUTRAL] 置信度0.78 | 您好我们已收到您的反馈正在处理中。 工单1004 [投诉建议 / NEGATIVE] 置信度0.95 | [人工介入] 已标记为高优先级需人工联系用户处理。测试代码时我的建议是先跑两条典型工单一条NEGATIVE、一条NEUTRAL确认分支逻辑对再跑全量。不要一上来就灌几百条因为你还没验证模型输出质量一次跑太多既费Token又难排查问题。6. 部署到真实环境API化、定时调度与成本控制6.1 用FastAPI把工作流暴露成接口本地跑通只是第一步实际部署时这条工作流通常要被外部系统调用比如客服后台点个按钮就提交一批工单去处理。最省事的方案是包一层FastAPI接口from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class WorkflowRequest(BaseModel): tickets: list[str] class WorkflowResponse(BaseModel): report: str high_risk_count: int app.post(/run_workflow, response_modelWorkflowResponse) def run_workflow_endpoint(req: WorkflowRequest): tickets [{id: fonline_{i}, content: text, platform: API} for i, text in enumerate(req.tickets)] try: results run_workflow(tickets) report build_report(results) high_risk_count len([r for r in results if r[need_review]]) return WorkflowResponse(reportreport, high_risk_counthigh_risk_count) except Exception as e: raise HTTPException(status_code500, detailstr(e))这里有两个部署细节值得注意。第一接口里不要同步跑完整流程尤其当你的工作流单次可能要调好几次模型、耗时十几秒甚至更久时HTTP请求会被拖到超时。更合理的做法是把任务丢进后台队列接口立刻返回task_id前端轮询任务状态。如果你不想引一堆消息队列最轻量的方案是用进程内的任务队列库。第二异常处理要放到最外层。前面说过模型输出解析失败、网络超时、并发冲突这些问题在长流程里不可避免接口层必须能捕获并返回可读的错误信息而不是把堆栈甩给前端。6.2 定时调度与队列如果要每天固定时间跑一次日报汇总我建议直接用轻量的调度库而不是上完整的工作流引擎。我常用的是APSchedulerfrom apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def daily_report_job(): tickets load_tickets() # 从数据库或文件加载当天增量数据 results run_workflow(tickets) report build_report(results) # 发送到企业微信/钉钉/邮件这里省略 save_report(report) scheduler BlockingScheduler() scheduler.add_job(daily_report_job, CronTrigger(hour18, minute30)) scheduler.start()定时任务虽然简单但有个隐藏问题如果某一次调度时工作流没跑完就被杀掉了怎么办所以我在前面强调状态要落库。定时任务加上幂等性处理更稳妥——比如每次生成一个批次号处理完更新状态为已完成下次调度先检查这个批次是否已经处理过避免重复调用模型产生废Token。6.3 并发上限与Token预算工作流一旦部署就一定会遇到并发和成本两个问题。模型API通常有每分钟请求数和Token数限制并发太高直接被限流太低又浪费资源。我的控制套路是三层信号量限流、模型分级、结果缓存。import threading semaphore threading.Semaphore(4) def safe_invoke(chain, payload): with semaphore: return chain.invoke(payload)这个Semaphore(4)配合上面模型的max_retries1能在并发达到上限前让请求排队等待而不是一把梭把API打爆。模型分级指的是像是否NEGATIVE这种二分类问题用便宜的轻量模型像生成回复草稿这种偏创意的任务才用更强的模型。千万别所有步骤都用一个又贵又慢的大模型成本差几倍不是开玩笑。结果缓存就更好理解了同一批工单、同一个处理版本结果应该是一致的做一层哈希缓存重复请求直接命中历史结果。7. 踩坑记录四个让我回头改代码的实战问题7.1 模型输出不稳定温度设0只是第一步我最早跑分类任务时明明给模型规定了只能返回技术问题/账单问题/使用咨询/投诉建议但它偶尔还是会返回技术类这种变体导致解析器直接报错。排查了半天发现根因是temperature默认值偏高——我之前从别的项目复制来配置忘了改。把temperature0设置好之后变体问题改善了八成但还没根治。真正有效的做法是双重校验既在提示词里枚举所有合法值又在解析器里做字段校验。如果枚举值有限更保险的方式是直接在解析完之后做一个Python层面的校正映射把技术类映射回技术问题。不要指望模型一次回答完美你的流程里总得有一层兜底逻辑。7.2 Prompt是需要版本管理的配置这个坑是我改Prompt踩出来的。某一天我发现同一批工单的输出质量突然下降翻聊天记录才想起来前一天顺手改过提示词里的一个措辞把请判断反馈类别改成了请对反馈进行分类。看起来差不多但模型输出风格确实变了。后来我把所有提示词模板挪到单独的配置文件里并且用语义化的方式做了版本管理。每次改动Prompt都要记录一句这次改了什么、为什么改。你可能觉得小题大做但Prompt直接影响模型行为它跟代码一样重要。一个完善的自动化工作流里Prompt绝不能被当成一次性字符串塞在代码里。7.3 状态持久化——内存里的记忆一旦重启就消失这个前面已经反复提醒过。我第一个版本的处理结果都放在进程内存里后来部署成常驻服务某次发布更新后服务重启当天还没处理完的批次数据全部清零只能重新跑一遍。那种明明已经处理了三十条重启后全部从头再来的心情体验过一次你就再也不想经历了。我的教训是批次处理类工作流状态必须持久化。哪怕是往数据库里写一行批次号、工单ID、处理状态、结果摘要成本忽略不计收益却巨大。这个状态表同时还能帮你做断点续跑和排查问题。7.4 全自动是理想人工闸门才是现实做自动化流程时容易有一种冲动把所有环节都交给模型追求零人工介入。但实践下来真正稳妥的设计是在关键位置留一道人工闸门。比如我上面Demo里的need_review标记NEGATIVE情绪或者低置信度的工单默认不直接发回复而是进入人工复核列表。有朋友问我这样不是增加人工成本了吗我的回答是模型回复错了后果是用户投诉升级那个成本远高于人工看一眼。自动化工作流的价值在于把人的精力从80%的常规工作中解放出来让人集中处理真正需要判断力的20%而不是试图消灭人的作用。这道闸门既是产品质量的保护也是对用户负责。我自己在这条工作流上线后又迭代了好几版加过自动化邮件通知、接过分时段限流、也给高风险工单单独开过加急通道。但核心的这套结构一直没变提示词模板、模型调用、输出解析、并行分叉、条件路由、状态落库、人工兜底。你把这几个点吃透了拿去套到任何批量文本处理决策输出的业务场景里基本都不会跑偏。这也是我最想让你从这篇文章里拿走的东西——LangChain的组件会更新API会变但流程编排的思路和工程实践是长期有效的。