AI 商业场景落地

商业 AI 不是“接一个聊天框”就结束了。真正能上线的 AI 应用，必须能解决业务问题，并且在权限、准确率、成本、延迟、审计、人工兜底、数据安全上可控。

一句话理解：

商业 AI 落地 = 模型能力 + 业务流程 + 私有数据 + 权限控制 + 质量评估 + 安全审计 + 成本治理。

学习目标

目标	需要掌握什么
知道能做什么	客服问答、企业知识库、合同抽取、工单质检、数据分析、Agent 自动化
知道为什么不能直接上线	模型会幻觉、会误调用工具、会泄露敏感信息、会产生成本失控
知道怎么设计链路	入口鉴权、意图识别、RAG、Tool Calling、输出校验、人工兜底
知道商用边界	金额、库存、权限、医疗诊断、法律结论不能只由模型直接决定
会写 Demo	能写最小 AI 网关、结构化抽取、RAG 问答和工具调用代码
会做生产治理	日志、评估集、灰度、监控、回滚、敏感数据脱敏、审计追踪

商业 AI 的核心原则

原则	为什么
AI 是能力组件，不是最终事实源	订单、库存、合同、病历、权限仍以业务系统和数据库为准
先限定场景，再选择技术	不要为了用模型而用模型，先看业务收益和风险
高风险结果必须可追溯	要记录输入、检索资料、模型版本、输出、用户反馈
私有知识必须走权限过滤	用户只能看到自己有权访问的资料
工具调用必须后端校验	模型不能直接决定扣款、删库、改权限、发通知
回答必须可评估	没有评估集，就无法判断上线后是变好还是变坏
成本必须可控	Token、模型级别、缓存、限流、并发都要有预算

一个商业 AI 应用的通用链路

flowchart TD
    A["用户请求"] --> B["身份认证和权限判断"]
    B --> C["输入清洗和敏感词检测"]
    C --> D["意图识别"]
    D --> E{"是否需要私有知识"}
    E -- "需要" --> F["按权限检索知识库"]
    E -- "不需要" --> G["组装 Prompt"]
    F --> H["召回、重排、引用片段"]
    H --> G
    G --> I{"是否需要调用工具"}
    I -- "需要" --> J["生成工具调用参数"]
    J --> K["后端校验权限和幂等"]
    K --> L["执行业务接口"]
    L --> G
    I -- "不需要" --> M["调用模型生成结果"]
    M --> N["输出校验和风险过滤"]
    N --> O["返回结果和引用来源"]
    O --> P["记录日志、反馈和评估数据"]

这条链路里，模型只是中间一环。真正的工程难点在：怎么让模型只能看该看的资料、只能做允许的操作、答错时能被发现、出问题时能回滚。

商业场景总览

场景	典型价值	核心技术	最大风险
智能客服	降低重复咨询、人力成本	RAG、引用、拒答、人工转接	胡乱承诺、错误政策、越权查询
企业知识库	让员工快速查制度、接口、项目文档	文档切分、权限过滤、重排、引用	资料过期、权限泄露、答非所问
合同/票据抽取	从非结构化文档提取字段	多模态、OCR、结构化输出、人工复核	金额、日期、主体识别错误
医疗/政务辅助	提供资料检索和填报辅助	私有化部署、脱敏、审计、RAG	高风险结论不能由模型直接决定
工单质检	检查客服、运维、交付过程是否合规	分类、摘要、规则+模型混合	模型误判导致绩效争议
数据分析助手	自然语言生成 SQL 和解释报表	Tool Calling、SQL 安全、权限控制	越权查数、慢 SQL、错误口径
业务 Agent	查询订单、创建工单、发送通知	工具调用、工作流、幂等、确认机制	误操作真实业务
研发助手	解释代码、生成单测、定位日志	RAG、代码检索、沙箱执行	生成不安全代码、泄露源码

场景一：智能客服

智能客服最适合回答“规则稳定、资料明确、重复率高”的问题，例如退换货规则、会员权益、发票开具、物流查询、系统使用帮助。

不能把用户问题直接丢给模型。正确链路是：

flowchart TD
    A["用户咨询"] --> B["识别问题类型"]
    B --> C{"是否需要查询业务数据"}
    C -- "不需要" --> D["检索知识库政策文档"]
    C -- "需要" --> E["调用订单/工单/会员接口"]
    D --> F["生成带引用的回答"]
    E --> F
    F --> G{"置信度是否足够"}
    G -- "足够" --> H["返回用户"]
    G -- "不足" --> I["转人工或澄清问题"]

关键设计：

1. 政策类问题必须带引用来源。
2. 订单、退款、账户信息必须先鉴权。
3. 模型不能自行承诺赔偿、退款、改价。
4. 低置信度要澄清或转人工。
5. 用户反馈要进入评估集，用来持续改进。

场景二：企业知识库问答

企业知识库常见资料包括制度文档、接口文档、项目方案、运维手册、FAQ、产品说明、培训材料。

核心问题不是“能不能搜到”，而是：

1. 用户有没有权限看这份资料。
2. 检索片段是否真的包含答案。
3. 资料是否过期。
4. 回答是否基于资料，而不是模型自由发挥。

flowchart TD
    A["上传企业文档"] --> B["清洗和切分"]
    B --> C["记录部门、角色、密级、版本"]
    C --> D["生成 Embedding"]
    D --> E["写入向量库"]
    F["用户提问"] --> G["按用户权限生成过滤条件"]
    G --> H["向量检索"]
    H --> I["重排和去重"]
    I --> J["组装带引用 Prompt"]
    J --> K["模型回答"]

如果不做权限过滤，最严重的问题不是答错，而是把本不该给用户看的合同、薪酬、客户资料、源代码、病历、项目报价泄露出去。

场景三：合同和票据结构化抽取

很多商业系统需要从合同、发票、报价单、病历、报告、工单描述中抽取字段。

例如合同抽取：

字段	说明
contractNo	合同编号
partyA	甲方
partyB	乙方
amount	合同金额
currency	币种
startDate	生效日期
endDate	终止日期
riskClauses	风险条款

结构化抽取不能只看模型输出，要做校验：

flowchart TD
    A["上传合同"] --> B["OCR 或文本解析"]
    B --> C["模型抽取 JSON"]
    C --> D["字段格式校验"]
    D --> E{"金额、日期、主体是否可靠"}
    E -- "可靠" --> F["写入业务表"]
    E -- "不可靠" --> G["人工复核"]
    G --> F

高价值字段必须有人审，尤其是金额、日期、主体、违约责任、付款条件。

场景四：数据分析助手

数据分析助手不是让模型随便写 SQL，而是让模型在受控条件下把自然语言转换为安全查询。

常见流程：

flowchart TD
    A["用户提问：本月订单转化率"] --> B["识别指标口径"]
    B --> C["检查用户数据权限"]
    C --> D["选择允许访问的表和字段"]
    D --> E["生成 SQL 草稿"]
    E --> F["SQL 安全检查"]
    F --> G{"是否通过"}
    G -- "通过" --> H["执行只读查询"]
    G -- "不通过" --> I["拒绝或要求改写"]
    H --> J["解释结果和口径"]

必须限制：

1. 只允许 SELECT。
2. 必须加时间范围和 LIMIT。
3. 用户只能查询自己有权限的数据域。
4. 慢 SQL 要拦截。
5. 指标口径要来自指标平台或数据字典，不能让模型编。

场景五：业务 Agent 自动化

Agent 适合处理多步骤任务，例如：

1. 查询客户最近订单。
2. 判断是否满足补发条件。
3. 创建售后工单。
4. 发送短信通知。
5. 记录处理日志。

但是商业 Agent 必须加确认和幂等。

sequenceDiagram
    participant U as 用户
    participant A as Agent
    participant S as 业务服务
    participant R as 审计记录
    U->>A: 帮客户创建补发工单
    A->>S: 查询订单和售后规则
    S-->>A: 返回订单状态和规则
    A-->>U: 展示计划操作并请求确认
    U->>A: 确认执行
    A->>S: 创建工单，携带幂等号
    S-->>A: 返回工单号
    A->>R: 记录工具调用和结果
    A-->>U: 返回处理结果

没有确认机制的 Agent 只能做低风险辅助，不能直接执行会改变业务状态的操作。

生产级架构怎么拆

模块	作用
AI Gateway	统一接收 AI 请求，做鉴权、限流、审计、模型路由
Prompt 管理	管理模板、版本、变量、灰度和回滚
RAG 服务	负责文档切分、向量化、检索、重排、引用
Tool 服务	暴露可调用工具，并做权限、参数、幂等校验
评估服务	管理评估集、自动评分、人工反馈、回归测试
安全服务	敏感词、脱敏、Prompt 注入检测、输出过滤
观测服务	记录 Token、耗时、失败率、命中率、成本

flowchart TD
    A["业务前端"] --> B["AI Gateway"]
    B --> C["Prompt 管理"]
    B --> D["RAG 服务"]
    B --> E["Tool 服务"]
    B --> F["模型供应商或本地模型"]
    B --> G["安全治理"]
    B --> H["日志和评估"]

数据表设计参考

表	作用
ai_request_log	记录请求、用户、场景、模型、耗时、Token、结果状态
ai_prompt_version	管理 Prompt 模板、版本、灰度范围
ai_knowledge_doc	文档元数据、部门、密级、版本、状态
ai_knowledge_chunk	文档片段、向量 ID、标题、位置、权限标签
ai_tool_call_log	工具调用参数、结果、幂等号、执行人、审计状态
ai_eval_case	评估样例、标准答案、场景、风险等级
ai_feedback	用户点赞、点踩、纠错、人工标注

这些表不是一开始都要做完整，但商业项目至少要有请求日志、知识权限、工具调用审计和反馈闭环。

Demo：最小商业 AI 网关

下面用 Python 写一个最小 AI 网关，演示商业项目里常见的“鉴权、场景路由、风险控制、日志记录”结构。真实项目可以把 call_model 替换成具体模型 API。

from dataclasses import dataclass
from typing import Literal


Scene = Literal["customer_service", "knowledge_base", "data_analysis"]


@dataclass
class User:
    user_id: str
    roles: set[str]
    department: str


@dataclass
class AIRequest:
    user: User
    scene: Scene
    question: str


def check_permission(req: AIRequest) -> None:
    if req.scene == "data_analysis" and "analyst" not in req.user.roles:
        raise PermissionError("没有数据分析权限")


def detect_risk(question: str) -> bool:
    risky_words = ["删除数据", "导出全部客户", "绕过权限", "修改价格"]
    return any(word in question for word in risky_words)


def call_model(prompt: str) -> str:
    return "这是模型根据受控上下文生成的回答。"


def build_prompt(req: AIRequest) -> str:
    return f"""
你是企业 AI 助手。
场景：{req.scene}
用户部门：{req.user.department}
要求：
1. 不要编造没有依据的信息。
2. 涉及权限、金额、订单状态时，必须提示以业务系统为准。
3. 不确定时要求用户补充信息。

用户问题：{req.question}
""".strip()


def handle_ai_request(req: AIRequest) -> dict:
    check_permission(req)

    if detect_risk(req.question):
        return {
            "answer": "这个操作风险较高，需要人工审批或管理员处理。",
            "risk": True,
        }

    prompt = build_prompt(req)
    answer = call_model(prompt)

    return {
        "answer": answer,
        "risk": False,
        "scene": req.scene,
    }


request = AIRequest(
    user=User(user_id="u1001", roles={"employee"}, department="售后部"),
    scene="customer_service",
    question="客户问发票多久能开出来，怎么回答？",
)

print(handle_ai_request(request))

这个 Demo 的重点不是模型调用，而是商业 AI 必须把权限、风险、场景和日志放到模型前后。

Demo：合同字段结构化抽取

结构化抽取要让模型输出固定 JSON，并在后端做二次校验。

import json
from pydantic import BaseModel, Field, ValidationError


class ContractInfo(BaseModel):
    contract_no: str = Field(description="合同编号")
    party_a: str = Field(description="甲方")
    party_b: str = Field(description="乙方")
    amount: float = Field(description="合同金额")
    currency: str = Field(description="币种")
    start_date: str = Field(description="生效日期，格式 yyyy-MM-dd")
    end_date: str = Field(description="终止日期，格式 yyyy-MM-dd")


def mock_model_extract(contract_text: str) -> str:
    return json.dumps({
        "contract_no": "HT-2026-001",
        "party_a": "甲方科技有限公司",
        "party_b": "乙方数据服务有限公司",
        "amount": 120000.00,
        "currency": "CNY",
        "start_date": "2026-01-01",
        "end_date": "2026-12-31",
    }, ensure_ascii=False)


def extract_contract(contract_text: str) -> ContractInfo:
    raw_json = mock_model_extract(contract_text)
    try:
        return ContractInfo(**json.loads(raw_json))
    except ValidationError as exc:
        raise ValueError(f"合同字段抽取结果不合格：{exc}") from exc


info = extract_contract("这里是合同正文......")
print(info.dict())

生产环境还要继续校验：金额是否为正、日期是否合理、主体是否和客户档案匹配、是否需要人工复核。

Demo：RAG 回答必须带引用

商业知识库问答里，回答没有引用就很难审计。

def build_rag_prompt(question: str, chunks: list[dict]) -> str:
    refs = []
    for i, chunk in enumerate(chunks, start=1):
        refs.append(f"[资料{i}] 标题：{chunk['title']}\n内容：{chunk['content']}")

    return f"""
你是企业知识库助手。只能基于给定资料回答。
如果资料中没有答案，请回答“资料中未找到依据”，不要编造。
回答末尾必须列出引用的资料编号。

用户问题：
{question}

可用资料：
{chr(10).join(refs)}
""".strip()


chunks = [
    {
        "title": "售后政策",
        "content": "普通商品签收后 7 天内可申请无理由退货，定制商品除外。",
    }
]

prompt = build_rag_prompt("签收 5 天还能退货吗？", chunks)
print(prompt)

这类 Prompt 的核心是把模型限制在资料内回答。不确定就拒答，比编一个看起来很顺的答案更可靠。

商业场景：医疗数据资产 AI 助手

医疗数据采集与资产平台里，AI 助手常见需求不是闲聊，而是回答这些问题：

1. 某个数据集有哪些字段。
2. 字段口径是什么意思。
3. 某个采集任务失败可能是什么原因。
4. 某类数据能不能给某部门使用。
5. 某个接口调用报错怎么排查。

这些问题必须接入企业私有知识库、资产元数据、权限系统和审计系统，不能让模型自由发挥。

正确链路

flowchart TD
    A["用户提问"] --> B["识别用户身份和部门"]
    B --> C["查询用户可访问的数据资产范围"]
    C --> D["生成权限过滤条件"]
    D --> E["在授权范围内检索文档和字段说明"]
    E --> F["重排并保留引用来源"]
    F --> G["构造受控 RAG Prompt"]
    G --> H["调用模型生成回答"]
    H --> I["校验回答是否引用资料"]
    I --> J["记录审计日志"]
    J --> K["返回答案和引用"]

关键原则：

原则	为什么
先鉴权再检索	未授权资料不能进入模型上下文
资料带来源	方便用户核对和后续追责
不确定就拒答	医疗、数据权限、接口故障不能胡编
业务事实查系统	任务状态、资产权限、字段密级以业务库为准
全链路审计	要知道谁问了什么、用了哪些资料、模型答了什么

为什么不能“先全库检索再让模型判断权限”

这是很多 RAG 新手最危险的误区。只要无权限资料进入 Prompt，上下文就已经泄露给模型了。即使你在 Prompt 里写“不要泄露无权限资料”，也不能作为安全边界。

错误链路：

flowchart TD
    A["用户提问"] --> B["全库向量检索"]
    B --> C["召回包含敏感资料的 Chunk"]
    C --> D["把敏感资料放进 Prompt"]
    D --> E["要求模型不要泄露"]
    E --> F["仍可能输出敏感内容或被 Prompt 注入绕过"]

正确做法是：权限过滤发生在检索之前或检索过程中。

flowchart TD
    A["用户提问"] --> B["根据用户生成权限过滤条件"]
    B --> C["只在授权文档集合中检索"]
    C --> D["召回安全 Chunk"]
    D --> E["进入 Prompt"]

Demo：带权限过滤的 RAG 检索

下面 Demo 不依赖真实向量库，用列表模拟检索，重点看权限过滤位置。

from dataclasses import dataclass


@dataclass
class User:
    user_id: str
    department: str
    roles: set[str]


@dataclass
class Chunk:
    chunk_id: str
    title: str
    content: str
    permission_tags: set[str]
    score: float


def build_permission_tags(user: User) -> set[str]:
    tags = {f"dept:{user.department}"}
    for role in user.roles:
        tags.add(f"role:{role}")
    return tags


def can_read(user_tags: set[str], chunk: Chunk) -> bool:
    return len(user_tags & chunk.permission_tags) > 0


def retrieve_with_permission(user: User, question: str, chunks: list[Chunk]) -> list[Chunk]:
    user_tags = build_permission_tags(user)

    # 真实项目里，这一步应该下推到向量库 metadata filter，
    # 而不是先把全量结果拿回来再过滤。
    allowed_chunks = [chunk for chunk in chunks if can_read(user_tags, chunk)]

    # 这里用 score 模拟向量相似度，真实项目由向量库和 rerank 模型给分。
    return sorted(allowed_chunks, key=lambda item: item.score, reverse=True)[:3]


def build_prompt(question: str, chunks: list[Chunk]) -> str:
    if not chunks:
        return "资料中未找到用户有权限访问的依据，请拒答。"

    refs = "\n\n".join(
        f"[{idx}] 标题：{chunk.title}\n内容：{chunk.content}"
        for idx, chunk in enumerate(chunks, start=1)
    )
    return f"""
你是医疗数据资产平台助手。
只能根据参考资料回答，不能编造。
如果资料不足，请回答“当前资料不足，无法确认”。
回答必须列出引用编号。

用户问题：{question}

参考资料：
{refs}
""".strip()


chunks = [
    Chunk(
        chunk_id="c1",
        title="门诊就诊明细字段说明",
        content="patient_id 是患者唯一标识，属于敏感字段，默认仅信息科可见。",
        permission_tags={"dept:信息科", "role:data_admin"},
        score=0.92,
    ),
    Chunk(
        chunk_id="c2",
        title="公开资产说明",
        content="公开资产可被普通业务部门查询，但不包含患者身份标识。",
        permission_tags={"role:employee"},
        score=0.78,
    ),
]

user = User(user_id="u1001", department="业务科", roles={"employee"})
result = retrieve_with_permission(user, "patient_id 字段是什么意思？", chunks)
prompt = build_prompt("patient_id 字段是什么意思？", result)
print(prompt)

这个例子里，业务科普通员工不能检索到 dept:信息科 或 role:data_admin 才能访问的敏感 Chunk。模型没有看到敏感资料，自然也不会基于敏感资料回答。

如果 RAG 答错，怎么排查

RAG 答错不要只说“模型不行”，要按链路拆。

flowchart TD
    A["RAG 回答错误"] --> B{"正确资料是否存在"}
    B -- "不存在" --> C["补文档或同步知识库"]
    B -- "存在" --> D{"是否被切成合理 Chunk"}
    D -- "否" --> E["调整切分策略和标题元数据"]
    D -- "是" --> F{"检索是否召回正确 Chunk"}
    F -- "否" --> G["调整 Embedding、TopK、关键词混合检索"]
    F -- "是" --> H{"重排是否把正确 Chunk 放前面"}
    H -- "否" --> I["增加 rerank 或规则排序"]
    H -- "是" --> J{"Prompt 是否要求基于资料回答"}
    J -- "否" --> K["收紧 Prompt 和拒答规则"]
    J -- "是" --> L["检查模型生成、引用校验和评估样例"]

排查表：

现象	可能原因	处理
完全找不到答案	文档未入库、权限过滤过严、TopK 太小	查入库任务、权限标签、召回日志
找到无关资料	Chunk 太长、标题缺失、只做向量检索	调整切分，加入关键词检索和 rerank
答案和资料矛盾	Prompt 约束弱、资料冲突、模型自由发挥	强制引用，冲突时拒答
越权资料被引用	权限过滤在检索后才做或漏了标签	权限下推到检索条件，补审计
回答没有引用	Prompt 和后处理没要求	输出必须包含引用编号，不合格重试或拒答

RAG 生产日志应该记录什么

生产排查不能只保存最终答案。至少要记录：

字段	作用
request_id	串起一次请求链路
user_id / department	排查权限问题
question	用户原始问题，注意脱敏
query_rewrite	多轮对话改写后的问题
permission_filter	本次检索使用了哪些权限条件
retrieved_chunk_ids	召回了哪些资料
rerank_scores	重排分数
prompt_version	判断是不是提示词变更导致
model	使用的模型版本
input_tokens / output_tokens	成本和截断排查
latency_ms	性能排查
answer	最终答案，必要时脱敏
feedback	用户是否认为有用

没有这些日志，RAG 线上出错只能靠猜。

评估指标

指标	说明
答案准确率	回答是否符合业务事实
引用命中率	引用资料是否真的支持答案
拒答正确率	没有资料时是否能拒答
工具调用成功率	参数是否正确，后端是否执行成功
越权拦截率	非授权请求是否被拦住
平均响应时间	用户能否接受等待
Token 成本	单次请求和总成本是否可控
人工转接率	AI 无法处理时是否合理转人工
用户纠错率	用户反馈错误的比例

没有这些指标，AI 应用上线后就只能靠感觉判断效果。

常见坑

坑	后果	正确做法
只调模型，不做权限	可能泄露内部资料	检索前按用户权限过滤
没有引用来源	错了无法追责	RAG 回答必须带文档引用
工具调用直接执行	可能误删、误改、误发通知	后端做权限、参数、幂等、二次确认
没有评估集	改 Prompt 后不知道变好还是变坏	按场景沉淀评估样例
不限流	成本和并发失控	按用户、场景、模型做限流
没有人工兜底	高风险问题无法处理	低置信度转人工或走审批
把 AI 当强一致系统	订单、库存、金额出错	强一致决策仍由业务系统负责

商业落地路线

建议按风险从低到高落地：

1. 内部知识库问答：先面向员工，风险较低，容易收集反馈。
2. 客服辅助回复：先给客服坐席建议，不直接发给客户。
3. 结构化抽取：先用于预填表单，关键字段人工确认。
4. 数据分析助手：只读查询，严格控制表和字段。
5. Agent 自动化：先做查询类工具，再做需要确认的写操作。
6. 对外用户自助 AI：必须有完善的权限、安全、评估和转人工机制。

关联知识点

知识点	作用
大模型基础	理解 Token、上下文、幻觉和成本
提示词工程	学会把任务、上下文、约束和格式说清楚
RAG 知识库	理解企业资料如何参与回答
RAG 流程	学习文档切分、召回、重排、引用
工具调用	学习模型如何安全调用后端接口
AI 评估	学习如何判断回答是否可上线
AI 安全	学习 Prompt 注入、越权和数据泄露防护
LLMOps	学习上线后的版本、监控、灰度和回滚

小结

商业 AI 的关键不是“模型会不会回答”，而是“回答能不能被业务信任、能不能被审计、成本能不能控制、出错能不能兜底”。真正可落地的方案一定是模型、RAG、工具、权限、评估、安全和人工流程共同组成的工程系统。