实践环节 · AI 工程课程

准备企业知识资料与运行环境

本练习使用专用飞书知识库中的演示保修政策与多维表格。资料已完成脱敏处理，用户身份与授权名单均限定在演示范围内。读取、检索、生成、评估与只读工具共用同一份资料；公开商品数据集将在Sprint 0《Amazon 电子产品类别数据集概览》作为独立研究基准介绍。

下载贯通示例、环境模板与脱敏练习数据。解压后按 README 执行；站点仅展示教程，API Key、应用密钥和资料由学习者自己的服务端管理。

python3 -m venv .venv
.venv/bin/pip install -r requirements.txt
.venv/bin/python -m unittest test_pipeline.py -v

离线测试采用 HTTP fixture、固定合成向量与真实 Qdrant 内存客户端，用于验证代码流程、消息格式与失败分支；真实的租户权限、模型可用性及检索效果则需在自己的演示环境中验收。

# 仅在学习者自己的服务端填写；课程网页不接收任何凭据。
FEISHU_APP_ID=REPLACE_WITH_APP_ID
FEISHU_APP_SECRET=REPLACE_WITH_APP_SECRET
FEISHU_SPACE_ID=REPLACE_WITH_DEMO_SPACE_ID
FEISHU_BITABLE_TABLE_ID=REPLACE_WITH_DEMO_TABLE_ID
MODEL_REGION=cn-beijing
CHAT_BASE_URL=https://REPLACE_WITH_WORKSPACE_ID.cn-beijing.maas.aliyuncs.com/compatible-mode/v1
CHAT_API_KEY=REPLACE_WITH_SAME_REGION_KEY
CHAT_MODEL=qwen3.8-max
STRUCTURED_BASE_URL=https://REPLACE_WITH_WORKSPACE_ID.cn-beijing.maas.aliyuncs.com/compatible-mode/v1
STRUCTURED_API_KEY=REPLACE_WITH_SAME_REGION_KEY
STRUCTURED_MODEL=qwen3.8-max
TOOLS_BASE_URL=https://REPLACE_WITH_WORKSPACE_ID.cn-beijing.maas.aliyuncs.com/compatible-mode/v1
TOOLS_API_KEY=REPLACE_WITH_SAME_REGION_KEY
TOOLS_MODEL=qwen3.8-max
EMBEDDING_BASE_URL=https://REPLACE_WITH_WORKSPACE_ID.cn-beijing.maas.aliyuncs.com/compatible-mode/v1
EMBEDDING_API_KEY=REPLACE_WITH_SAME_REGION_KEY
EMBEDDING_MODEL=text-embedding-v4
EMBEDDING_DIMENSIONS=1024
QDRANT_URL=http://127.0.0.1:6333
QDRANT_COLLECTION=course_feishu_v1
LANGFUSE_BASE_URL=http://127.0.0.1:3000
LANGFUSE_PUBLIC_KEY=REPLACE_WITH_PROJECT_PUBLIC_KEY
LANGFUSE_SECRET_KEY=REPLACE_WITH_PROJECT_SECRET_KEY
PROMPT_VERSION=1
DATASET_VERSION=REPLACE_WITH_DATASET_VERSION_TIMESTAMP
DINGTALK_APP_KEY=REPLACE_WITH_APP_KEY
DINGTALK_APP_SECRET=REPLACE_WITH_APP_SECRET
DINGTALK_OPERATOR_UNION_ID=REPLACE_WITH_DEMO_OPERATOR_UNION_ID

读取、分块与检索

企业知识问答贯通流程 — 飞书知识资料 → 读取与分块 → 独立 embedding → Qdrant + ACL → 结构化生成 → 评估与只读工具

飞书 Wiki 节点记录了实际资源类型（obj_type）与资源标识（obj_token）：docx 调用文档块 API，doc（适配器内部同样兼容 document 名称）调用旧版纯文本接口，bitable 则调用多维表格记录查询接口。分页读取时需根据 has_more 和 page_token 持续翻页，即便遇到空的 items 页也需继续拉取；子节点则通过递归方式枚举。

def read_node(self, node, table_id=None):
        kind, token = node['obj_type'], token_path(node['obj_token'])
        if kind == 'docx':
            blocks = list(self.pages('GET', f'/open-apis/docx/v1/documents/{token}/blocks', 500))
            text, omissions = blocks_to_text(blocks)
            return text, omissions
        if kind in {'doc', 'document'}:
            data = self.request('GET', f'/open-apis/doc/v2/{token}/raw_content')
            return data['content'], ['旧版文档纯文本缺少表格结构与附件内容']
        if kind == 'bitable':
            if not table_id:
                raise SourceError('bitable_table_id_required')
            records = list(self.pages('POST', f'/open-apis/bitable/v1/apps/{token}/tables/{token_path(table_id)}/records/search',
                                      500, body={'automatic_fields': False}))
            return '\n'.join(json.dumps({'record_id': r['record_id'], 'fields': r['fields']}, ensure_ascii=False) for r in records), []
        raise SourceError(f'unsupported_obj_type={kind}')

相关接口文档参见：Wiki 节点、docx 全部块、多维表格记录与应用 token。应用需同时具备相应 API 权限及目标资料的访问授权。注意，docx 读取接口受应用级 5 次/秒的频控限制，多进程部署时需使用共享限流器；HTTP 400 响应中若返回 code=99991400，同样需按限流处理。

解析文档块树时需保留原始文档顺序、block_id 及表格单元格内容。内嵌电子表格、图片、附件或未知块元素统一归入 omissions 列表；一旦发现读取内容不完整，立即停止对该资料建立索引。多维表格若开启了高级权限，接口可能返回空结果，需针对专用演示表单独核验；查询时设置单页大小 500、并将 automatic_fields 设为 false。此外，即便后端应用自身有权限读取资料，在检索时仍必须独立判定当前提问用户的访问权限。

def chunk_text(source_id, text, size=600, overlap=80):
    if size <= overlap or overlap < 0:
        raise ValueError('invalid_chunk_window')
    digest = hashlib.sha256(text.encode()).hexdigest()
    chunks = []
    for start in range(0, len(text), size - overlap):
        part = text[start:start + size]
        chunk_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f'{source_id}:{digest}:{start}'))
        chunks.append({'id': chunk_id, 'source_id': source_id, 'content_hash': digest, 'text': part})
        if start + size >= len(text):
            break
    return chunks

class Embedding:
    def __init__(self, client, model, dimensions):
        self.client, self.model, self.dimensions = client, model, dimensions

    def encode(self, texts):
        output = []
        for offset in range(0, len(texts), 10):
            batch = texts[offset:offset + 10]
            result = self.client.embeddings.create(model=self.model, input=batch, dimensions=self.dimensions,
                                                   encoding_format='float')
            vectors = [d.embedding for d in sorted(result.data, key=lambda d: d.index)]
            if len(vectors) != len(batch) or any(len(v) != self.dimensions for v in vectors):
                raise ValueError('embedding_dimension_or_count_mismatch')
            output.extend(vectors)
        return output

文档索引与在线查询使用相同的 embedding 模型及 1024 维配置。每次向量化请求最多提交 10 条文本，并校验返回向量数量与维度是否匹配；输入 token 长度也需对照所选地域的模型限制进行检查。更换模型或调整向量维度后，需全量重建索引。Qdrant 可采用本地或自托管方式部署，数据库地址由服务端统一管理。

def retrieve(self, principal, question, limit=4):
        allowed = self.acl.visible_ids(principal)
        if not allowed:
            return []
        vector = self.embedding.encode([question])[0]
        points = self.client.query_points(self.collection, query=vector, limit=limit, with_payload=True,
            query_filter=models.Filter(must=[
                models.FieldCondition(key='tenant', match=models.MatchValue(value=principal.tenant)),
                models.FieldCondition(key='source_id', match=models.MatchAny(any=allowed))])).points
        # 模型收到内容之前复核，阻止刚被撤回权限的内容进入上下文。
        return [p.payload for p in points if self.acl.allowed(principal, p.payload['source_id'])]

运行自己的后端与界面

完成环境配置与演示资料授权后，在本地开发环境运行 live.py。该脚本会完整枚举知识库、读取资料并更新 Qdrant 索引，随后执行结构化问答。资料读取器与模型调用逻辑均保留在 FastAPI 等后端服务中；Streamlit 前端只负责提交问题并展示答案与引用。

.venv/bin/python live.py

每个分块的稳定索引 ID 由来源 ID（source_id）、内容哈希（content_hash）和分块起始位置（start）共同生成。单写入者更新时会自动清理旧分块；完整枚举成功后，通过比对 indexed-sources.json 删除已下线或消失的资料。若分页拉取、网络请求或限流重试中途失败，则立即终止当前批次，避免误删数据。在生产环境中，还需要引入暂存版本、原子切换、并发控制、权限撤回即时清理与定时全量对账——不能仅依赖事件通知来保证所有资料删除与继承 ACL 变更都已同步。

记录脱敏追踪

每次请求的追踪记录仅保存问题 ID、资料 ID、语料版本、模型名称、耗时、token 用量与错误类型。如需接入可观测性平台，可使用自托管的 Langfuse（接入方式参考官方追踪接口）。访问凭据、完整原文资料和个人敏感信息一律不写入日志，所有上报内容均需符合实际数据使用授权。

建立 JSONL 评估数据集

{"id":"warranty","input":{"question":"演示设备的保修期限是多少？"},"expected_output":{"source_ids":["feishu:docx:demo-policy"],"answer":"演示设备保修一年。"},"metadata":{"synthetic":true,"corpus_version":"demo-v1"}}
{"id":"unknown","input":{"question":"未发布型号的上市日期是什么？"},"expected_output":{"source_ids":[],"answer":"当前授权资料中没有足够依据。"},"metadata":{"synthetic":true,"corpus_version":"demo-v1"}}

每行保存 input.question、expected_output.source_ids、expected_output.answer 与语料版本。人工核验参考答案和引用，加入无权限、缺少依据、更新后旧答案、删除和限流案例。合成数据只是起点，需要按真实问题分布补充。

评估一个案例

def retrieval_score(expected_ids, contexts):
    expected, found = set(expected_ids), {c['source_id'] for c in contexts}
    if not expected:
        return {'precision': float(not found), 'recall': float(not found)}
    matches = len(expected & found)
    return {'precision': matches / len(found) if found else 0, 'recall': matches / len(expected)}

该函数在资料 ID（source_id）粒度上计算 Precision 与 Recall。完整的系统评估还需进一步检查引用 ID 是否真实合法、答案内容是否受检索材料支撑，以及无依据时是否正确拒答。由于向量 top-k 搜索总会返回距离最近的候选结果，针对未知问题需额外设置相似度阈值或独立的相关性判定模块；评估集中的“未发布型号”案例即用于检验这一边界。

固定版本并评估完整数据集

本地 JSONL 评估是基础评测路径。若选用可选的 Langfuse 评估路径，需遵循真实的 dataset item 字段及版本时间戳规范（示例固定使用 Langfuse SDK 4.17.0）。提示词采用整数版本号管理，每次实验记录同步保存语料 hash、模型名称、调用参数与数据集版本时间戳。

def upload_dataset():
    langfuse = get_client()
    langfuse.create_dataset(name='course-feishu-qa', metadata={'corpus_version': 'demo-v1'})
    for line in Path('fixtures/evaluation.jsonl').read_text().splitlines():
        item = json.loads(line)
        langfuse.create_dataset_item(dataset_name='course-feishu-qa', id=item['id'],
                                    input=item['input'], expected_output=item['expected_output'], metadata=item['metadata'])
    langfuse.flush()

采用 RAGAS 0.2.15 路径时，需显式配置 judge 与 embedding 模型（官方模型配置说明了二者的自定义方法）。下载包中的 ragas_eval.py 接收 user_input、response、retrieved_contexts 和 reference 四个字段；由于调用 judge 会产生真实的模型推理请求，请结合自己的预算与数据授权范围执行。

练习一：让 Wiki 第一页为空且 has_more=true，确认仍读取第二页。
练习二：撤回 demo-user 对保修资料的访问授权，确认向量索引查询与只读工具均不再返回该资料正文。
练习三：将保修期由“一年”更新为“两年”，随后删除该份资料，确认旧分块已被清理。
练习四：在 JSONL 评估集中加入未知问题，设计拒答阈值并报告该阈值对整体召回率的影响。
练习五：运行只读工具与 MCP 链路，对比同一查询返回的引用 ID 与普通 RAG 链路是否一致。

chunk_text 中的 size=600 与 overlap=80 以 Python 字符计数，不是 tokenizer token 数。实际送入 embedding 或生成模型前，还需按所选模型的 tokenizer 与输入上限检查长度。飞书接口中的 docx 指飞书新版云文档类型，与本地 Word .docx 文件的解析路径不同。