报价审批这件事放在任何一家做项目制生意的公司里都是那种看起来不起眼、卡起来要人命的环节。销售催着要报价单技术要核成本财务要盯毛利老板要控风险一圈签字下来客户那边黄花菜都凉了。我所在的团队去年接手了一个内部改造项目目标很明确把原来靠邮件和表格流转的报价审批改造成由 LLM 加工作流引擎驱动的自动化流程。关键词里出现的 LLM、工作流引擎、LangGraph、FastAPI基本就是这个项目的技术底座。这篇东西不是产品宣传是我把整个从选型到上线、从踩坑到调优的过程完整复盘一遍给正在考虑用大模型改造企业内部流程的同行一个可参考的样本。不管你是刚接触 LangGraph 的新手还是已经在用 FastAPI 搭后端的老手应该都能从里面找到能直接抄作业的部分。1. 为什么报价审批值得用 LLM 加工作流引擎重做一遍1.1 传统报价审批到底卡在哪先说清楚痛点不然改造就是无的放矢。我们原来的报价审批流程大致是这样销售在 Excel 模板里填好客户信息、产品清单、折扣率然后邮件发给技术负责人核成本技术改完再转财务核毛利财务觉得没问题再转分管领导审批领导批完销售才能把报价单发出去。整个链路平均耗时两天半遇到领导出差能拖到一周。这里面有三个结构性问题。第一是信息在传递中衰减销售填的原始需求经过三四次转手技术看到的可能已经是残缺版本经常要回头问销售这个折扣是含税还是不含税。第二是判断标准不统一同样一个 15% 的折扣A 技术觉得能接B 技术觉得亏本全凭个人经验没有沉淀成规则。第三是审批动作本身没有增值财务和领导大部分时候只是在确认有没有明显异常真正需要人拍板的极端情况不到两成。我统计过我们过去半年的报价单大约 78% 的审批是标准通过也就是折扣在授权范围内、毛利高于红线、客户是存量客户。这 78% 完全没必要占用四个人的时间。真正需要人介入的是那些折扣超限、新客户大额、或者产品组合复杂的单子。这就给自动化留出了巨大空间。1.2 LLM 在这里扮演的到底是什么角色很多人一听用大模型改造审批第一反应是让模型直接决定批不批。这个思路我认为是错的至少在企业内控场景下非常危险。大模型有幻觉有随机性你不可能把几十万的报价决策交给一个可能编造理由的模型。我们给 LLM 的定位是信息抽取器加规则解释器而不是决策者。具体来说它干三件事第一把销售用自然语言写的需求描述比如这个客户是老客户介绍来的希望走个优惠价大概能接受八折左右解析成结构化的字段——客户类型、期望折扣、紧急程度。第二把非结构化的产品清单和历史报价做语义匹配找出相似项目作为参考。第三对触发人工审批的单子自动生成一份审批摘要把关键风险点用自然语言列出来让领导三十秒看懂而不是翻五页表格。决策权始终在规则引擎和授权矩阵手里。LLM 的输出只是给规则引擎提供更干净的输入以及给人提供更好读的上下文。这个边界划清楚后面所有的架构设计才不会跑偏。1.3 工作流引擎为什么不能省有人会问既然 LLM 只做抽取和解释那我写个 Python 脚本串起来不就行了为什么要上工作流引擎我一开始也这么想直到发现审批流程的本质是有状态、可中断、可回溯、可分支的。举个真实场景一个报价单提交后LLM 抽取完成规则引擎判定折扣超限需要技术总监审批。这时候流程要暂停等总监在系统里点同意或驳回可能等两小时也可能等两天。这期间流程状态必须持久化服务器重启不能丢。总监驳回后流程要回到销售节点让他修改修改完重新走一遍但历史记录要保留。如果总监三天没处理要自动升级给上级。这些需求用裸脚本写会变成一堆 if-else 和数据库轮询维护起来是灾难。LangGraph 这类工作流引擎提供的正是状态机加持久化加人工中断的能力。它把每个审批节点定义成图上的一个节点节点之间的流转由条件边控制整个图的执行状态可以 checkpoint 到数据库随时恢复。这才是企业级流程该有的样子。FastAPI 则负责把整个图包装成 HTTP 接口让前端和外部系统能调用。2. 技术选型LangGraph、FastAPI 和 LLM 各自的位置2.1 为什么是 LangGraph 而不是自己写状态机市面上的工作流方案大致分三类一类是 BPMN 引擎如 Camunda、Activiti功能全但重改一个节点要动 XML 配置和 Python 生态割裂一类是纯代码状态机如 transitions 库轻但缺持久化和人工中断支持第三类就是 LangGraph 这种面向 LLM 应用的图编排框架。我选 LangGraph 的核心理由是它原生支持人工中断interrupt和状态持久化checkpointer而且和 LLM 调用天然集成。它的 StateGraph 允许你定义一个 TypedDict 作为全局状态每个节点函数接收状态、返回状态更新条件边根据状态决定下一步走哪。最关键的是interrupt()函数可以让图在某个节点暂停把控制权交还给调用方等外部输入后再用Command(resume...)恢复。这正好对应审批流程里等人点按钮的场景。对比下来Camunda 适合纯人工流程但接入 LLM 要额外写 Java 服务transitions 库适合简单状态流转但持久化要自己实现。LangGraph 在这个场景里是甜点区。当然它也有代价后面踩坑章节会讲。2.2 FastAPI 承担的是哪一层职责FastAPI 在这个架构里是对外网关加流程调度器。它不参与流程逻辑本身只做四件事接收报价单提交请求、触发 LangGraph 图的执行、暴露审批操作接口、查询流程状态。为什么不用 Flask 或 DjangoFlask 异步支持弱Django 太重且 ORM 绑定深。FastAPI 的异步原生支持配合 LangGraph 的异步执行很顺Pydantic 模型做请求校验也省事。而且它的依赖注入系统很适合把数据库连接、LLM 客户端、图实例这些资源管理起来。一个典型的接口设计是这样的POST /quotes提交报价单返回流程 IDGET /quotes/{id}/status查状态POST /quotes/{id}/approve提交审批意见GET /quotes/{id}/summary拿 LLM 生成的审批摘要。前端只需要轮询状态或接 WebSocket 推送。2.3 LLM 选型本地还是云端这是个成本问题LLM 这块我们试过两条路。早期用云端 API抽取效果好、接入快但有两个问题一是报价数据涉及客户信息和价格走外部接口有合规顾虑二是量大之后 token 成本不低我们日均两百多单每单抽取加摘要大概消耗三千 token一个月下来是笔不小的开销。后来切到本地部署的开源模型用 Ollama 跑一个 7B 到 14B 级别的模型。抽取任务其实不需要太强的推理能力7B 模型配合好的 prompt 足够。本地部署的好处是数据不出内网成本固定坏处是要自己维护推理服务显存不够时并发上不去。我们的做法是混合路由常规抽取走本地模型遇到复杂语义比如客户描述特别绕才降级到云端。这个路由逻辑本身也是图上的一个条件边。提示LLM 选型不要一上来就追求最强模型。审批场景里 80% 的抽取任务是模式化的小模型加好 prompt 的性价比远超大模型裸跑。先把任务拆细再决定哪部分需要强模型。3. 用 LangGraph 把审批流程画成一张图3.1 状态设计整个流程的单一事实来源LangGraph 的核心是状态。我们定义了一个QuoteState作为全局状态所有节点读写它。这个状态设计得好不好直接决定后面顺不顺。我们的状态字段大致分四组原始输入组raw_quote_text销售填的原始描述、attachments附件引用、submitter_id。抽取结果组customer_type、expected_discount、product_items、urgency、extraction_confidence。规则判定组discount_within_limit、margin_above_redline、requires_manual_approval、approval_level。流程控制组current_node、approval_history、llm_summary、error。这里有个经验状态字段要扁平不要嵌套太深。LangGraph 的 checkpointer 会把状态序列化存库嵌套结构在版本升级时容易出兼容问题。我们一开始把产品清单做成嵌套的 list of dict后来发现每次改字段都要写迁移脚本索性拍平成几个并列的 list。另外extraction_confidence这个字段很关键。LLM 抽取不是百分百准我们让模型对自己的抽取结果给一个 0 到 1 的置信度。低于 0.7 的流程直接走人工复核分支不进入自动判定。这是防止模型瞎猜导致错误审批的第一道闸。3.2 节点划分每个节点只干一件事我们把整个流程拆成七个节点每个节点职责单一extract_node调 LLM 把原始文本抽成结构化字段。validate_node校验抽取结果的完整性和置信度。rule_check_node跑规则引擎判定折扣、毛利、客户类型。route_node根据规则结果决定走自动通过还是人工审批。auto_approve_node自动通过分支写审批记录。human_approval_node人工审批分支用interrupt()暂停等输入。finalize_node汇总结果生成最终报价单和审批摘要。节点拆得细的好处是每个节点可单独测试、可单独替换。比如后来我们换了 LLM只改extract_node其他节点纹丝不动。如果当初把抽取和校验写在一个大节点里换模型就得整体重测。节点之间用条件边连接。validate_node之后有个条件边置信度够就走rule_check_node不够就走human_approval_node并标记抽取存疑。route_node之后的条件边根据requires_manual_approval分流。这种显式的分支让流程一目了然比藏在代码里的 if-else 好维护太多。3.3 人工中断interrupt 的正确打开方式human_approval_node是整个流程里最需要小心处理的节点。它的逻辑是调用interrupt()暂停图执行把当前状态和需要审批的信息返回给调用方然后等外部通过Command(resume审批结果)恢复。这里有个坑我踩得很深。interrupt()恢复后节点函数会从头重新执行而不是从 interrupt 那行继续。这意味着 interrupt 之前的代码会再跑一遍。如果你在 interrupt 前调了 LLM 或者写了数据库恢复时会重复执行。正确做法是把所有副作用操作放在 interrupt 之后或者用幂等设计。我们的human_approval_node最终长这样先检查状态里有没有approval_result有就直接用说明是恢复执行没有才调interrupt()。这样无论执行几次结果都一致。这个模式我强烈建议所有用 LangGraph 做人工审批的人都用上。def human_approval_node(state: QuoteState): if state.get(approval_result): # 恢复执行直接使用已有结果 return {approval_history: state[approval_history] [state[approval_result]]} # 首次执行暂停等待人工输入 decision interrupt({ quote_id: state[quote_id], summary: state[llm_summary], risk_points: state[risk_points], }) return {approval_result: decision}3.4 持久化checkpointer 选型与状态恢复LangGraph 的 checkpointer 决定状态存哪。开发阶段用MemorySaver就行进程内内存重启即丢。生产必须换成持久化的官方提供SqliteSaver和PostgresSaver。我们选了 Postgres因为审批数据本来就要进主库复用现有实例省事。checkpointer 的配置有个细节thread_id必须唯一标识一个流程实例。我们用报价单 ID 加时间戳生成保证同一单子的多次提交是不同 thread。恢复时用graph.get_state(config)拿当前状态用graph.update_state()注入人工审批结果再graph.invoke(None, config)继续执行。实测下来 Postgres checkpointer 在并发五十个流程实例时表现稳定单次状态读写延迟在十毫秒级。如果并发再高可以考虑给 checkpointer 单独建库避免和业务查询抢连接。4. FastAPI 层把图包装成可用的服务4.1 项目目录结构怎么组织才不乱FastAPI 项目最容易写成一锅粥所有路由塞在 main.py 里。我们参考了社区里比较成熟的分层结构最终是这样quote_approval/ ├── app/ │ ├── main.py # FastAPI 实例和生命周期 │ ├── api/ │ │ ├── routes_quotes.py # 报价相关路由 │ │ └── routes_admin.py # 管理路由 │ ├── core/ │ │ ├── config.py # 配置加载 │ │ └── graph.py # LangGraph 图定义 │ ├── nodes/ # 各个图节点 │ ├── services/ │ │ ├── llm_client.py # LLM 调用封装 │ │ └── rule_engine.py # 规则引擎 │ ├── models/ # Pydantic 模型 │ └── db/ # 数据库连接和 checkpointer ├── tests/ └── pyproject.toml关键原则是图定义和路由分离。core/graph.py只负责编译图不关心 HTTP。路由层通过依赖注入拿到图实例调用graph.invoke()或graph.update_state()。这样图可以独立测试路由也可以 mock 图来测。4.2 提交接口与流程触发POST /quotes接口接收销售提交的报价单核心逻辑是构造初始状态、生成 thread_id、异步触发图执行。这里要注意图执行可能耗时LLM 抽取要几秒不能让 HTTP 请求一直挂着。我们的做法是提交后立即返回流程 ID图在后台任务里跑。router.post(/quotes) async def submit_quote(payload: QuoteSubmit, graphDepends(get_graph)): thread_id fquote-{payload.quote_id}-{int(time.time())} config {configurable: {thread_id: thread_id}} initial_state build_initial_state(payload) # 后台执行不阻塞响应 background_tasks.add_task(run_graph, graph, initial_state, config) return {thread_id: thread_id, status: processing}run_graph里捕获异常并写入状态避免后台任务静默失败。我们一开始没做异常捕获结果 LLM 超时导致流程卡死前端一直显示 processing排查了半天才发现是后台任务抛异常没人接。4.3 审批接口与状态恢复审批接口是POST /quotes/{thread_id}/approve接收审批人的决定和意见。核心是调graph.update_state()把审批结果写进状态然后graph.invoke(None, config)恢复执行。router.post(/quotes/{thread_id}/approve) async def approve_quote(thread_id: str, decision: ApprovalDecision, graphDepends(get_graph)): config {configurable: {thread_id: thread_id}} state graph.get_state(config) if not state.next: raise HTTPException(400, 流程不在等待审批状态) graph.update_state(config, {approval_result: decision.dict()}) result await graph.ainvoke(None, config) return {status: resumed, current_node: result.get(current_node)}state.next这个字段很关键它告诉你图当前停在哪个节点、是否在等待恢复。如果state.next为空说明流程已经结束或没在中断状态这时候调 approve 应该报错而不是硬塞数据。4.4 日志与可观测性uvicorn 日志丢失的坑FastAPI 配 uvicorn 跑起来后我发现自定义的 logger 输出经常丢尤其是后台任务里的日志。查下来原因是 uvicorn 自己配置了 logging把 root logger 的 handler 覆盖了。解决办法是在应用启动时显式配置 logging并且禁用 uvicorn 的默认配置。# main.py 启动时 logging.config.dictConfig(LOGGING_CONFIG) uvicorn.run(app, log_configNone) # 关键不让 uvicorn 覆盖另外 LangGraph 的执行过程建议开LANGCHAIN_TRACING之类的追踪把每个节点的进出状态记下来。审批流程出问题时能回放整个执行链路比看日志高效得多。我们后来接了一个轻量的追踪后端每个流程实例的节点耗时、LLM 调用参数和返回都留痕排查效率提升明显。5. 上线后踩过的坑和调优记录5.1 LLM 抽取不稳定从 prompt 到 schema 的加固上线第一周最头疼的是抽取结果不稳定。同一段文本模型这次抽出的折扣是 0.8下次变成 80%。原因是 prompt 里没约束格式模型自由发挥。我们做了三层加固。第一层是prompt 里给 few-shot 示例把三五个典型报价描述和对应结构化输出写进去让模型照着格式来。第二层是用 Pydantic 定义输出 schema配合结构化输出能力让模型直接返回 JSON 而不是自然语言。第三层是后处理校验折扣字段统一归一化到 0 到 1 区间产品名称做模糊匹配对齐到标准库。这三层下来抽取准确率从最初的七成出头提到九成五以上。剩下那百分之几的疑难单子靠置信度阈值兜底走人工。注意结构化输出不是万能的。有些模型对 schema 的支持不完整会返回带多余字段或缺失字段的 JSON。后处理校验必须做不能假设模型一定听话。5.2 流程卡死interrupt 恢复失败的排查链路有一次生产环境出现流程卡死前端显示审批中但审批人点同意后没反应。排查过程值得记录。第一步查数据库里该 thread 的 checkpoint发现状态停在human_approval_nodeapproval_result为空。第二步查审批接口日志发现 update_state 调用成功了但 ainvoke 恢复时报错。第三步看报错信息是状态里某个字段类型不匹配——审批意见我们前端传的是字符串但状态定义里approval_result期望的是 dict。根因是接口层没做严格的类型校验Pydantic 模型定义得太宽松。修复方案是收紧ApprovalDecision模型强制要求结构化字段并在 update_state 前做一次 schema 校验。这个坑告诉我们LangGraph 的状态字段类型必须严格任何宽松都会在恢复时爆炸。5.3 并发下的 checkpointer 连接池问题压测时发现并发到三十以上流程提交开始变慢偶尔报数据库连接超时。查下来是 checkpointer 和业务查询共用了一个连接池LangGraph 的状态读写比较频繁把连接占满了。解决办法是给 checkpointer 单独配一个连接池大小按并发流程数估算。我们的经验值是连接池大小约为峰值并发流程数的 1.5 倍。另外把 checkpointer 的写入改成批量提交减少事务开销。调整后并发一百个流程实例P99 延迟稳定在两百毫秒以内。5.4 成本与延迟的平衡本地模型加云端降级前面提过混合路由实际跑下来效果不错。本地 7B 模型处理常规抽取单次延迟约八百毫秒成本几乎为零。遇到置信度低于阈值或文本特别复杂的路由到云端模型延迟两秒左右但准确率高。路由的判定逻辑我们迭代过几版。最初只看文本长度后来发现长度不是好指标短文本也可能很绕。最终用本地模型的自评置信度加文本特征是否含否定词、是否有多个折扣条件综合判定。这个路由本身也是图上的一个条件边改起来很方便。6. 关于这套架构还能怎么扩展跑了大半年这套 LLM 加 LangGraph 加 FastAPI 的组合已经稳定支撑日均两百多单的报价审批人工介入率从原来的接近百分之百降到两成左右。回头看最有价值的不是某个具体技术而是把 LLM 的边界划清楚——它做它擅长的语义理解和文本生成决策和状态管理交给确定性的工作流引擎。如果要在类似场景复用这套架构我的建议是先别急着写代码把流程画成图标出哪些节点需要 LLM、哪些需要人工中断、哪些是纯规则。图画清楚了LangGraph 的代码基本就是照着图翻译。另外状态设计要一次到位字段类型严格后面能省掉大量调试时间。这套东西往其他审批场景迁移也很自然比如合同审批、采购申请、费用报销流程骨架几乎一样换的是抽取字段和规则。我们内部已经在把报销审批往同一个框架上搬节点复用率能到七成。真正花时间的永远是业务规则的梳理和 LLM prompt 的调优技术框架本身反而是最省心的部分。