报价审批这件事,放在任何一家做项目制生意的公司里,都是那种"看起来不起眼、卡起来要人命"的环节。销售催着要报价单,技术要核成本,财务要盯毛利,老板要控风险,一圈签字下来,客户那边黄花菜都凉了。我所在的团队去年接手了一个内部改造项目,目标很明确:把原来靠邮件和表格流转的报价审批,改造成由 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 或 Django?Flask 异步支持弱,Django 太重且 ORM 绑定深。FastAPI 的异步原生支持配合 LangGraph 的异步执行很顺,Pydantic 模型做请求校验也省事。而且它的依赖注入系统很适合把数据库连接、LLM 客户端、图实例这些资源管理起来。
一个典型的接口设计是这样的:POST /quotes提交报价单,返回流程 ID;GET /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, graph=Depends(get_graph)): thread_id = f"quote-{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, graph=Depends(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_config=None) # 关键:不让 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_node,approval_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 的调优,技术框架本身反而是最省心的部分。