企业官网建设流程全解析

做供应链风控这几年，我最大的感受是：规则引擎够稳但不够聪明，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, schema=None, temperature=0)。
• 场景路由：抽取类任务走便宜快速的小模型，复杂推理走强模型。这个路由写在配置里，可以根据线上效果调整。
• 调用日志：记录模型名、输入 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 节点会检查：如果存在action=REJECT的命中，直接返回拒绝；否则继续看是否有 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 到 100；edge_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(decision=state["final_decision"], risk_score=state["risk_score"], request_id=state["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 做 Demo，FastAPI 做生产

很多开源项目喜欢用 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 输出不一致：同样的输入得到不同结果

风控场景最怕模型"看心情"：同一个供应商资料，上午查没问题，下午查就多出一个关联信号。排查下来，原因往往是：

• 没设temperature=0。虽然不保证完全稳定，但必须设为 0。
• 模型被随机采样影响。换用确定性更高的解码参数，比如top_p设为 1，禁用随机采样。
• prompt 里缺少引用原文的要求，模型开始自由发挥。

我的经验是：在 prompt 里强制要求"每个结论附带原文引用"，比单纯调参数更能稳定输出。模型一旦必须引用原文，它就被限制在已有文本里，编造空间小很多。

6.3 Token 超限：合同全文塞进 LLM 必死

供应链风控里，一个采购合同动辄几十页。直接全文塞给 LLM，token 直接爆。我的处理思路分三层：

• 先让规则引擎做字段提取，把地址、法人、金额、日期这些关键字段摘出来，这些就是模型的"锚点"。
• 如果还需要全文理解，做切片，每段控制在 1500 token 左右，逐段抽取，最后合并。
• 把历史案例向量化，做检索增强，只把相似案件的摘要给模型参考，而不是把所有资料都塞进去。

这套做法其实跟 RAG 的路线一致。区别是，风控场景里检索的召回质量直接影响决策，所以我会在检索结果里额外要求附上原文位置，方便审计复核。

6.4 规则引擎和 LLM 判断冲突怎么办

线上跑久了必然遇到这种单子：规则引擎说通过，LLM 却报出高风险信号。冲突处理规则是我在系统设计阶段就跟业务对齐的：

规则引擎的 REJECT 是最高优先级，任何情况下不可被 LLM 覆盖。规则引擎 REVIEW 加 LLM 低置信度，按通过处理但留痕。规则引擎 APPROVE 加 LLM 高置信度，升级为 REVIEW，转人工。

这个策略背后的逻辑是：规则引擎代表已知风险，LLM 代表未知风险。已知风险必须拦截，未知风险可以放行或人工复核，但不能自动拦截。否则哪天模型抽风，把一个白名单供应商拦了，商业上损失很大。

做这套系统的过程中，我最大的体会是：LLM 在风控里的定位不是"决策者"，而是"信号源"。它负责从非结构化的文本里挖出规则引擎看不到的关联信号，但最终决策仍然要由规则引擎和人的流程来兜底。可审计性是这个设计里最重要的关键词——模型可以出错了，但出错的记录必须完整保留，这样系统才能持续改进。另外给所有节点加上版本号，重放时就能清楚知道是哪一步的哪一版产生了不同结果，这个做法强烈推荐。