做供应链风控这几年我最大的感受是规则引擎够稳但不够聪明LLM够聪明但不够稳。今年上半年我开源了一个用 FastAPI LangGraph 搭的供应链风控服务核心思路是把 LLM 和规则引擎串成一条可审计的流水线让两者互相兜底。简单说就是用 LangGraph 编排节点FastAPI 暴露接口规则引擎负责守住硬性红线LLM 负责挖掘规则写不出来的语义风险每个节点的输入输出全部落审计日志既能追溯又能重放。这篇文章就把整个系统的设计思路、目录结构、核心流水线实现、选型对比以及我踩过的坑完整写一遍。适合两类人看一类是正在做风控、审核、合规系统想引入 LLM 但又怕它“胡说八道”的另一类是 LangChain 用得差不多了想搞明白 LangGraph 到底比 Chain 强在哪的。看完你至少知道怎么从零搭一个可审计的 LLM 风控流水线。1. 为什么要把 LLM 和规则引擎串成一条流水线1.1 传统规则引擎的短板我在公司里最早接触的风控系统本质上就是一堆 if-else 加配置文件。供应商名称命中黑名单就拦截金额超过阈值就转人工证件过期就驳回。这套东西用 Drools、Easy Rules 或者自研的规则脚本都能实现优点是延迟低、执行结果完全可预期、出问题了好解释。但它最大的问题在于规则只能识别写得出来的风险。我跟业务同事聊天时经常听到这样的抱怨——这家供应商跟之前被拉黑的那家注册地址和法人电话一模一样为什么系统没拦因为规则引擎比对的是字符串它不知道地址相似法人关联股权穿透这些语义层面的东西。就算你把规则写得再细也只能覆盖建模时能想到的模式而灰产和关联方风险恰恰是藏在文本和关系网络里的。而且规则引擎还有一个隐性成本规则膨胀。上线两年之后规则文件能有几千条互相之间还有优先级冲突。改一条规则要拉着业务、法务、技术三方开会最后谁都不敢动。这时候你自然会把目光投向 LLM。1.2 LLM 单打独斗的问题我最早做技术验证的时候想法很简单把所有供应商资料直接丢给 GPT让它输出有没有风险。Demo 效果确实惊艳给它几张发票和合同它能给你说出很多疑似关联信号比如两家公司共用一部联系电话注册地址在同一层楼这些。但一旦进入生产环境问题就全冒出来了。首先是幻觉。同一个供应商资料你问三次它可能给你三个不同的风险结论每次还都振振有词。风控是要做拦截决策的你没法拿一个这次说有问题、下次说没问题的模型去做最终判断。其次是不可审计。LLM 的结论没有可复现性也没有清晰的证据链。业务被拦截之后一定会问凭什么拦这一单依据是什么你要是回答大模型觉得有问题那这个系统是上不了线的。再次是成本。供应链风控的请求量跟订单量挂钩高峰期每秒可能有几十上百个审核请求不可能每一个都调用大模型。实测下来一个中等规模的抽取任务一次调用就要几百上千 token全走 LLM 成本完全不可控。1.3 流水线设计让 LLM 探边界让规则守底线所以问题不是用规则还是用 LLM而是怎么把两者组织成一个整体。我最后定的方案是LLM 和规则引擎在流水线里各有明确分工谁也不替代谁。规则引擎做前置过滤和高置信度拦截。比如黑名单命中、制裁名单命中、金额超过硬阈值这些直接拦截不需要 LLM 参与。规则引擎的输出是布尔值加规则编号完全可解释。LLM 做规则引擎覆盖不到的语义抽取和关联信号发现。它不直接下拦截结论只输出结构化的事实和置信度比如检测到供应商 A 与黑名单企业 B 的注册地址相似置信度 0.87。这些信号会进入下游的风险评分也可能触发人工审核。规则引擎在 LLM 之后再做一次复检。这一步用来防止 LLM 跑偏。比如 LLM 从文本里抽取了一个疑似法人实体规则引擎会把这个人名拿去比对历史黑名单命中了就要求转人工。这样即使 LLM 有幻觉最终决策仍然被规则约束。而把这些节点串成流水线的就是 LangGraph。它让每个节点变成一个图里的步骤状态显式传递节点之间支持条件分支和循环天然适合先规则、再 LLM、再规则这种流程。2. 系统整体设计与目录结构2.1 FastAPI 项目目录结构怎么拆我的项目目录结构是从 fastapi 官方模板改出来的经过几个版本迭代最后长这样supply_risk_control/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── core/ │ │ ├── config.py # 环境变量/配置项 │ │ └── logging.py # 审计日志配置 │ ├── api/ │ │ └── v1/ │ │ ├── router.py # 路由汇总 │ │ ├── endpoints/ │ │ │ └── risk.py # /api/v1/risk/check 接口 │ │ └── schemas.py # Pydantic 请求/响应模型 │ ├── graph/ │ │ ├── state.py # LangGraph 状态定义 │ │ ├── nodes/ │ │ │ ├── preprocess.py │ │ │ ├── rule_engine.py │ │ │ ├── llm_analyze.py │ │ │ ├── rule_recheck.py │ │ │ ├── decision.py │ │ │ └── audit.py │ │ ├── pipeline.py # 状态图构建 │ │ └── edges.py # 条件边定义 │ ├── rules/ │ │ ├── engine.py # 规则引擎执行器 │ │ └── rulesets/ │ │ ├── blacklist.yaml │ │ ├── threshold.yaml │ │ └── relation.yaml │ ├── services/ │ │ ├── llm_gateway.py # LLM 网关 │ │ ├── vector_store.py # 历史案例向量库 │ │ └── audit_api.py # 审计日志写入 │ └── models/ │ └── risk_models.py # ORM 模型 ├── tests/ ├── docker/ └── pyproject.toml这样拆的原因很简单API 层只负责解析请求和返回响应graph 层负责流水线编排rules 层负责规则定义services 层负责外部依赖。每一层都可以单独测试替换实现也方便。比如你不想用我这种 YAML 规则格式可以保留接口、换掉 engine.py 内部实现不影响流水线其他节点。目录结构里最容易被忽略的是app/core/config.py。风控系统最怕配置散落各处模型名称、超时时间、阈值、黑名单版本号我都集中放在配置里并且每次跑批时把配置版本写入审计日志。这样将来排查为什么同一单上次放行、这次拦截时可以直接对比两次请求的配置版本。这个习惯让我少踩了很多坑。2.2 LangGraph 状态图把流水线变成有状态的图LangGraph 的核心概念我理解下来就是三个State、Node、Edge。State 是节点之间传递的数据对象Node 是处理函数Edge 决定下一步走哪个节点。我在graph/state.py里定义的状态长这样from typing import TypedDict, List, Optional class RiskState(TypedDict): request_id: str supplier_id: str raw_docs: List[dict] # 原始单据合同、发票、注册资料 rule_hits: List[dict] # 规则引擎命中记录 llm_extractions: List[dict] # LLM 抽取的结构化事实 llm_confidence: float rule_recheck_hits: List[dict] # 复核命中记录 risk_score: float final_decision: str # APPROVE / REJECT / REVIEW audit_trail: List[dict] # 审计轨迹每个节点接收整个 state 的当前值返回一个字典对 state 做局部更新。比如规则引擎节点只负责追加rule_hits不关心其他字段。这种设计保证了节点之间低耦合也方便调试你随时可以把某个节点的入参和出参单独拎出来跑一遍。图的结构我用的是条件边如果规则引擎在预筛阶段命中了硬拦截规则就直接进入 decision 节点跳过 LLM否则走到 LLM 分析节点。LLM 分析完成后无论结果如何都要经过规则复核最后进入决策节点。这样设计是为了审计一致性——即使被硬规则拦截这条请求也有一条完整的流水线轨迹。LangGraph 有一个很实用的能力是 checkpoint。开箱之后每个节点的输入输出都会被自动记录下来支持从任意节点恢复执行。我后面做一键重放功能时发现这远比自己在外面包一层日志要省事因为 checkpoint 保留了完整的状态历史而不是只有最终结果。2.3 LLM 网关不让模型调用散落得到处都是如果直接在 llm_analyze 节点里写openai.ChatCompletion.create(...)那后续会非常痛苦。模型要切换、要加缓存、要做日志、要限流散落在各个节点根本没法维护。所以我单独做了一层llm_gateway.py所有模型调用都走这里。网关做了几件事统一接口不管底层是 OpenAI、通义、还有本地部署的 vLLM对外都暴露一个async def complete(messages, schemaNone, temperature0)。场景路由抽取类任务走便宜快速的小模型复杂推理走强模型。这个路由写在配置里可以根据线上效果调整。调用日志记录模型名、输入 token 数、输出 token 数、耗时、返回码。缓存对完全相同的输入直接返回历史结果实测能省掉 30% 左右的调用量。风控系统对模型请求还有一个额外要求必须做 schema 校验。我会要求模型按 JSON Schema 返回结构化数据网关拿到结果后会先做一次格式校验解析失败就自动重试一次还失败就把错误信息写进审计日志。这一步比在业务代码里 try-except 要干净得多。3. 核心流水线实现从一条风控请求说起3.1 第一步规则引擎预筛请求从 FastAPI 进来后先经过 preprocess 节点生成 request_id、标准化供应商名称、把各种单据转成统一结构。然后进入规则引擎预筛。预筛阶段跑三类规则黑名单规则、阈值规则、强制字段规则。黑名单规则是最简单的字符串精确匹配阈值规则比如单笔采购金额超过 500 万必须人工复核强制字段规则比如供应商缺少营业执照编号无法通过。规则我放在 YAML 文件里管理rules: - id: BLACKLIST_NAME_EXACT type: BLACKLIST field: supplier_name operator: IN value: blacklist_name action: REJECT priority: 100 - id: AMOUNT_OVER_REVIEW_LIMIT type: THRESHOLD field: invoice_amount operator: value: 5000000 action: REVIEW priority: 90 - id: LICENSE_MISSING type: REQUIRED_FIELD field: license_no operator: IS_EMPTY action: REJECT priority: 80这里有个经验规则引擎只做两类判断——直接拒绝和转人工不做放行。因为预筛阶段还没走 LLM覆盖面有限除非是完全可信的白名单否则不要轻易放行。宁可让它多转人工也不要漏掉风险。规则引擎执行完之后state 里的rule_hits就有数据了。decision 节点会检查如果存在actionREJECT的命中直接返回拒绝否则继续看是否有 REVIEW 命中。有 REVIEW 命中则走 LLM 节点做深度分析这也是流水线的一个分支点。3.2 第二步LLM 语义抽取与风险信号发现LLM 节点的任务不是让模型给结论而是让模型做事实抽取。我把这个设计原则写在项目 README 里LLM 只报告它看到了什么不报告应该怎么处理。具体做法是设计一套严格的 JSON Schema要求模型抽取以下字段供应商注册地址、实际经营地址法人代表、主要股东、联系电话与其他供应商的潜在关联通过地址、电话、法人重合异常信号比如营业执照地址与发票地址不一致成立时间与业务规模明显不匹配每个信号都必须附上置信度和原文引用为了让模型输出更稳定我在 prompt 里要求先引用原文再给出判断。也就是说模型回答里必须出现类似根据合同第 3 条注册地址为 XX与发票地址不同这样的数据来源。这个约束能显著减少幻觉因为模型被强制引导去检索文本中的证据。如果企业资料很多、超过上下文窗口我会先把重要字段用规则引擎做一次定位——比如通过正则把注册地址法人这些字段直接从文档里摘出来——再交给 LLM。实在需要全文理解时才把文本切片后分段抽取再合并结果。我在项目里还加了一个可选的历史案例向量检索模块做法类似 RAG/GraphRAG把过去人工审核确认过的风险案例建索引LLM 分析前先召回相似案例把案例摘要作为参考上下文。这块本质上就是llm wiki 知识库思路的变体让模型不靠记忆而是靠检索。3.3 第三步规则复核与置信度融合LLM 抽取出的信号不能直接使用要经过规则复核节点。复核有三件事第一把 LLM 抽取的实体拿去跟黑名单企业库比对第二做实体归一化——比如地址里的室和房、电话里的区号差异先归一化再比较第三跨供应商关联检测看同一个人名下挂着多少家正在合作的供应商。复核完成后进入风险评分阶段。这个分数是规则引擎和 LLM 信号融合的结果我用的公式比较简单risk_score max(rule_penalty, llm_signal_score) edge_count_penalty其中rule_penalty来自规则引擎的命中项硬规则直接给 100 分llm_signal_score是 LLM 信号中最高置信度对应的分值0 到 100edge_count_penalty是关联边的数量乘以 5。举个例子规则引擎没命中0 分LLM 发现一个关联地址信号置信度 0.87折算 87 分同时这家供应商与 2 家黑名单企业存在关联加 10 分最终 97 分。超过 90 分走人工复核不是直接拒绝。为什么因为 LLM 的信号可能是误判直接拒绝会产生客诉风险。我在系统里明确区分硬规则可以自动拒绝LLM 信号只能升级为 REVIEW。这是整个系统最重要的安全设计。3.4 第四步审计日志与一键重放每个节点结束时都会往audit_trail里追加一条记录包括节点名、版本号、输入摘要、输出摘要、耗时、模型名、token 数、异常信息。最后统一写入审计表。审计表结构上更像一个时序记录{ request_id: req_20240601_0001, nodes: [ { node: rule_engine_precheck, version: ruleset_v20240501, input_digest: sha256:..., output: {rule_hits: [BLACKLIST_NAME_EXACT]}, duration_ms: 12 }, { node: llm_analyze, version: qwen-plus-20240501, input_digest: sha256:..., output: {signals: [...]}, duration_ms: 1280, token_usage: {prompt: 3200, completion: 512} } ] }因为 LangGraph 的 checkpoint 帮我保存了每个节点执行时的完整状态我只需要把 checkpoint 和审计表里的事件关联起来就能实现一次重放——把某条历史请求从头再跑一遍对比本次结果与历史结果的差异。这个功能在应对供应商申诉、监管质疑时特别管用。4. LangGraph 与 LangChain 的区别以及我踩过的选型坑4.1 工具箱 vs 编排器网上聊 langchain 和 langgraph 的区别时最常见的一个说法是LangChain 是工具箱LangGraph 是编排器。这个类比我觉得挺准确。LangChain 给你一大堆工具LLM 封装、Prompt Template、Loader、Output Parser、Chain。它确实能帮你快速把模型跑起来但 Chain 的组织方式是线性的一个接一个。一旦你要做如果 A 命中就走 B否则走 C然后无论结果都要回到 D用 LangChain 的 Chain 写起来就很别扭最后会变成一个大函数里嵌套无数条件分支可读性和可维护性都很差。LangGraph 的视角不一样。它让你先画一张图节点是函数边是连接关系节点之间共享状态。流程图怎么画代码就怎么写。风控流水线天生就是一张图所以我最终选了 LangGraph。如果你只是做一个读文档-总结-发邮件的简单任务LangChain 完全够用没必要上 LangGraph。4.2 为什么风控系统必须用图而不是 Chain风控流水线有几个特点是 Chain 处理不好的。第一是条件分支。上面说过预筛命中规则就跳过 LLM。这个分支如果在 Chain 里你要么用ConditionalChain要么在函数里写一堆 if-else。在 LangGraph 里只是一个add_conditional_edges。第二是重试和恢复。LLM 经常超时或者返回格式错误。如果流水线中间一个节点挂了我希望整个请求能保存状态修复后从失败的节点继续跑而不是从头来过。LangGraph 的 checkpoint 天然支持恢复。Chain 要做这事得自己存中间结果代码量立刻上去。第三是子图复用。后期我加了一个人工审核结果回写功能需要把人工标注的数据重新灌回图里做模型微调样本这种图套图的结构Chain 很难优雅表达。当然 LangGraph 也有学习成本。它的状态结构、节点函数的签名、条件边的返回值规则刚开始很容易写错。我的建议是先画图把节点和边写清楚再动手写代码。流程图就是最好的注释。4.3 工具调用与 planning 模式我到底怎么用LangGraph 的工具调用可以理解为LLM 在回答过程中需要外部信息时就输出一个调用某个工具的结构化请求系统去执行工具把结果返回给模型继续推理。我在流水线里接了几个工具企业工商查询、历史案件查询、供应商关联图谱查询。刚开始我把所有工具一股脑塞给模型结果非常不稳定。模型经常在不需要查工商信息的时候也乱调用耽误时间。后来改成 planning 模式先给 LLM 一个规划器角色让它决定当前这个供应商资料需要调用哪些工具再让工具调用节点去执行。这其实就是 langgraph 教程里常说的 planner 和执行器分离。实测下来改成这样之后平均调用工具次数从 3-4 次降到了 1-2 次响应时间下降一半。而且审计日志里能清楚看到模型先决定查什么然后工具返回了什么最后模型基于这些信息做出了什么判断整条推理链路清清楚楚。5. FastAPI 接口层设计与生产化细节5.1 异步接口、线程池与并发调优FastAPI 是异步框架但你不一定所有代码都要写成async def。这里有个很容易踩的坑LangGraph 的节点默认是同步函数如果在 async 端点里直接调用同步节点会阻塞事件循环。我最终的做法是FastAPI 端点用async def接收请求但把整条 LangGraph 流水线放在线程池里运行然后通过asyncio.to_thread等待结果。规则引擎的计算是 CPU 密集LLM 网关的调用是 IO 密集两者混在一起时统一交给线程池比强行全异步要省心。虽然会损耗一点点性能但代码清晰很多。接口层还对内做了瘦身对外暴露两个端点router.post(/risk/check) async def risk_check(payload: RiskCheckRequest): state await asyncio.to_thread(run_pipeline, payload) return RiskCheckResponse(decisionstate[final_decision], risk_scorestate[risk_score], request_idstate[request_id]) router.get(/audit/{request_id}) async def get_audit(request_id: str): return await audit_api.get_trail(request_id)run_pipeline是 LangGraph 图的编译入口内部维护状态、执行节点、写审计日志。接口层只做数据校验和状态转换不碰业务逻辑。5.2 Gradio 做 DemoFastAPI 做生产很多开源项目喜欢用 Gradio 搭一个交互界面演示效果。我也做了但要把 Gradio 和 FastAPI 的边界想清楚Gradio 适合业务方小范围试跑、给领导演示、看 prompt 效果但它不是生产 API 的替代品。我见过有人直接把 Gradio 部署到生产环境接受请求结果并发一上来就卡死。正确的做法是Gradio 只作为旁路 Demo内部仍然调用 FastAPI 暴露的接口而不是自己直接执行流水线。这样你调 prompt、调规则、调模型时业务方在 Gradio 上看到的跟生产接口跑的是同一套逻辑不会有Demo 跑得好、生产跑飞了的尴尬。FastAPI 这边我额外加了三个中间件全局限流每 IP 每秒最多 10 次、请求体大小限制、统一响应结构。5.3 超时、重试与降级LLM 调用是流水线里最不稳定的环节。我所有模型调用都做了三层保护超时首次请求最多等 30 秒。重试超时或返回 5xx 时重试最多 2 次重试间隔递增。降级如果重试仍然失败流水线不报错而是把这条请求标记为 REVIEW 转人工同时写审计日志。这个降级设计是我被生产事故教育出来的。之前有一个版本LLM 网关一挂整个风控接口跟着 5xx业务方直接打爆电话。后来我意识到风控系统的核心目标是不放走重大风险在 LLM 不可用的时候宁可让所有请求先走规则引擎、再交给人工也不能让接口直接失败。规则引擎本身就是最基础的兜底能力。6. 常见问题与排查技巧实录6.1 LLM request failed: provider rejected the request schema or tool payload这个报错我在开发期遇到不下十次几个典型原因模型版本不支持 function calling但你强行传了 tools 参数。JSON Schema 里用了 provider 不认的关键字比如minItems或者某些内部引用的$ref。schema 嵌套层级太深超过 provider 的限制。排查方法也简单先在网关上把本轮请求的完整 payload 拿到手去掉 tools 参数只保留纯文本看能否正常返回然后逐渐把 schema 加回去二分定位问题字段。如果确实要传 schema我建议用stop参数配合结构化输出模式或者干脆去掉 tools改在 prompt 里让模型输出 JSON再用json.loads解析。后者虽然不严格但兼容性最好。6.2 LLM 输出不一致同样的输入得到不同结果风控场景最怕模型看心情同一个供应商资料上午查没问题下午查就多出一个关联信号。排查下来原因往往是没设temperature0。虽然不保证完全稳定但必须设为 0。模型被随机采样影响。换用确定性更高的解码参数比如top_p设为 1禁用随机采样。prompt 里缺少引用原文的要求模型开始自由发挥。我的经验是在 prompt 里强制要求每个结论附带原文引用比单纯调参数更能稳定输出。模型一旦必须引用原文它就被限制在已有文本里编造空间小很多。6.3 Token 超限合同全文塞进 LLM 必死供应链风控里一个采购合同动辄几十页。直接全文塞给 LLMtoken 直接爆。我的处理思路分三层先让规则引擎做字段提取把地址、法人、金额、日期这些关键字段摘出来这些就是模型的锚点。如果还需要全文理解做切片每段控制在 1500 token 左右逐段抽取最后合并。把历史案例向量化做检索增强只把相似案件的摘要给模型参考而不是把所有资料都塞进去。这套做法其实跟 RAG 的路线一致。区别是风控场景里检索的召回质量直接影响决策所以我会在检索结果里额外要求附上原文位置方便审计复核。6.4 规则引擎和 LLM 判断冲突怎么办线上跑久了必然遇到这种单子规则引擎说通过LLM 却报出高风险信号。冲突处理规则是我在系统设计阶段就跟业务对齐的规则引擎的 REJECT 是最高优先级任何情况下不可被 LLM 覆盖。规则引擎 REVIEW 加 LLM 低置信度按通过处理但留痕。规则引擎 APPROVE 加 LLM 高置信度升级为 REVIEW转人工。这个策略背后的逻辑是规则引擎代表已知风险LLM 代表未知风险。已知风险必须拦截未知风险可以放行或人工复核但不能自动拦截。否则哪天模型抽风把一个白名单供应商拦了商业上损失很大。做这套系统的过程中我最大的体会是LLM 在风控里的定位不是决策者而是信号源。它负责从非结构化的文本里挖出规则引擎看不到的关联信号但最终决策仍然要由规则引擎和人的流程来兜底。可审计性是这个设计里最重要的关键词——模型可以出错了但出错的记录必须完整保留这样系统才能持续改进。另外给所有节点加上版本号重放时就能清楚知道是哪一步的哪一版产生了不同结果这个做法强烈推荐。