Python与AI开发

Python 是 AI 应用开发最常用的语言之一，但“会调用一次模型 API”不等于会做 AI 工程。商业项目中的 AI 开发要把模型、数据、检索、权限、成本、评估、日志、接口、任务和安全串成一条可靠链路。

本页面向应用开发者：重点不是从零训练大模型，而是学会用 Python 构建可上线的 AI 应用，例如企业知识库、医疗数据资产问答、报告解析、智能客服、代码助手、自动分类和数据抽取。

学习目标

学完本页你应该能回答：

1. Python 在 AI 应用里通常负责哪些工作。
2. 模型 API 调用为什么要封装超时、重试、日志和成本统计。
3. Prompt、结构化输出、Tool Calling、RAG 分别解决什么问题。
4. RAG 从文档到答案的每一步怎么工作。
5. 为什么商用 RAG 必须做权限过滤、引用来源和评估。
6. AI 接口为什么要限流、缓存、异步任务和流式输出。
7. 模型回答错了、慢了、贵了、越权了应该怎么排查。
8. 如何写一个最小可运行但具备工程意识的 Demo。

Python 在 AI 项目中的位置

AI 应用不是只有模型。Python 常承担“胶水层”和“评估层”：

环节	Python 做什么
数据处理	文档解析、清洗、切分、去重、脱敏
模型调用	调用 Chat、Embedding、多模态、重排序模型
RAG 检索	生成向量、写入向量库、召回上下文
Web API	用 FastAPI 暴露问答、上传、任务接口
工具调用	让模型调用查询、计算、审批、工单等工具
评估	构造测试集、批量评测、回归对比
监控	记录 token、耗时、错误、命中率、满意度
安全	脱敏、权限过滤、防 Prompt 注入

如果只写一个能聊天的 Demo，上线后常见问题会是：模型胡编、回答慢、成本暴涨、用户看到不该看的文档、接口超时、错误没有日志、效果变差没人知道。

AI 应用整体链路

flowchart TD
    A["明确业务问题"] --> B["整理数据和权限"]
    B --> C["设计 Prompt 和输出格式"]
    C --> D{"是否需要私有知识"}
    D -- "需要" --> E["构建 RAG 检索链路"]
    D -- "不需要" --> F["直接调用模型"]
    E --> G["组装上下文和问题"]
    F --> H["模型生成答案"]
    G --> H
    H --> I["解析结构化输出"]
    I --> J["安全检查和引用校验"]
    J --> K["返回 API 或写入任务结果"]
    K --> L["记录日志 成本 评估数据"]

这条链路里任何一环缺失，都会影响商用效果。例如没有评估，就不知道改 Prompt 后是变好了还是变差了；没有权限过滤，RAG 可能召回用户无权查看的文档；没有日志，模型偶发失败时无法定位。

常用库和职责

方向	常用库	作用
HTTP 调用	requests、httpx	调用模型 API 或外部服务
Web API	FastAPI	封装 AI 接口、流式响应、任务查询
数据处理	pandas、numpy	清洗表格、统计评估结果
文档解析	pypdf、python-docx、openpyxl	读取 PDF、Word、Excel
向量检索	faiss、chromadb、pymilvus	构建 RAG 向量库
模型生态	transformers、torch	使用开源模型或本地推理
测试评估	pytest、自定义脚本	回归测试和效果评估
服务工程	logging、pydantic	日志、参数校验、结构化数据

框架可以提高效率，但不要一开始就被框架绑住。先理解数据流，再使用 LangChain、LlamaIndex、Spring AI 等编排框架会更稳。

模型 API 调用封装

不要在业务代码里到处直接写 requests.post()。模型调用要封装成 Client，统一处理密钥、超时、错误、日志和成本。

错误做法：

response = requests.post(url, json=payload)
answer = response.json()["choices"][0]["message"]["content"]

问题：

1. 没有超时，接口可能一直卡住。
2. 没有错误处理，失败时直接抛不清楚的异常。
3. 没有记录模型、耗时、token、request_id。
4. API Key 可能写死在代码里。
5. 多处复制，后续改模型平台很痛苦。

推荐封装：

import os
import time
import logging
import requests

logger = logging.getLogger(__name__)


class ModelClient:
    def __init__(self, api_url: str, api_key_env: str = "MODEL_API_KEY"):
        self.api_url = api_url
        self.api_key = os.environ[api_key_env]

    def chat(self, messages: list[dict], model: str, timeout: int = 30) -> str:
        start = time.perf_counter()
        payload = {
            "model": model,
            "messages": messages,
            "temperature": 0.2,
        }
        try:
            response = requests.post(
                self.api_url,
                headers={"Authorization": f"Bearer {self.api_key}"},
                json=payload,
                timeout=timeout,
            )
            response.raise_for_status()
            data = response.json()
            answer = data["choices"][0]["message"]["content"]
            usage = data.get("usage", {})
            cost_ms = int((time.perf_counter() - start) * 1000)
            logger.info(
                "model chat success model=%s cost_ms=%s input_tokens=%s output_tokens=%s",
                model,
                cost_ms,
                usage.get("prompt_tokens"),
                usage.get("completion_tokens"),
            )
            return answer
        except requests.Timeout as exc:
            logger.exception("model chat timeout model=%s", model)
            raise RuntimeError("模型调用超时") from exc
        except requests.HTTPError as exc:
            logger.exception("model chat http error model=%s", model)
            raise RuntimeError("模型接口返回异常") from exc

密钥必须从环境变量读取，不要提交到 Git。日志里也不要打印完整 prompt、token、患者隐私或业务敏感数据。

Prompt 不只是提示词

Prompt 是对模型的任务约束。一个商用 Prompt 通常包括：

部分	作用
角色	告诉模型以什么身份处理
任务	明确要做什么
背景	给出业务上下文
输入	用户问题、文档片段、结构化数据
约束	不知道就拒答、只能基于资料、输出格式
示例	给出少量输入输出样例

示例：

你是医疗数据资产平台的知识库助手。
任务：根据给定资料回答用户问题。
约束：
1. 只能基于资料回答。
2. 资料不足时回答“当前资料不足，无法确认”。
3. 不要编造系统、接口或字段。
4. 输出必须包含 answer 和 references。

资料：
{contexts}

问题：
{question}

为什么要强调“不知道就拒答”？因为大模型默认倾向于生成一个看起来合理的答案。如果业务要求严谨，就必须把拒答策略写进 Prompt，并在评估中检查。

结构化输出

AI 输出如果直接给用户看，可以是自然语言；如果要进入系统流程，最好是 JSON。

场景：从用户描述中抽取工单信息。

import json


def build_extract_prompt(text: str) -> str:
    return f"""
请从文本中抽取工单信息，只返回 JSON，不要输出其他文字。
字段：
- title: 工单标题
- priority: low/middle/high
- department: 处理部门

文本：
{text}
"""


def parse_json_output(output: str) -> dict:
    try:
        data = json.loads(output)
    except json.JSONDecodeError as exc:
        raise ValueError("模型输出不是合法 JSON") from exc
    required_fields = {"title", "priority", "department"}
    missing = required_fields - set(data)
    if missing:
        raise ValueError(f"模型输出缺少字段：{missing}")
    return data

结构化输出要做二次校验。不能因为模型说“这是 JSON”，就直接入库或执行业务动作。

RAG 是什么

RAG 是 Retrieval-Augmented Generation，检索增强生成。它解决的问题是：模型本身不知道企业私有文档，或者模型知识不够新、不够准，所以先从知识库检索相关资料，再让模型基于资料回答。

RAG 不是“把所有文档塞给模型”。上下文窗口有限，直接塞大量文档会导致成本高、噪声多、答案不稳定。

RAG 完整链路：

flowchart TD
    A["原始文档"] --> B["解析文本"]
    B --> C["清洗和脱敏"]
    C --> D["切分 chunk"]
    D --> E["生成 Embedding"]
    E --> F["写入向量库"]
    G["用户问题"] --> H["问题改写和权限识别"]
    H --> I["问题 Embedding"]
    I --> J["按权限过滤后检索"]
    J --> K["召回候选片段"]
    K --> L["重排序和去重"]
    L --> M["组装 Prompt"]
    M --> N["模型生成答案"]
    N --> O["返回答案和引用"]

每一步都可能出问题：

步骤	常见问题	后果
文档解析	PDF 表格、扫描件解析差	知识源错误
文本切分	chunk 太小或太大	召回丢上下文或噪声多
Embedding	模型不适合中文或业务术语	相似度不准
权限过滤	检索后才过滤或不做过滤	可能越权
topK	太少或太多	漏资料或干扰模型
Prompt	没要求基于资料	模型胡编
引用	不返回来源	用户无法验证

RAG 极简 Demo

下面用关键词模拟检索，帮助理解 RAG 数据流。真实项目中会把 retrieve() 换成向量库检索。

documents = [
    {"id": "doc-1", "text": "FastAPI 可以自动生成 OpenAPI 文档。", "dept": "dev"},
    {"id": "doc-2", "text": "RAG 会先检索知识片段，再调用大模型生成答案。", "dept": "ai"},
    {"id": "doc-3", "text": "资产导入失败时要记录 batch_id 和失败行号。", "dept": "data"},
]


def retrieve(question: str, allowed_depts: set[str]) -> list[dict]:
    result = []
    for doc in documents:
        if doc["dept"] not in allowed_depts:
            continue
        if any(word in doc["text"] for word in question):
            result.append(doc)
    return result[:3]


def rag_answer(question: str, allowed_depts: set[str], model_client: ModelClient) -> dict:
    chunks = retrieve(question, allowed_depts)
    if not chunks:
        return {
            "answer": "当前资料不足，无法确认。",
            "references": [],
        }

    contexts = "\n".join(f"[{item['id']}] {item['text']}" for item in chunks)
    messages = [
        {"role": "system", "content": "你只能基于资料回答，资料不足就拒答。"},
        {"role": "user", "content": f"资料：\n{contexts}\n\n问题：{question}"},
    ]
    answer = model_client.chat(messages, model="demo-model")
    return {
        "answer": answer,
        "references": [item["id"] for item in chunks],
    }

这里故意加入 allowed_depts，是为了强调：商用 RAG 必须先按用户权限限制可检索范围，再召回文档。不能先召回全部文档再让模型自己判断权限。

Embedding 和向量检索原理

Embedding 是把文本变成一串数字向量。语义相近的文本，向量距离通常更近。

flowchart TD
    A["文本：如何排查接口慢"] --> B["Embedding 模型"]
    B --> C["向量 0.12 0.87 ..."]
    D["用户问题：接口响应很慢怎么办"] --> E["Embedding 模型"]
    E --> F["问题向量"]
    F --> G["向量库相似度搜索"]
    C --> G
    G --> H["返回语义相近的文档片段"]

常见相似度：

方法	含义
余弦相似度	看两个向量方向是否接近
点积	常用于归一化后的向量相似度
欧氏距离	看两个点的距离

为什么不是简单关键词搜索？因为用户可能问“接口卡顿怎么处理”，文档写的是“响应时间过长排查流程”。关键词不同，但语义相近，Embedding 能更容易召回。

Tool Calling 思路

Tool Calling 是让模型在需要时调用系统工具，例如查询订单、查库存、创建工单、计算价格。

核心原则：模型只负责“判断是否需要调用工具和生成参数”，真正执行工具的是后端代码。

flowchart TD
    A["用户问题"] --> B["模型判断是否调用工具"]
    B --> C{"需要工具吗"}
    C -- "否" --> D["直接回答"]
    C -- "是" --> E["输出工具名和参数"]
    E --> F["后端校验权限和参数"]
    F --> G["执行工具"]
    G --> H["工具结果返回模型"]
    H --> I["模型生成最终回答"]

工具调用 Demo：

TOOLS = {
    "query_asset": lambda asset_id: {"id": asset_id, "name": "检验报告"},
}


def execute_tool(tool_name: str, arguments: dict, current_user: dict) -> dict:
    if tool_name not in TOOLS:
        raise ValueError("工具不存在")
    if "asset:read" not in current_user["permissions"]:
        raise PermissionError("无权查询资产")
    if tool_name == "query_asset":
        asset_id = int(arguments["asset_id"])
        return TOOLS[tool_name](asset_id)
    raise ValueError("工具参数不支持")

高风险工具，例如删除数据、发起支付、修改权限，不能让模型直接执行。必须有人确认、权限校验、审计日志和幂等控制。

封装成 FastAPI

AI 接口要比普通 CRUD 更重视超时、限流、日志、脱敏和异步。

import time
from fastapi import FastAPI, Depends
from pydantic import BaseModel, Field

app = FastAPI()


class ChatRequest(BaseModel):
    question: str = Field(min_length=1, max_length=2000)


class ChatResponse(BaseModel):
    answer: str
    references: list[str]
    cost_ms: int


def get_current_user():
    return {"id": 1, "permissions": ["dept:ai", "asset:read"]}


@app.post("/ai/chat", response_model=ChatResponse)
def chat_api(request: ChatRequest, current_user: dict = Depends(get_current_user)):
    start = time.perf_counter()
    allowed_depts = {"ai"}
    result = rag_answer(request.question, allowed_depts, model_client)
    return ChatResponse(
        answer=result["answer"],
        references=result["references"],
        cost_ms=int((time.perf_counter() - start) * 1000),
    )

真实项目还要增加：

1. 用户维度限流。
2. 请求日志和 token 统计。
3. 输入敏感信息脱敏。
4. 输出安全检查。
5. 模型调用超时和降级。
6. 长任务改成任务 ID 轮询。

流式输出和任务轮询

AI 接口可能很慢。两种常见方案：

方案	适合	特点
流式输出	聊天、生成文本	用户可以边生成边看
任务轮询	文件解析、批量评估、长报告	提交后查状态，不占用长连接

流式响应示例：

import asyncio
from fastapi.responses import StreamingResponse


async def fake_ai_stream():
    for word in ["正在", "检索", "知识库", "并生成答案"]:
        yield f"data: {word}\n\n"
        await asyncio.sleep(0.2)


@app.get("/ai/chat/stream")
async def chat_stream():
    return StreamingResponse(fake_ai_stream(), media_type="text/event-stream")

任务轮询流程：

flowchart TD
    A["提交文件解析任务"] --> B["返回 task_id"]
    B --> C["后台处理文件和模型调用"]
    B --> D["前端轮询任务状态"]
    C --> E["写入任务结果"]
    D --> F["完成后读取结果"]

评估 AI 效果

AI 不能只靠“我试了几个问题感觉还行”。要有评估集。

评估数据示例：

eval_cases = [
    {
        "id": "case-001",
        "question": "FastAPI 为什么会返回 422？",
        "expected_keywords": ["参数校验", "Pydantic", "请求不符合接口契约"],
        "expected_doc_ids": ["python-web-api"],
    },
    {
        "id": "case-002",
        "question": "没有权限的用户能查财务文档吗？",
        "must_refuse": True,
        "expected_keywords": ["无权限", "无法查看"],
    },
]

简单评估脚本：

def evaluate_answer(answer: str, case: dict, references: list[str]) -> dict:
    keyword_hit = all(word in answer for word in case.get("expected_keywords", []))
    doc_hit = set(case.get("expected_doc_ids", [])).issubset(set(references))
    refuse_ok = True
    if case.get("must_refuse"):
        refuse_ok = "无权限" in answer or "无法查看" in answer
    return {
        "case_id": case["id"],
        "keyword_hit": keyword_hit,
        "doc_hit": doc_hit,
        "refuse_ok": refuse_ok,
        "passed": keyword_hit and doc_hit and refuse_ok,
    }

评估维度：

维度	看什么
答案正确性	是否答到关键点
召回质量	是否检索到正确文档
引用一致性	答案是否基于引用资料
拒答能力	资料不足或无权限时是否拒答
格式稳定	JSON 或模板是否稳定
成本耗时	token、延迟是否可接受

成本和性能控制

AI 成本通常来自 token 和模型调用次数。常见优化：

问题	优化
Prompt 太长	压缩上下文、减少 topK、摘要
重复问题反复调用	缓存相同问题或检索结果
模型太贵	简单任务用小模型，复杂任务用大模型
响应太慢	流式输出、异步任务、减少上下文
并发太高	用户限流、队列削峰

日志至少记录：

request_id user_id model input_tokens output_tokens cost_ms status error_code

没有这些字段，就无法回答“今天成本为什么涨了”“哪个接口最慢”“哪个用户调用最多”。

安全和权限

AI 商用安全重点：

1. API Key 不进前端，不进 Git。
2. 用户输入和日志要脱敏。
3. RAG 检索前做权限过滤。
4. Prompt 注入不能改变系统边界。
5. Tool Calling 必须做权限、参数、审计。
6. 高风险动作必须二次确认。
7. 模型输出不能直接执行 SQL 或系统命令。

RAG 权限过滤错误做法：

先检索全部文档 -> 拼给模型 -> 让模型自己不要泄露

正确做法：

先根据用户权限确定可见文档范围 -> 只在可见范围内检索 -> 再给模型

生产级模型调用链路

很多同学第一次写 AI 功能时，会把模型调用写成“Controller 里拼 Prompt，然后直接请求模型接口”。这能跑 Demo，但不适合生产。生产链路至少要把请求校验、模型客户端、重试、降级、日志、成本、审计和错误处理拆清楚。

flowchart TD
    A["前端或业务系统请求"] --> B["API 入参校验"]
    B --> C["鉴权、限流、敏感词和脱敏"]
    C --> D["业务 Service 组装任务上下文"]
    D --> E["Prompt 模板渲染"]
    E --> F["ModelClient 统一调用模型"]
    F --> G{"调用是否成功"}
    G -- "成功" --> H["解析模型输出"]
    G -- "失败" --> I["按错误类型重试或降级"]
    I --> J{"是否可恢复"}
    J -- "可恢复" --> F
    J -- "不可恢复" --> K["返回可理解的业务错误"]
    H --> L["输出校验、引用校验、安全过滤"]
    L --> M["返回结果"]
    M --> N["记录 request_id、token、耗时、模型版本"]

关键点不是“多写几层代码”，而是每一层承担不同责任：

层	负责什么	不这样会怎样
API 层	校验问题长度、文件大小、参数格式	超长输入拖垮模型调用，错误参数进入业务
权限安全层	鉴权、限流、脱敏、权限范围	产生越权回答、敏感信息进入日志或模型
Service 层	决定业务流程，选择 RAG、工具或直接问答	业务规则散落在 Controller 和 Prompt 中
Prompt 层	统一管理模板、版本、变量	Prompt 改动无法追踪，效果回退不好定位
ModelClient 层	统一封装模型供应商、超时、重试、日志	到处复制调用代码，换模型和排查问题很痛
后处理层	JSON 校验、引用校验、风险过滤	模型输出格式错也被当成正常结果入库

ModelClient 为什么要统一封装

模型服务有几个和普通 HTTP 接口不同的特点：

1. 耗时更长：大模型生成文本通常比普通 CRUD 接口慢，必须设置超时和流式输出。
2. 成本可变：输入越长、输出越长、模型越大，成本越高，必须记录 token。
3. 结果不完全确定：同样输入可能有细微差异，所以要做结构化校验和评估。
4. 供应商可能切换：今天用云模型，明天可能用私有模型，业务层不能绑定某一家接口格式。
5. 错误类型复杂：限流、超时、内容安全拦截、上下文超长、JSON 格式错误要分开处理。

下面是一个更接近生产写法的模型客户端骨架。它没有绑定具体供应商，重点是结构：

import logging
import os
import time
from dataclasses import dataclass
from typing import Any

import requests

logger = logging.getLogger(__name__)


@dataclass
class ModelResult:
    content: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    cost_ms: int
    request_id: str | None = None


class ModelCallError(RuntimeError):
    pass


class ModelClient:
    def __init__(self, base_url: str, api_key_env: str = "MODEL_API_KEY"):
        self.base_url = base_url.rstrip("/")
        self.api_key = os.environ[api_key_env]

    def chat(self, *, messages: list[dict[str, str]], model: str, timeout: int = 30) -> ModelResult:
        start = time.perf_counter()
        payload: dict[str, Any] = {
            "model": model,
            "messages": messages,
            "temperature": 0.2,
        }

        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json=payload,
                timeout=timeout,
            )
            response.raise_for_status()
            data = response.json()
        except requests.Timeout as exc:
            raise ModelCallError("模型调用超时，请稍后重试") from exc
        except requests.HTTPError as exc:
            status = exc.response.status_code if exc.response is not None else "unknown"
            raise ModelCallError(f"模型接口 HTTP 异常：{status}") from exc
        except requests.RequestException as exc:
            raise ModelCallError("模型网络调用失败") from exc

        usage = data.get("usage", {})
        cost_ms = int((time.perf_counter() - start) * 1000)
        result = ModelResult(
            content=data["choices"][0]["message"]["content"],
            model=model,
            prompt_tokens=int(usage.get("prompt_tokens", 0)),
            completion_tokens=int(usage.get("completion_tokens", 0)),
            cost_ms=cost_ms,
            request_id=data.get("id"),
        )
        logger.info(
            "ai_model_call model=%s cost_ms=%s prompt_tokens=%s completion_tokens=%s request_id=%s",
            result.model,
            result.cost_ms,
            result.prompt_tokens,
            result.completion_tokens,
            result.request_id,
        )
        return result

注意：示例里没有把完整 Prompt 打进日志。生产环境里 Prompt 可能包含客户资料、患者信息、合同内容、账号信息，日志平台通常有很多人能查，直接打印会造成二次泄露。

RAG 关键过程拆解

RAG 看起来是“检索后再生成”，但每一步都有设计取舍。小白最容易误解的是：以为只要把文档向量化就能回答准确。实际上，RAG 的效果往往取决于文档治理、切分、元数据、检索策略和评估，而不是模型本身。

文档解析：先保证知识源是对的

文档解析负责把 PDF、Word、Excel、网页、数据库字段说明转换成可检索文本。解析错了，后面向量化再好也没用。

文档类型	常见问题	处理方式
PDF	表格断行、页眉页脚干扰、扫描件没有文本	使用 PDF 解析 + OCR，清理页眉页脚
Word	标题层级、表格、批注、修订痕迹	保留标题路径，过滤无关批注
Excel	多 sheet、合并单元格、字段说明分散	转成“表名-字段-含义-口径”的结构
数据库元数据	字段注释缺失、同名字段含义不同	补业务口径和系统来源

Chunk 切分：为什么不是越大越好

Chunk 是 RAG 检索的最小片段。切分太小，会丢上下文；切分太大，会引入噪声并增加 token 成本。

flowchart TD
    A["原始文档"] --> B["按标题、段落、表格边界切分"]
    B --> C["给每个 Chunk 加标题路径和来源"]
    C --> D{"Chunk 是否过长"}
    D -- "过长" --> E["按语义继续拆分并保留重叠"]
    D -- "合适" --> F["生成 Embedding"]
    E --> F

常见策略：

策略	适合	风险
固定长度切分	快速入门、纯文本	可能把一句话或表格拆断
按标题切分	制度、手册、接口文档	标题下内容过长时还要二次切
按表格行切分	数据字典、资产目录	需要保留表名、字段名、系统名
重叠切分	上下文连续文本	重叠过大会增加存储和召回重复

商业数据资产问答中，推荐给 Chunk 加元数据：

from dataclasses import dataclass


@dataclass
class DocumentChunk:
    chunk_id: str
    doc_id: str
    title_path: str
    content: str
    system_code: str
    department: str
    permission_tags: set[str]
    version: str

没有 permission_tags，就很难做权限过滤；没有 version，资料更新后就难以判断回答基于旧文档还是新文档；没有 title_path，模型拿到片段后不知道上下文来自哪里。

检索：向量检索为什么还要混合关键词

向量检索擅长语义相似，比如“医保结算字段”和“费用报销口径”可能有关；关键词检索擅长精确匹配，比如表名、字段名、接口名、错误码。

商业系统中很多问题包含精确实体：

patient_id 是什么？
T_ASSET_FIELD 表怎么关联资产表？
ERR_10023 为什么报错？

这类问题只靠向量相似度可能把“患者基本信息”召回到前面，却漏掉真正包含 patient_id 字段定义的 Chunk。因此更常见的生产方案是混合检索：

flowchart TD
    A["用户问题"] --> B["问题改写和实体识别"]
    B --> C["向量检索 TopK"]
    B --> D["关键词/BM25 检索 TopK"]
    C --> E["候选合并去重"]
    D --> E
    E --> F["按权限、版本、业务规则过滤"]
    F --> G["Rerank 重排序"]
    G --> H["取最终上下文"]

如果召回阶段没把正确资料找出来，后面的模型再强也只能“无米之炊”。所以排查 RAG 时必须先看召回日志，而不是先改 Prompt。

Rerank：为什么要二次排序

向量库返回的相似度分数不一定等于“最能回答问题”。Rerank 会把用户问题和候选 Chunk 成对比较，重新排序。

Rerank 适合：

1. TopK 召回比较多，前几条不稳定。
2. 业务文档里相似概念很多，例如“采集任务”“采集规则”“采集日志”。
3. 需要把包含关键字段、标题更匹配、版本更新的片段排前面。

不使用 Rerank 的后果是：模型可能拿到一堆“看着相关但答不到点”的资料，最后回答含糊或者编造。

Tool Calling 生产规则

Tool Calling 是让模型根据用户意图生成工具调用参数，再由后端执行工具。它适合“查订单、查库存、创建工单、计算报价、查询指标”等场景。

但模型不能直接操作真实系统。模型只负责“提出调用建议”，真正执行必须由后端校验。

flowchart TD
    A["用户：帮我创建补采任务"] --> B["模型识别需要调用 create_task"]
    B --> C["生成工具参数"]
    C --> D["后端校验参数、权限、幂等号"]
    D --> E{"是否高风险操作"}
    E -- "是" --> F["要求用户确认"]
    E -- "否" --> G["执行工具"]
    F --> G
    G --> H["记录审计日志"]
    H --> I["把执行结果交给模型总结"]

工具定义示例：

from pydantic import BaseModel, Field, field_validator


class CreateCollectTaskArgs(BaseModel):
    asset_id: str = Field(min_length=1)
    cron: str = Field(description="采集周期表达式")
    target_department: str
    reason: str = Field(min_length=5, max_length=200)

    @field_validator("cron")
    @classmethod
    def validate_cron(cls, value: str) -> str:
        if value.strip() == "* * * * *":
            raise ValueError("禁止创建每分钟执行的高频采集任务")
        return value


def create_collect_task(args: CreateCollectTaskArgs, current_user: dict) -> dict:
    if "task:create" not in current_user["permissions"]:
        raise PermissionError("当前用户没有创建采集任务权限")

    # 真实项目里要写入数据库，并使用 request_id 或业务唯一键做幂等。
    return {
        "task_id": "task-20260706-001",
        "asset_id": args.asset_id,
        "status": "created",
    }

为什么要做这些校验？

校验	目的	不做的后果
参数校验	防止模型生成非法字段、危险频率、错误格式	创建出不可执行或高风险任务
权限校验	确保用户真的能执行该工具	普通用户通过模型绕过权限
幂等校验	防止重试导致重复创建、重复扣款、重复通知	业务状态被重复修改
审计日志	记录谁在什么时候让 AI 做了什么	出问题无法追责和回放
人工确认	高风险动作执行前二次确认	模型误判直接影响真实业务

AI 评估集怎么建设

AI 上线前不能只靠人工随便问几个问题。评估集是用来回答三个问题：

1. 当前版本能不能上线？
2. 改 Prompt、换模型、改切分后有没有变差？
3. 哪类问题最容易失败？

评估集至少包含：

类型	样例	评估重点
正常问答	“LIS 系统有哪些核心资产？”	答案是否正确、引用是否命中
无资料问题	“不存在的字段含义是什么？”	是否拒答
无权限问题	“查询信息科敏感资产字段”	是否拦截越权
易混问题	“采集规则和采集任务区别？”	是否区分相似概念
结构化抽取	“从文本中抽取工单字段”	JSON 是否合法、字段是否完整
工具调用	“创建一个补采任务”	工具参数和权限是否正确

评估流程：

flowchart TD
    A["准备黄金测试集"] --> B["运行当前 Prompt 和模型"]
    B --> C["记录答案、引用、工具调用、耗时、Token"]
    C --> D["自动规则评分"]
    D --> E["人工抽检高风险样例"]
    E --> F{"是否达到上线阈值"}
    F -- "达到" --> G["灰度发布"]
    F -- "未达到" --> H["回到数据、检索、Prompt 或模型调整"]

一个简单评估结果可以长这样：

def score_rag_case(answer: str, references: list[str], case: dict) -> dict:
    keyword_ok = all(word in answer for word in case["must_include"])
    reference_ok = set(case["expected_refs"]).issubset(set(references))
    refuse_ok = True
    if case.get("must_refuse"):
        refuse_ok = "无法确认" in answer or "无权限" in answer
    return {
        "case_id": case["id"],
        "keyword_ok": keyword_ok,
        "reference_ok": reference_ok,
        "refuse_ok": refuse_ok,
        "passed": keyword_ok and reference_ok and refuse_ok,
    }

规则评分不能替代人工评审，但可以快速发现明显回退。比如换了 Embedding 模型后，引用命中率从 86% 掉到 61%，这就是不能直接上线的信号。

常见问题排查

模型胡编

flowchart TD
    A["答案胡编"] --> B{"是否使用 RAG"}
    B -- "否" --> C["增加资料上下文或拒答约束"]
    B -- "是" --> D["检查召回文档是否正确"]
    D --> E{"召回正确吗"}
    E -- "否" --> F["调 chunk embedding topK rerank"]
    E -- "是" --> G["检查 Prompt 是否要求基于资料回答"]

RAG 答错

层	排查点
数据层	文档是否解析正确、是否过期
切分层	chunk 是否割裂上下文
检索层	topK、Embedding、关键词混合检索
权限层	是否过滤掉正确文档
生成层	Prompt 是否约束基于资料
输出层	引用是否和答案一致

接口超时

先看耗时拆分：

1. 文档检索耗时。
2. 重排序耗时。
3. 模型首 token 时间。
4. 模型总生成时间。
5. 网络和网关超时。

优化优先级：减少上下文、设置超时、流式输出、任务轮询、缓存、队列削峰。

商业场景：医疗数据资产问答

需求：用户问“LIS 系统有哪些核心数据资产？字段含义是什么？”系统基于资产目录和数据字典回答。

设计：

模块	设计
文档入库	资产表、字段表、数据字典转成 chunk
权限	用户只能检索自己医院和部门可见资产
检索	资产名称、字段名关键词 + Embedding 混合检索
生成	要求基于资料回答并返回引用
API	/ai/chat 问答，/ai/tasks 长任务
日志	记录 request_id、用户、token、耗时、引用
评估	准备资产问答集，检查引用是否正确

为什么不能只把所有表结构塞给模型？因为数据量大、成本高、上下文混乱，而且容易越权。必须先按权限和问题检索相关片段，再让模型回答。

面试标准回答

Python 在 AI 应用中通常做什么？
Python 常用于模型 API 调用、数据清洗、文档解析、Embedding、向量检索、RAG 后端、FastAPI 接口、评估脚本和监控日志。应用开发者重点不是一开始训练模型，而是把模型能力工程化。

RAG 的完整流程是什么？
离线阶段解析文档、清洗脱敏、切分 chunk、生成 Embedding、写入向量库；在线阶段把用户问题向量化，在权限范围内检索相关片段，重排序后组装 Prompt，调用大模型生成答案，并返回引用来源。

为什么 RAG 要做权限过滤？
因为向量检索可能召回用户无权查看的文档。如果先检索全部再让模型判断权限，敏感内容可能已经进入上下文。正确做法是先根据用户权限限定可检索范围，再在范围内召回文档。

AI 接口为什么要记录 token 和耗时？
token 决定成本，耗时决定用户体验和系统容量。记录模型、输入输出 token、cost_ms、request_id、错误码后，才能分析成本上涨、慢请求、模型失败和用户调用行为。

关联知识点

• AI 大模型基础
• AI RAG
• Embedding 与向量数据库
• AI 评估体系
• AI 安全
• Python Web API开发
• Python 日志与调试
• Python 面试题

小结

Python AI 开发的核心不是“调通一个聊天接口”，而是把模型能力变成可上线的系统：输入要校验，数据要治理，检索要有权限，模型要有超时，输出要能校验，效果要能评估，成本要能统计，问题要能排查。只有这些链路都建立起来，AI 应用才不是 Demo，而是可以进入商业项目的工程能力。