从交易台到临床试验垂直 AI 跨行业复用的技术底层逻辑如果你团队负责交易监控每天要面对几万条行情信号、新闻流和异常告警模型给出的判断一旦出错带来的可能就是直接的资金损失如果你是临床试验数据团队每天要处理海量的受试者数据、不良事件记录和方案偏离报告一次漏报或误判就可能影响药品审批进度甚至威胁受试者安全。这两个场景隔行如隔山但你大概率会发现它们需要的 AI 系统长得很像数据先被清洗成结构化事件模型在事件流里挑出可疑候选规则引擎把明确的风险拦下来其余交给人工复核最后所有决定留下审计日志。最近大家讨论的 Vertical AI真正吸引人的不是某一个行业爆款应用而是这种领域数据管道 模型能力 业务决策工作流的组合正在金融、医疗、制造等不同行业里反复出现。这篇文章想做的事情也很具体把交易台与临床试验放在一起对比拆出 Vertical AI 落地的共性技术架构并给出你可以直接复用的代码思路。读完你会理解为什么垂直 AI 不是简单地把 ChatGPT 包装成本行业助手也会知道一个垂直 AI 项目从 POC 走向生产环境最该先做哪三件事。1. 垂直 AI 不是行业版 ChatGPT过去两年所有人都见识过通用大模型的文本生成、摘要和问答能力。于是很多团队产生了同一种冲动用企业私有文档做一次 RAG再套上一个行业界面就宣称自己做出了医疗大模型金融大模型法律大模型。从材料和我们观察到的实际案例来看这类产品多数停留在演示层面。原因在于垂直 AI 的核心价值不在能聊领域话题而在能介入业务决策。我们可以这样区分行业问答助手输入是一段自然语言问题输出是一段文本答案。用户自己判断答案对不对模型并不对业务结果负责。垂直 AI 系统输入是真实业务数据流输出是一个可执行的业务动作比如标记该事件为严重不良事件候选暂停该交易信号推送生成需要人工复核的监查报告。动作执行错了系统要承担责任。这中间的差距不是模型能力提升就能填平的而是缺少一层关键的业务结构。垂直 AI 系统必须知道事件是什么、状态怎么流转、哪个环节必须人工看、哪个环节可以自动过、决策依据有没有留存。这些结构来自对业务过程的数字化而不是来自参数更多的大模型。所以我的判断是垂直 AI 的本质是把某个领域的业务过程重新建模再把模型能力嵌入到这个过程的特定环节里。模型是发动机但整车是业务过程本身。如果一个团队只盯着模型参数却说不清楚自己的业务流程有哪些状态和节点那么离真正的垂直 AI 还很远。2. 交易台与临床试验两个行业里藏着的同一个 AI 架构交易台和临床试验表面上毫无关系但如果我们抽象到决策系统层面它们的相似度非常高。先看交易台。交易员面前有行情报价、宏观经济数据、新闻舆情、历史订单流系统要回答的问题是当前有没有异常交易机会要不要提醒交易员该不该自动执行风险控制动作整个过程里速度很重要、错误代价很高、事后必须能回溯每一笔决策依据。再看临床试验。数据监查员面前有电子病例报告表eCRF、实验室检查结果、不良事件记录、方案偏离记录系统要回答的问题是这名受试者有没有出现新的严重不良事件这个事件和试验药物的关系能不能初步判断需不需要在 24 小时内向监管方上报同样速度影响患者安全错误代价更高事后审计要求甚至比金融更严格。把两者放在一起对比能看到完全同构的结构维度交易台临床试验对 AI 系统的要求输入数据行情、订单、舆情、历史交易受试者数据、检验指标、不良事件、方案记录数据管道能处理高噪声、多来源的流式输入决策类型提醒、拦截、自动执行标记、初判、上报、生成报告模型只负责候选生成最终决策权可控失败成本资金损失、合规罚款受试者伤害、审评失败必须有风控规则和人工兜底合规约束交易留痕、监管报送审计追踪、电子记录合规每个判断都要能还原、可审计专家稀缺经验丰富的交易员成本极高医学监查员培养周期长AI 做初筛专家处理疑难这张表背后其实是 Vertical AI 最典型的价值区间数据信息密度足够高人工处理不过来决策成本高值得用系统兜底领域专家稀缺AI 可以把专家精力从重复劳动里释放出来。如果你准备在自己的行业里推动垂直 AI可以先拿这张表来对照。一个业务要成为垂直 AI 的合适场景至少满足三个条件有持续产生的结构化潜力很高的原始数据有明确的决策动作或者提醒动作有清晰的责任边界和审计要求。三者缺一项目后期大概率会陷入模型很酷但没有业务价值的困境。3. 垂直 AI 的四层技术栈既然结构相似落地时采用的技术栈也可以提炼成四层。以下四层是我在整理各种行业实践后归纳出的一套通用框架适合作为团队讨论和架构设计的起点。第一层领域数据管道。负责接入原始数据完成解析、清洗、去重、脱敏、字段标准化最终输出统一的领域事件。交易台里的行情快照和临床试验里的实验室检验记录格式完全不同但经过这一层后输出的都是带时间戳、事件类型、标准化字段的通用事件对象。第二层模型能力层。这一层包括基座模型、领域知识增强RAG、可选微调、规则引擎和打分模型。实际应用中很多判断不需要每次都调用大模型规则引擎可以解决大部分确定性逻辑模型负责处理语义理解和模糊判断。两者结合才能兼顾成本和效果。第三层业务工作流层。这是垂直 AI 区别于通用 AI 助手的核心。业务上有明确流程候选生成以后要过什么规则、哪些情况要强制人工、人工审核超时了怎么办、审批通过以后触发什么下游动作。这一层必须用工作流引擎或状态机落地保证流程可以被测试、被回放。第四层反馈闭环层。模型上线后不能只是跑了就行要把人工复核的结果回传形成新的标注数据定期重训或微调模型同时用影子模式持续对比模型建议和实际决策量化模型价值。这四层里最容易低估的是第一层和第三层。绝大多数垂直 AI 项目失败不是模型效果不够好而是数据进不来、流程接不上。接下来的内容我会用代码把这四层的关键部分串起来。4. 第一层领域数据管道先把野外数据变成标准事件假设我们要同时支持一个交易信号场景和一个临床试验不良事件场景。先不关心它们各自的数据库结构而是定义一个通用的标准事件模型。# vertical_ai_pipeline.py 垂直 AI 数据管道最小骨架 原始记录 - 规则校验 - 脱敏 - 标准事件 在交易台和临床试验场景中只需要替换数据源接入层和字段映射规则。 from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional dataclass class RawRecord: source: str # 数据来源market_feed / edc_form / lab_report raw_payload: Dict[str, Any] # 原始字段 received_at: datetime field(default_factorydatetime.utcnow) dataclass class StandardEvent: event_id: str # 稳定事件ID用于全链路追踪 event_type: str # 如 order_anomaly / adverse_event occurred_at: datetime # 业务发生时间不是接收时间 normalized_fields: Dict[str, Any] # 标准化后的字段 raw_source: str # 追溯原始记录 data_grade: str original # original / modified / obfuscated def validate_raw_record(record: RawRecord, required_fields: List[str]) - List[str]: 最简单的字段完整性校验返回缺失字段列表。 missing [f for f in required_fields if f not in record.raw_payload] return missing def obfuscate_payload(payload: Dict[str, Any], sensitive_keys: List[str]) - Dict[str, Any]: 对敏感字段做脱敏。生产环境建议使用可逆脱敏并保存映射便于审计回查。 safe {k: v for k, v in payload.items() if k not in sensitive_keys} for key in sensitive_keys: if key in payload: # 这里用简单哈希示意生产环境应使用密钥管理的脱敏算法 safe[key] fREDACTED:{hash(payload[key]) % 10**8:08d} return safe def to_standard_event(record: RawRecord, event_type: str, mapping: Dict[str, str], sensitive_keys: List[str], occurred_at_key: str) - Optional[StandardEvent]: mapping 把原始字段名映射为标准化字段名。 例如交易场景 {symbol: ticker, qty: quantity}。 missing validate_raw_record(record, list(mapping.keys()) [occurred_at_key]) if missing: return None normalized {} for src_key, dst_key in mapping.items(): normalized[dst_key] record.raw_payload[src_key] safe_payload obfuscate_payload(normalized, sensitive_keys) occurred_at record.raw_payload[occurred_at_key] return StandardEvent( event_idf{record.source}-{record.received_at.timestamp():.0f}-{abs(hash(str(record.raw_payload))) % 10**8}, event_typeevent_type, occurred_atoccurred_at, normalized_fieldssafe_payload, raw_sourcerecord.source, data_gradeobfuscated if sensitive_keys else original ) # 示例处理一条交易异常记录 trade_record RawRecord( sourcemarket_feed, raw_payload{ symbol: AAPL, price: 189.5, quantity: 5000, trader_id: T-10086, ts: datetime(2025, 1, 10, 9, 30, 15) } ) trade_event to_standard_event( recordtrade_record, event_typeorder_anomaly, mapping{symbol: ticker, price: price, quantity: quantity, ts: ts}, sensitive_keys[trader_id], occurred_at_keyts ) # 示例处理一条临床试验不良事件记录 ae_record RawRecord( sourceedc_form, raw_payload{ subject_id: SUBJ-0421, ae_term: 肝功能异常, severity: 3级, causality: 可能有关, reported_at: datetime(2025, 1, 10, 11, 5, 0) } ) ae_event to_standard_event( recordae_record, event_typeadverse_event, mapping{subject_id: subject_id, ae_term: ae_term, severity: severity, causality: causality, reported_at: reported_at}, sensitive_keys[subject_id], occurred_at_keyreported_at ) print(trade_event) print(ae_event)这段代码的核心思路是无论下层数据来自行情接口还是医院系统最终都收敛成一个延迟低、可追踪的标准事件。设计上有几个细节值得注意。第一occurred_at和received_at必须分开保存。交易场景里行情发生时间和系统接收时间可能只差毫秒但临床试验里受试者出现症状的日期和录入系统的日期可能相差几天。后续做时延分析、判断数据新鲜度时这两个时间缺一不可。第二脱敏不能简单丢弃敏感字段。临床试验的受试者 ID、交易的交易员 ID在审计时需要追溯原始事件但模型层又不应该直接接触到。所以生产环境建议使用可逆脱敏并保存映射关系权限隔离比强行删除更合理。第三字段映射要显式声明。不要在大模型 Prompt 里让模型自己去猜字段含义字段层面的标准化的责任应该由确定性代码完成模型只负责更高层的语义判断。这也符合垂直 AI 的一个基本原则能用规则解决的事情不要交给模型。5. 第二层决策工作流用状态机把模型放进业务流程标准事件进入系统后接下来要决定怎么办。这一步如果只靠大模型发散式地回答垂直 AI 会立刻失控。更稳妥的做法是用状态机把流程固定下来模型只负责其中一个或几个节点的输出其余环节由规则、人工和确定性逻辑控制。交易台和临床试验在这一点上的流程几乎可以一一对应。交易场景信号候选生成 → 风控规则检查 → 交易员复核 → 执行/拒绝。临床试验场景不良事件候选 → 自动去重和严重性规则 → 医学监查员审核 → 上报/关闭。这两个流程抽象成同一套状态机# vertical_ai_workflow.py 垂直 AI 决策工作流最小骨架。 核心思想模型只生成候选确定性规则做闸门人工对关键决策兜底。 交易台的订单拦截、临床试验的 AE 上报审核都可以用这套状态机表达。 from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional class WorkflowState(Enum): CANDIDATE_GENERATED CANDIDATE_GENERATED RULE_CHECK_PASSED RULE_CHECK_PASSED RULE_CHECK_FAILED RULE_CHECK_FAILED PENDING_HUMAN_REVIEW PENDING_HUMAN_REVIEW APPROVED APPROVED REJECTED REJECTED class RuleAction(Enum): PASS PASS REJECT REJECT FORCE_HUMAN_REVIEW FORCE_HUMAN_REVIEW dataclass class RuleResult: rule_code: str action: RuleAction reason: str dataclass class WorkflowContext: event_id: str model_confidence: float model_rationale: str rules: List[RuleResult] field(default_factorylist) human_reviewer: Optional[str] None human_comment: Optional[str] None current_state: WorkflowState WorkflowState.CANDIDATE_GENERATED class VerticalAIWorkflow: 一个极简状态机故意不引入外部依赖。 真实项目建议使用 Zeebe、Temporal 或自研流程引擎并持久化每个状态转移。 def __init__(self, force_review_rules: List[str]): self.force_review_rules force_review_rules def apply_rules(self, ctx: WorkflowContext, rule_results: List[RuleResult]) - None: 确定性规则闸门任何 FORCE_HUMAN_REVIEW 或 REJECT 优先于 PASS。 actions [r.action for r in ctx.rules] [r.action for r in rule_results] ctx.rules.extend(rule_results) if RuleAction.REJECT in actions: ctx.current_state WorkflowState.RULE_CHECK_FAILED return if RuleAction.FORCE_HUMAN_REVIEW in actions: ctx.current_state WorkflowState.PENDING_HUMAN_REVIEW return ctx.current_state WorkflowState.RULE_CHECK_PASSED def human_decision(self, ctx: WorkflowContext, reviewer: str, approved: bool, comment: str ) - None: 人工复核是最终闸门模型建议只作为参考。 if ctx.current_state ! WorkflowState.PENDING_HUMAN_REVIEW and ctx.current_state ! WorkflowState.RULE_CHECK_PASSED: raise ValueError(f当前状态不允许人工决策: {ctx.current_state}) ctx.human_reviewer reviewer ctx.human_comment comment ctx.current_state WorkflowState.APPROVED if approved else WorkflowState.REJECTED def run(self, ctx: WorkflowContext, candidate_rules: List[RuleResult]) - WorkflowContext: self.apply_rules(ctx, candidate_rules) return ctx # 交易场景订单异常候选 order_ctx WorkflowContext( event_idorder-anomaly-20250110-001, model_confidence0.93, model_rationale检测到该账户 5 分钟内交易量激增与历史行为偏差明显。 ) wf VerticalAIWorkflow(force_review_rules[RULE_LARGE_ORDER, RULE_NEW_CLIENT]) order_ctx wf.run(order_ctx, [ RuleResult(RULE_LARGE_ORDER, RuleAction.FORCE_HUMAN_REVIEW, 单笔订单数量超过阈值强制人工确认), RuleResult(RULE_BLACKLIST, RuleAction.PASS, 交易标的不在黑名单), ]) print(订单场景状态:, order_ctx.current_state) # 人工复核交易员决定是否拦截 wf.human_decision(order_ctx, reviewertrader-zhang, approvedTrue, comment量价走势不自然同意拦截) print(订单复核结果:, order_ctx.current_state, order_ctx.human_comment) # 临床试验场景不良事件候选 ae_ctx WorkflowContext( event_idae-20250110-0421-001, model_confidence0.87, model_rationale受试者报告肝功能异常根据既往病史和用药时间提示可能与试验药物相关。 ) ae_wf VerticalAIWorkflow(force_review_rules[RULE_SAE, RULE_UNBLINDED]) ae_ctx ae_wf.run(ae_ctx, [ RuleResult(RULE_SAE, RuleAction.FORCE_HUMAN_REVIEW, 事件满足严重不良事件定义需要医学监查员确认), RuleResult(RULE_DUPLICATE, RuleAction.PASS, 未发现重复上报), ]) print(AE场景状态:, ae_ctx.current_state) # 人工复核医学监查员决定是否上报 ae_wf.human_decision(ae_ctx, reviewermedical-monitor-li, approvedFalse, comment实验室复查结果已恢复评估为非 SAE) print(AE复核结果:, ae_ctx.current_state, ae_ctx.human_comment)这里最关键的工程决策是模型永远不直接产生最终状态。模型可以输出置信度和理由但状态迁移必须由规则引擎和人工审核控制。交易台里即使模型强烈建议拦截某笔订单风控规则如果判定为低风险系统也不该自动拦截临床试验里即使模型把某个事件标记为严重不良事件医学监查员复核后仍可以把它降级关闭。在这个设计里模型误判不会直接导致业务事故它最多产生一个错误候选错误最终会被规则或人工拦住。这也是垂直 AI 和生产型通用助手最大的不同它必须被设计成一个永远可以被否决的系统。真实项目中状态机还需要考虑两点。一是状态持久化事件进入工作流后每一步状态转移都要写入数据库方便异常恢复和审计查询。二是审核超时兜底比如临床试验要求 24 小时内完成 AE 初步判断超时必须有升级机制不能一直挂在待办里。这些都是业务规则不是模型能力问题。6. 第三层影子模式与反馈闭环让模型在生产环境中持续变好很多团队上线模型后只盯着准确率曲线却忽略了最重要的反馈来源人工复核结果。实际上交易员的拦截决定、医学监查员的上报判断才是模型迭代最宝贵的新标注数据。要拿到这些数据最安全的做法是让模型先生跑一段影子模式。影子模式的意思是线上业务流程完全不受模型建议影响系统同时把模型建议记录到日志中与人工最终决策结果对齐。这样可以无风险地验证两个问题模型如果真的上线它的建议和人工决策一致吗在哪些场景下不一致这些不一致是模型错了还是流程本身有优化空间# shadow_mode_evaluation.py 影子模式与离线评估骨架 模型建议先记录不阻断业务。 积累一定量的人工复核标签后离线计算一致率、误报率、漏报率。 from dataclasses import dataclass from typing import Dict, List, Optional dataclass class ModelSuggestion: event_id: str suggestion: str # 例如 需人工复核 / 无需处理 confidence: float rationale: str dataclass class HumanDecisionLog: event_id: str final_decision: str # 例如 复核通过 / 复核驳回 reviewer: str def evaluate_shadow_mode( suggestions: List[ModelSuggestion], decisions: List[HumanDecisionLog], positive_class: str 需人工复核 ) - Dict[str, float]: 把模型建议和人工最终结果对齐计算基础评估指标。 真实项目中还需要按事件类型分层评估避免单一指标掩盖结构性问题。 decision_map {d.event_id: d.final_decision for d in decisions} tp fp fn tn 0 for sug in suggestions: real decision_map.get(sug.event_id) if real is None: continue predicted_positive 1 if sug.suggestion positive_class else 0 actual_positive 1 if real 复核通过 else 0 if predicted_positive 1 and actual_positive 1: tp 1 elif predicted_positive 1 and actual_positive 0: fp 1 elif predicted_positive 0 and actual_positive 1: fn 1 else: tn 1 precision tp / (tp fp) if (tp fp) 0 else 0.0 recall tp / (tp fn) if (tp fn) 0 else 0.0 accuracy (tp tn) / (tp fp fn tn) if (tp fp fn tn) 0 else 0.0 return { precision: round(precision, 4), recall: round(recall, 4), accuracy: round(accuracy, 4), shadow_pairs: tp fp fn tn } # 模拟影子模式积累的数据 suggestions [ ModelSuggestion(evt-1, 需人工复核, 0.92, 风险信号明显), ModelSuggestion(evt-2, 无需处理, 0.68, 疑似噪声), ModelSuggestion(evt-3, 需人工复核, 0.85, 指标异常), ] decisions [ HumanDecisionLog(evt-1, 复核通过), HumanDecisionLog(evt-2, 复核驳回), HumanDecisionLog(evt-3, 复核通过), ] metrics evaluate_shadow_mode(suggestions, decisions) print(影子模式评估结果:, metrics)评估代码不复杂但真正困难的是建立两条对齐链路。第一模型建议里的event_id必须和人工决策日志里的event_id一致这要求数据管道从一开始就生成稳定的追踪 ID这也是我在第 4 节强调StandardEvent.event_id的原因。第二人工决策日志要记录复核通过/复核驳回之外的原因比如交易员驳回是因为认为模型误报还是因为有其他业务考量医学监查员批准上报是因为符合标准方案还是因为受试者状态变化。没有原因后面做 badcase 分析时只能靠猜。积累到足够多的对齐数据以后就可以定期对模型做增量训练或者微调。但垂直 AI 项目通常不建议频繁全量微调更稳妥的做法是先优化规则闸门、调整提示词、补充 RAG 知识片段把规则和检索能解决的问题优先解决模型训不动了再考虑微调。反馈闭环不是让模型无限变聪明而是让整个系统越来越懂业务。7. 从 POC 到生产三个躲不开的控制点演示版垂直 AI 到处都能做出来但生产环境完全不同。根据跨行业的实践有三个控制点几乎决定了项目能不能长期存活。控制点一上线前必须有业务基线。很多项目一开始就说AI 准确率 90%但没有回答不用 AI 时人工处理准确率是多少、人均处理能力是多少。没有这个基线你无法证明模型产生了增量价值。交易台可以用历史人工拦截率作为基线临床试验可以用过去一个季度的漏报率、监查员平均处理时长作为基线。模型指标必须和业务基线放在同一张表里对比。控制点二人工兜底不能是一句口号。设计工作流时要明确哪类事件强制人工复核、复核时限是多少、审核人不够时如何升级。从代码层面看PENDING_HUMAN_REVIEW是最重要的中间状态任何高成本决策动作都不能从CANDIDATE_GENERATED直接跳到APPROVED。我们前面状态机示例里的FORCE_HUMAN_REVIEW规则就是这种思想的体现。控制点三审计追踪必须从第一天就做。交易行业有交易留痕要求临床试验有电子记录法规约束但即使没有监管压力垂直 AI 系统也应该默认记录完整决策链。具体包括模型建议内容、置信度、命中的规则、人工复核人、复核结论、修改后的字段、时间戳。不要等到出现业务争议时再补日志到时候已经补不回来了。这三个控制点本质上是把垂直 AI 从一个模型变成一个业务系统。它们的实现不依赖先进算法依赖的是工程纪律。8. 垂直 AI 落地常见问题与排查垂直 AI 项目的失败模式比较集中。以下这些问题是跨行业反复出现的团队可以对照排查。问题现象可能原因排查方式解决方案模型建议与人工决策经常不一致人工检查时使用了模型没见过的信息比如电话沟通、线下记录对比模型输入特征和人工参考材料梳理信息差把关键信息接入数据管道无法自动化的信息在提示词中要求模型标注缺少信息规则闸门误伤过多规则阈值设置过严或者规则条件过宽统计规则命中率和最终复核通过率用历史数据回放规则调优阈值规则只做硬性边界不做精细判断模型产生幻觉字段Prompt 让模型做字段级推断超出模型能力边界检查代表性 badcase定位是提示词问题还是字段映射问题字段标准化交给确定性代码模型只做语义候选生成影子模式数据无法对齐事件 ID 不统一或日志缺少关键字段检查全链路日志是否都有稳定 event_id从数据管道入口生成统一追踪 ID并强制写入所有日志生产环境效果远低于测试集训练/测试数据与实际业务数据分布不一致对比特征分布、事件类型分布构建持续的线上评估集用影子模式数据做分布漂移检测人工复核成为瓶颈系统把太多事件推到人工审核队列积压按事件类型统计人工处理耗时和积压量拆分为自动通过、抽查、全量人工三级只对高成本决策强制人工这些问题的共性是表面看是模型效果问题实际大多是数据链路和流程设计问题。所以排查时不要一上来就调 Prompt 或重训模型先看数据管道有没有把信息完整地带进来再看规则闸门是不是承担了超出能力的职责最后才回到模型本身。9. 给技术团队的工程建议垂直 AI 对技术团队的要求和做纯算法团队不完全一样。结合交易台、临床试验这类场景的共性我给出几条比较实际的建议。先搭数据管道再谈模型。很多团队一上来就用大模型处理原始报文结果模型每跑一次都可能输出不同字段。正确的顺序是先把原始数据清洗成标准事件让模型在标准事件之上工作这样模型的输入和输出都可控。用规则表达确定性部分用模型表达模糊部分。比如该订单金额是否超过阈值该事件是否满足严重不良事件的时限定义这类有明确标准的判断应该写成规则这条舆情是否暗示重大风险这个不良事件和药物的因果关系如何初判这类语义模糊的判断才适合模型。不要让一个 Agent 做完整决策。交易台不需要一个 Agent 直接下单临床试验更不应该让一个 Agent 直接生成监管报告。稳妥的做法是多个小模型和规则分工一个模型做候选筛选一个模型做报告草稿规则和人工在最关键的闸门处把关。从第一天就引入业务专家进入迭代循环。交易员和医学监查员不只是最终用户他们还是 badcase 评审者。让专家参与每周的模型效果回顾能显著加快闭环迭代速度。这也是我说垂直 AI 真正的门槛在领域过程数字化的原因你不可能在一个自己不懂流程的领域里做出好系统。用 YAML 把工作流和规则配置化。规则不是写死在代码里而是沉淀成可评审、可测试的配置。下面给一个简单样例实际项目可以扩展成规则中心。# vertical_workflow_config.yaml workflow: name: adverse_event_triage source: clinical_data_pipeline model: vertical_model_v1 stages: candidate_generation: output_state: CANDIDATE_GENERATED model_input: standard_event max_candidates: 50 risk_rules: input_state: CANDIDATE_GENERATED rules: - code: RULE_SAE description: 满足严重不良事件定义时强制人工 action: FORCE_HUMAN_REVIEW condition: normalized_fields.severity in [4级,5级] or event_type death - code: RULE_DUPLICATE description: 同一受试者同一 AE 去重 action: MERGE_OR_DROP condition: dedupe_key exists human_review: input_state: PENDING_HUMAN_REVIEW timeout_hours: 24 fallback_action: ESCALATE reviewer_role: medical_monitor配置化之后业务人员可以参与规则评审测试人员可以针对每条规则单独造数据验证运维人员也更容易在故障时快速定位是哪一条规则出了问题。10. 总结垂直 AI 的真正门槛在领域过程数字化回到交易台和临床试验的对比。两个行业共享的并不是某个模型而是同一套落地逻辑数据管道负责让模型接收到干净的标准事件工作流保证模型建议永远可以被规则拦截和人工否决反馈闭环让系统在真实业务反馈中持续迭代。垂直 AI 的真正门槛不在模型参数量不在算力预算而在于一个团队能不能把一个模糊的业务问题逐步数字化、流程化、可测试化。模型能力会越来越强但业务过程的建模能力、工程落地能力、人机协作的设计能力才是垂直 AI 时代技术团队的核心竞争力。如果你的团队正准备启动垂直 AI 项目建议先不要急着微调模型。先从一条最关键的业务流程开始画清楚事件流转图梳理出哪些环节适合规则、哪些环节适合模型、哪些环节必须保留人工然后按本文的管道、工作流、反馈闭环三层骨架搭建一个最小系统。跑通一个场景以后你会发现其他场景的接入会比想象中更快。
垂直AI跨行业落地：从交易台到临床试验的技术架构与工程实践
从交易台到临床试验垂直 AI 跨行业复用的技术底层逻辑如果你团队负责交易监控每天要面对几万条行情信号、新闻流和异常告警模型给出的判断一旦出错带来的可能就是直接的资金损失如果你是临床试验数据团队每天要处理海量的受试者数据、不良事件记录和方案偏离报告一次漏报或误判就可能影响药品审批进度甚至威胁受试者安全。这两个场景隔行如隔山但你大概率会发现它们需要的 AI 系统长得很像数据先被清洗成结构化事件模型在事件流里挑出可疑候选规则引擎把明确的风险拦下来其余交给人工复核最后所有决定留下审计日志。最近大家讨论的 Vertical AI真正吸引人的不是某一个行业爆款应用而是这种领域数据管道 模型能力 业务决策工作流的组合正在金融、医疗、制造等不同行业里反复出现。这篇文章想做的事情也很具体把交易台与临床试验放在一起对比拆出 Vertical AI 落地的共性技术架构并给出你可以直接复用的代码思路。读完你会理解为什么垂直 AI 不是简单地把 ChatGPT 包装成本行业助手也会知道一个垂直 AI 项目从 POC 走向生产环境最该先做哪三件事。1. 垂直 AI 不是行业版 ChatGPT过去两年所有人都见识过通用大模型的文本生成、摘要和问答能力。于是很多团队产生了同一种冲动用企业私有文档做一次 RAG再套上一个行业界面就宣称自己做出了医疗大模型金融大模型法律大模型。从材料和我们观察到的实际案例来看这类产品多数停留在演示层面。原因在于垂直 AI 的核心价值不在能聊领域话题而在能介入业务决策。我们可以这样区分行业问答助手输入是一段自然语言问题输出是一段文本答案。用户自己判断答案对不对模型并不对业务结果负责。垂直 AI 系统输入是真实业务数据流输出是一个可执行的业务动作比如标记该事件为严重不良事件候选暂停该交易信号推送生成需要人工复核的监查报告。动作执行错了系统要承担责任。这中间的差距不是模型能力提升就能填平的而是缺少一层关键的业务结构。垂直 AI 系统必须知道事件是什么、状态怎么流转、哪个环节必须人工看、哪个环节可以自动过、决策依据有没有留存。这些结构来自对业务过程的数字化而不是来自参数更多的大模型。所以我的判断是垂直 AI 的本质是把某个领域的业务过程重新建模再把模型能力嵌入到这个过程的特定环节里。模型是发动机但整车是业务过程本身。如果一个团队只盯着模型参数却说不清楚自己的业务流程有哪些状态和节点那么离真正的垂直 AI 还很远。2. 交易台与临床试验两个行业里藏着的同一个 AI 架构交易台和临床试验表面上毫无关系但如果我们抽象到决策系统层面它们的相似度非常高。先看交易台。交易员面前有行情报价、宏观经济数据、新闻舆情、历史订单流系统要回答的问题是当前有没有异常交易机会要不要提醒交易员该不该自动执行风险控制动作整个过程里速度很重要、错误代价很高、事后必须能回溯每一笔决策依据。再看临床试验。数据监查员面前有电子病例报告表eCRF、实验室检查结果、不良事件记录、方案偏离记录系统要回答的问题是这名受试者有没有出现新的严重不良事件这个事件和试验药物的关系能不能初步判断需不需要在 24 小时内向监管方上报同样速度影响患者安全错误代价更高事后审计要求甚至比金融更严格。把两者放在一起对比能看到完全同构的结构维度交易台临床试验对 AI 系统的要求输入数据行情、订单、舆情、历史交易受试者数据、检验指标、不良事件、方案记录数据管道能处理高噪声、多来源的流式输入决策类型提醒、拦截、自动执行标记、初判、上报、生成报告模型只负责候选生成最终决策权可控失败成本资金损失、合规罚款受试者伤害、审评失败必须有风控规则和人工兜底合规约束交易留痕、监管报送审计追踪、电子记录合规每个判断都要能还原、可审计专家稀缺经验丰富的交易员成本极高医学监查员培养周期长AI 做初筛专家处理疑难这张表背后其实是 Vertical AI 最典型的价值区间数据信息密度足够高人工处理不过来决策成本高值得用系统兜底领域专家稀缺AI 可以把专家精力从重复劳动里释放出来。如果你准备在自己的行业里推动垂直 AI可以先拿这张表来对照。一个业务要成为垂直 AI 的合适场景至少满足三个条件有持续产生的结构化潜力很高的原始数据有明确的决策动作或者提醒动作有清晰的责任边界和审计要求。三者缺一项目后期大概率会陷入模型很酷但没有业务价值的困境。3. 垂直 AI 的四层技术栈既然结构相似落地时采用的技术栈也可以提炼成四层。以下四层是我在整理各种行业实践后归纳出的一套通用框架适合作为团队讨论和架构设计的起点。第一层领域数据管道。负责接入原始数据完成解析、清洗、去重、脱敏、字段标准化最终输出统一的领域事件。交易台里的行情快照和临床试验里的实验室检验记录格式完全不同但经过这一层后输出的都是带时间戳、事件类型、标准化字段的通用事件对象。第二层模型能力层。这一层包括基座模型、领域知识增强RAG、可选微调、规则引擎和打分模型。实际应用中很多判断不需要每次都调用大模型规则引擎可以解决大部分确定性逻辑模型负责处理语义理解和模糊判断。两者结合才能兼顾成本和效果。第三层业务工作流层。这是垂直 AI 区别于通用 AI 助手的核心。业务上有明确流程候选生成以后要过什么规则、哪些情况要强制人工、人工审核超时了怎么办、审批通过以后触发什么下游动作。这一层必须用工作流引擎或状态机落地保证流程可以被测试、被回放。第四层反馈闭环层。模型上线后不能只是跑了就行要把人工复核的结果回传形成新的标注数据定期重训或微调模型同时用影子模式持续对比模型建议和实际决策量化模型价值。这四层里最容易低估的是第一层和第三层。绝大多数垂直 AI 项目失败不是模型效果不够好而是数据进不来、流程接不上。接下来的内容我会用代码把这四层的关键部分串起来。4. 第一层领域数据管道先把野外数据变成标准事件假设我们要同时支持一个交易信号场景和一个临床试验不良事件场景。先不关心它们各自的数据库结构而是定义一个通用的标准事件模型。# vertical_ai_pipeline.py 垂直 AI 数据管道最小骨架 原始记录 - 规则校验 - 脱敏 - 标准事件 在交易台和临床试验场景中只需要替换数据源接入层和字段映射规则。 from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional dataclass class RawRecord: source: str # 数据来源market_feed / edc_form / lab_report raw_payload: Dict[str, Any] # 原始字段 received_at: datetime field(default_factorydatetime.utcnow) dataclass class StandardEvent: event_id: str # 稳定事件ID用于全链路追踪 event_type: str # 如 order_anomaly / adverse_event occurred_at: datetime # 业务发生时间不是接收时间 normalized_fields: Dict[str, Any] # 标准化后的字段 raw_source: str # 追溯原始记录 data_grade: str original # original / modified / obfuscated def validate_raw_record(record: RawRecord, required_fields: List[str]) - List[str]: 最简单的字段完整性校验返回缺失字段列表。 missing [f for f in required_fields if f not in record.raw_payload] return missing def obfuscate_payload(payload: Dict[str, Any], sensitive_keys: List[str]) - Dict[str, Any]: 对敏感字段做脱敏。生产环境建议使用可逆脱敏并保存映射便于审计回查。 safe {k: v for k, v in payload.items() if k not in sensitive_keys} for key in sensitive_keys: if key in payload: # 这里用简单哈希示意生产环境应使用密钥管理的脱敏算法 safe[key] fREDACTED:{hash(payload[key]) % 10**8:08d} return safe def to_standard_event(record: RawRecord, event_type: str, mapping: Dict[str, str], sensitive_keys: List[str], occurred_at_key: str) - Optional[StandardEvent]: mapping 把原始字段名映射为标准化字段名。 例如交易场景 {symbol: ticker, qty: quantity}。 missing validate_raw_record(record, list(mapping.keys()) [occurred_at_key]) if missing: return None normalized {} for src_key, dst_key in mapping.items(): normalized[dst_key] record.raw_payload[src_key] safe_payload obfuscate_payload(normalized, sensitive_keys) occurred_at record.raw_payload[occurred_at_key] return StandardEvent( event_idf{record.source}-{record.received_at.timestamp():.0f}-{abs(hash(str(record.raw_payload))) % 10**8}, event_typeevent_type, occurred_atoccurred_at, normalized_fieldssafe_payload, raw_sourcerecord.source, data_gradeobfuscated if sensitive_keys else original ) # 示例处理一条交易异常记录 trade_record RawRecord( sourcemarket_feed, raw_payload{ symbol: AAPL, price: 189.5, quantity: 5000, trader_id: T-10086, ts: datetime(2025, 1, 10, 9, 30, 15) } ) trade_event to_standard_event( recordtrade_record, event_typeorder_anomaly, mapping{symbol: ticker, price: price, quantity: quantity, ts: ts}, sensitive_keys[trader_id], occurred_at_keyts ) # 示例处理一条临床试验不良事件记录 ae_record RawRecord( sourceedc_form, raw_payload{ subject_id: SUBJ-0421, ae_term: 肝功能异常, severity: 3级, causality: 可能有关, reported_at: datetime(2025, 1, 10, 11, 5, 0) } ) ae_event to_standard_event( recordae_record, event_typeadverse_event, mapping{subject_id: subject_id, ae_term: ae_term, severity: severity, causality: causality, reported_at: reported_at}, sensitive_keys[subject_id], occurred_at_keyreported_at ) print(trade_event) print(ae_event)这段代码的核心思路是无论下层数据来自行情接口还是医院系统最终都收敛成一个延迟低、可追踪的标准事件。设计上有几个细节值得注意。第一occurred_at和received_at必须分开保存。交易场景里行情发生时间和系统接收时间可能只差毫秒但临床试验里受试者出现症状的日期和录入系统的日期可能相差几天。后续做时延分析、判断数据新鲜度时这两个时间缺一不可。第二脱敏不能简单丢弃敏感字段。临床试验的受试者 ID、交易的交易员 ID在审计时需要追溯原始事件但模型层又不应该直接接触到。所以生产环境建议使用可逆脱敏并保存映射关系权限隔离比强行删除更合理。第三字段映射要显式声明。不要在大模型 Prompt 里让模型自己去猜字段含义字段层面的标准化的责任应该由确定性代码完成模型只负责更高层的语义判断。这也符合垂直 AI 的一个基本原则能用规则解决的事情不要交给模型。5. 第二层决策工作流用状态机把模型放进业务流程标准事件进入系统后接下来要决定怎么办。这一步如果只靠大模型发散式地回答垂直 AI 会立刻失控。更稳妥的做法是用状态机把流程固定下来模型只负责其中一个或几个节点的输出其余环节由规则、人工和确定性逻辑控制。交易台和临床试验在这一点上的流程几乎可以一一对应。交易场景信号候选生成 → 风控规则检查 → 交易员复核 → 执行/拒绝。临床试验场景不良事件候选 → 自动去重和严重性规则 → 医学监查员审核 → 上报/关闭。这两个流程抽象成同一套状态机# vertical_ai_workflow.py 垂直 AI 决策工作流最小骨架。 核心思想模型只生成候选确定性规则做闸门人工对关键决策兜底。 交易台的订单拦截、临床试验的 AE 上报审核都可以用这套状态机表达。 from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional class WorkflowState(Enum): CANDIDATE_GENERATED CANDIDATE_GENERATED RULE_CHECK_PASSED RULE_CHECK_PASSED RULE_CHECK_FAILED RULE_CHECK_FAILED PENDING_HUMAN_REVIEW PENDING_HUMAN_REVIEW APPROVED APPROVED REJECTED REJECTED class RuleAction(Enum): PASS PASS REJECT REJECT FORCE_HUMAN_REVIEW FORCE_HUMAN_REVIEW dataclass class RuleResult: rule_code: str action: RuleAction reason: str dataclass class WorkflowContext: event_id: str model_confidence: float model_rationale: str rules: List[RuleResult] field(default_factorylist) human_reviewer: Optional[str] None human_comment: Optional[str] None current_state: WorkflowState WorkflowState.CANDIDATE_GENERATED class VerticalAIWorkflow: 一个极简状态机故意不引入外部依赖。 真实项目建议使用 Zeebe、Temporal 或自研流程引擎并持久化每个状态转移。 def __init__(self, force_review_rules: List[str]): self.force_review_rules force_review_rules def apply_rules(self, ctx: WorkflowContext, rule_results: List[RuleResult]) - None: 确定性规则闸门任何 FORCE_HUMAN_REVIEW 或 REJECT 优先于 PASS。 actions [r.action for r in ctx.rules] [r.action for r in rule_results] ctx.rules.extend(rule_results) if RuleAction.REJECT in actions: ctx.current_state WorkflowState.RULE_CHECK_FAILED return if RuleAction.FORCE_HUMAN_REVIEW in actions: ctx.current_state WorkflowState.PENDING_HUMAN_REVIEW return ctx.current_state WorkflowState.RULE_CHECK_PASSED def human_decision(self, ctx: WorkflowContext, reviewer: str, approved: bool, comment: str ) - None: 人工复核是最终闸门模型建议只作为参考。 if ctx.current_state ! WorkflowState.PENDING_HUMAN_REVIEW and ctx.current_state ! WorkflowState.RULE_CHECK_PASSED: raise ValueError(f当前状态不允许人工决策: {ctx.current_state}) ctx.human_reviewer reviewer ctx.human_comment comment ctx.current_state WorkflowState.APPROVED if approved else WorkflowState.REJECTED def run(self, ctx: WorkflowContext, candidate_rules: List[RuleResult]) - WorkflowContext: self.apply_rules(ctx, candidate_rules) return ctx # 交易场景订单异常候选 order_ctx WorkflowContext( event_idorder-anomaly-20250110-001, model_confidence0.93, model_rationale检测到该账户 5 分钟内交易量激增与历史行为偏差明显。 ) wf VerticalAIWorkflow(force_review_rules[RULE_LARGE_ORDER, RULE_NEW_CLIENT]) order_ctx wf.run(order_ctx, [ RuleResult(RULE_LARGE_ORDER, RuleAction.FORCE_HUMAN_REVIEW, 单笔订单数量超过阈值强制人工确认), RuleResult(RULE_BLACKLIST, RuleAction.PASS, 交易标的不在黑名单), ]) print(订单场景状态:, order_ctx.current_state) # 人工复核交易员决定是否拦截 wf.human_decision(order_ctx, reviewertrader-zhang, approvedTrue, comment量价走势不自然同意拦截) print(订单复核结果:, order_ctx.current_state, order_ctx.human_comment) # 临床试验场景不良事件候选 ae_ctx WorkflowContext( event_idae-20250110-0421-001, model_confidence0.87, model_rationale受试者报告肝功能异常根据既往病史和用药时间提示可能与试验药物相关。 ) ae_wf VerticalAIWorkflow(force_review_rules[RULE_SAE, RULE_UNBLINDED]) ae_ctx ae_wf.run(ae_ctx, [ RuleResult(RULE_SAE, RuleAction.FORCE_HUMAN_REVIEW, 事件满足严重不良事件定义需要医学监查员确认), RuleResult(RULE_DUPLICATE, RuleAction.PASS, 未发现重复上报), ]) print(AE场景状态:, ae_ctx.current_state) # 人工复核医学监查员决定是否上报 ae_wf.human_decision(ae_ctx, reviewermedical-monitor-li, approvedFalse, comment实验室复查结果已恢复评估为非 SAE) print(AE复核结果:, ae_ctx.current_state, ae_ctx.human_comment)这里最关键的工程决策是模型永远不直接产生最终状态。模型可以输出置信度和理由但状态迁移必须由规则引擎和人工审核控制。交易台里即使模型强烈建议拦截某笔订单风控规则如果判定为低风险系统也不该自动拦截临床试验里即使模型把某个事件标记为严重不良事件医学监查员复核后仍可以把它降级关闭。在这个设计里模型误判不会直接导致业务事故它最多产生一个错误候选错误最终会被规则或人工拦住。这也是垂直 AI 和生产型通用助手最大的不同它必须被设计成一个永远可以被否决的系统。真实项目中状态机还需要考虑两点。一是状态持久化事件进入工作流后每一步状态转移都要写入数据库方便异常恢复和审计查询。二是审核超时兜底比如临床试验要求 24 小时内完成 AE 初步判断超时必须有升级机制不能一直挂在待办里。这些都是业务规则不是模型能力问题。6. 第三层影子模式与反馈闭环让模型在生产环境中持续变好很多团队上线模型后只盯着准确率曲线却忽略了最重要的反馈来源人工复核结果。实际上交易员的拦截决定、医学监查员的上报判断才是模型迭代最宝贵的新标注数据。要拿到这些数据最安全的做法是让模型先生跑一段影子模式。影子模式的意思是线上业务流程完全不受模型建议影响系统同时把模型建议记录到日志中与人工最终决策结果对齐。这样可以无风险地验证两个问题模型如果真的上线它的建议和人工决策一致吗在哪些场景下不一致这些不一致是模型错了还是流程本身有优化空间# shadow_mode_evaluation.py 影子模式与离线评估骨架 模型建议先记录不阻断业务。 积累一定量的人工复核标签后离线计算一致率、误报率、漏报率。 from dataclasses import dataclass from typing import Dict, List, Optional dataclass class ModelSuggestion: event_id: str suggestion: str # 例如 需人工复核 / 无需处理 confidence: float rationale: str dataclass class HumanDecisionLog: event_id: str final_decision: str # 例如 复核通过 / 复核驳回 reviewer: str def evaluate_shadow_mode( suggestions: List[ModelSuggestion], decisions: List[HumanDecisionLog], positive_class: str 需人工复核 ) - Dict[str, float]: 把模型建议和人工最终结果对齐计算基础评估指标。 真实项目中还需要按事件类型分层评估避免单一指标掩盖结构性问题。 decision_map {d.event_id: d.final_decision for d in decisions} tp fp fn tn 0 for sug in suggestions: real decision_map.get(sug.event_id) if real is None: continue predicted_positive 1 if sug.suggestion positive_class else 0 actual_positive 1 if real 复核通过 else 0 if predicted_positive 1 and actual_positive 1: tp 1 elif predicted_positive 1 and actual_positive 0: fp 1 elif predicted_positive 0 and actual_positive 1: fn 1 else: tn 1 precision tp / (tp fp) if (tp fp) 0 else 0.0 recall tp / (tp fn) if (tp fn) 0 else 0.0 accuracy (tp tn) / (tp fp fn tn) if (tp fp fn tn) 0 else 0.0 return { precision: round(precision, 4), recall: round(recall, 4), accuracy: round(accuracy, 4), shadow_pairs: tp fp fn tn } # 模拟影子模式积累的数据 suggestions [ ModelSuggestion(evt-1, 需人工复核, 0.92, 风险信号明显), ModelSuggestion(evt-2, 无需处理, 0.68, 疑似噪声), ModelSuggestion(evt-3, 需人工复核, 0.85, 指标异常), ] decisions [ HumanDecisionLog(evt-1, 复核通过), HumanDecisionLog(evt-2, 复核驳回), HumanDecisionLog(evt-3, 复核通过), ] metrics evaluate_shadow_mode(suggestions, decisions) print(影子模式评估结果:, metrics)评估代码不复杂但真正困难的是建立两条对齐链路。第一模型建议里的event_id必须和人工决策日志里的event_id一致这要求数据管道从一开始就生成稳定的追踪 ID这也是我在第 4 节强调StandardEvent.event_id的原因。第二人工决策日志要记录复核通过/复核驳回之外的原因比如交易员驳回是因为认为模型误报还是因为有其他业务考量医学监查员批准上报是因为符合标准方案还是因为受试者状态变化。没有原因后面做 badcase 分析时只能靠猜。积累到足够多的对齐数据以后就可以定期对模型做增量训练或者微调。但垂直 AI 项目通常不建议频繁全量微调更稳妥的做法是先优化规则闸门、调整提示词、补充 RAG 知识片段把规则和检索能解决的问题优先解决模型训不动了再考虑微调。反馈闭环不是让模型无限变聪明而是让整个系统越来越懂业务。7. 从 POC 到生产三个躲不开的控制点演示版垂直 AI 到处都能做出来但生产环境完全不同。根据跨行业的实践有三个控制点几乎决定了项目能不能长期存活。控制点一上线前必须有业务基线。很多项目一开始就说AI 准确率 90%但没有回答不用 AI 时人工处理准确率是多少、人均处理能力是多少。没有这个基线你无法证明模型产生了增量价值。交易台可以用历史人工拦截率作为基线临床试验可以用过去一个季度的漏报率、监查员平均处理时长作为基线。模型指标必须和业务基线放在同一张表里对比。控制点二人工兜底不能是一句口号。设计工作流时要明确哪类事件强制人工复核、复核时限是多少、审核人不够时如何升级。从代码层面看PENDING_HUMAN_REVIEW是最重要的中间状态任何高成本决策动作都不能从CANDIDATE_GENERATED直接跳到APPROVED。我们前面状态机示例里的FORCE_HUMAN_REVIEW规则就是这种思想的体现。控制点三审计追踪必须从第一天就做。交易行业有交易留痕要求临床试验有电子记录法规约束但即使没有监管压力垂直 AI 系统也应该默认记录完整决策链。具体包括模型建议内容、置信度、命中的规则、人工复核人、复核结论、修改后的字段、时间戳。不要等到出现业务争议时再补日志到时候已经补不回来了。这三个控制点本质上是把垂直 AI 从一个模型变成一个业务系统。它们的实现不依赖先进算法依赖的是工程纪律。8. 垂直 AI 落地常见问题与排查垂直 AI 项目的失败模式比较集中。以下这些问题是跨行业反复出现的团队可以对照排查。问题现象可能原因排查方式解决方案模型建议与人工决策经常不一致人工检查时使用了模型没见过的信息比如电话沟通、线下记录对比模型输入特征和人工参考材料梳理信息差把关键信息接入数据管道无法自动化的信息在提示词中要求模型标注缺少信息规则闸门误伤过多规则阈值设置过严或者规则条件过宽统计规则命中率和最终复核通过率用历史数据回放规则调优阈值规则只做硬性边界不做精细判断模型产生幻觉字段Prompt 让模型做字段级推断超出模型能力边界检查代表性 badcase定位是提示词问题还是字段映射问题字段标准化交给确定性代码模型只做语义候选生成影子模式数据无法对齐事件 ID 不统一或日志缺少关键字段检查全链路日志是否都有稳定 event_id从数据管道入口生成统一追踪 ID并强制写入所有日志生产环境效果远低于测试集训练/测试数据与实际业务数据分布不一致对比特征分布、事件类型分布构建持续的线上评估集用影子模式数据做分布漂移检测人工复核成为瓶颈系统把太多事件推到人工审核队列积压按事件类型统计人工处理耗时和积压量拆分为自动通过、抽查、全量人工三级只对高成本决策强制人工这些问题的共性是表面看是模型效果问题实际大多是数据链路和流程设计问题。所以排查时不要一上来就调 Prompt 或重训模型先看数据管道有没有把信息完整地带进来再看规则闸门是不是承担了超出能力的职责最后才回到模型本身。9. 给技术团队的工程建议垂直 AI 对技术团队的要求和做纯算法团队不完全一样。结合交易台、临床试验这类场景的共性我给出几条比较实际的建议。先搭数据管道再谈模型。很多团队一上来就用大模型处理原始报文结果模型每跑一次都可能输出不同字段。正确的顺序是先把原始数据清洗成标准事件让模型在标准事件之上工作这样模型的输入和输出都可控。用规则表达确定性部分用模型表达模糊部分。比如该订单金额是否超过阈值该事件是否满足严重不良事件的时限定义这类有明确标准的判断应该写成规则这条舆情是否暗示重大风险这个不良事件和药物的因果关系如何初判这类语义模糊的判断才适合模型。不要让一个 Agent 做完整决策。交易台不需要一个 Agent 直接下单临床试验更不应该让一个 Agent 直接生成监管报告。稳妥的做法是多个小模型和规则分工一个模型做候选筛选一个模型做报告草稿规则和人工在最关键的闸门处把关。从第一天就引入业务专家进入迭代循环。交易员和医学监查员不只是最终用户他们还是 badcase 评审者。让专家参与每周的模型效果回顾能显著加快闭环迭代速度。这也是我说垂直 AI 真正的门槛在领域过程数字化的原因你不可能在一个自己不懂流程的领域里做出好系统。用 YAML 把工作流和规则配置化。规则不是写死在代码里而是沉淀成可评审、可测试的配置。下面给一个简单样例实际项目可以扩展成规则中心。# vertical_workflow_config.yaml workflow: name: adverse_event_triage source: clinical_data_pipeline model: vertical_model_v1 stages: candidate_generation: output_state: CANDIDATE_GENERATED model_input: standard_event max_candidates: 50 risk_rules: input_state: CANDIDATE_GENERATED rules: - code: RULE_SAE description: 满足严重不良事件定义时强制人工 action: FORCE_HUMAN_REVIEW condition: normalized_fields.severity in [4级,5级] or event_type death - code: RULE_DUPLICATE description: 同一受试者同一 AE 去重 action: MERGE_OR_DROP condition: dedupe_key exists human_review: input_state: PENDING_HUMAN_REVIEW timeout_hours: 24 fallback_action: ESCALATE reviewer_role: medical_monitor配置化之后业务人员可以参与规则评审测试人员可以针对每条规则单独造数据验证运维人员也更容易在故障时快速定位是哪一条规则出了问题。10. 总结垂直 AI 的真正门槛在领域过程数字化回到交易台和临床试验的对比。两个行业共享的并不是某个模型而是同一套落地逻辑数据管道负责让模型接收到干净的标准事件工作流保证模型建议永远可以被规则拦截和人工否决反馈闭环让系统在真实业务反馈中持续迭代。垂直 AI 的真正门槛不在模型参数量不在算力预算而在于一个团队能不能把一个模糊的业务问题逐步数字化、流程化、可测试化。模型能力会越来越强但业务过程的建模能力、工程落地能力、人机协作的设计能力才是垂直 AI 时代技术团队的核心竞争力。如果你的团队正准备启动垂直 AI 项目建议先不要急着微调模型。先从一条最关键的业务流程开始画清楚事件流转图梳理出哪些环节适合规则、哪些环节适合模型、哪些环节必须保留人工然后按本文的管道、工作流、反馈闭环三层骨架搭建一个最小系统。跑通一个场景以后你会发现其他场景的接入会比想象中更快。
