当前位置：首页 > news >正文

权限日志没做好，大模型项目上线就崩？数据工程师的转型实战复盘

news 2026/7/28 23:57:13

这篇我按“先跑起来、再讲取舍”的方式写《别急着换赛道：大数据经验在 AI 项目里到底值多少？》。概念会讲，但重点放在代码怎么组织、哪里容易踩坑。

摘要

本文以一个大模型应用从 Demo 到生产环境的落地经历为线索，探讨大数据背景下的数据工程师如何在权限、日志和可观测性上补齐短板。通过真实案例与代码片段，给出可落地的技术选型建议和经验总结。

以前做数仓、写 ETL、搞 Spark 任务时，我们最关心的指标是吞吐量、延迟、容错率；现在做 LLM 应用，核心指标变成了“谁有权读/写”、“调用链路能否追踪”、“异常返回是否可控”。这两套体系的交汇点，往往不在模型参数调优，而在工程层面的权限控制与日志审计。

我在某企业内部问答系统项目中，一开始只把精力放在 Prompt 优化和检索精度上，结果上线后出现两个严重问题：

越权访问：普通员工能查询到高管薪酬数据，因为 RAG 检索未做行级过滤；
不可追溯：当某个回答被投诉错误时，无法定位是哪个 LLM 版本、哪个 Prompt、哪条检索记录导致的。

这些坑，恰恰是传统数据工程师可以发挥优势的地方——数据治理思维 + 工程化落地能力。

---

2. 数据治理：权限与日志的先行设计

在做任何 LLM 应用之前，先问自己三个问题：
1. 用户身份如何认证？（OAuth2 / JWT / 内部账号体系）
2. 数据访问是否受角色控制？（RBAC / ABAC）
3. 每次推理是否有完整日志记录？（输入、输出、模型版本、耗时、来源 IP）

我建议采用以下架构思路：

[客户端] ↓ (携带 JWT Token) [API Gateway] → 验证 Token + 提取用户 ID / Role ↓ [业务服务] → 检查权限策略（如：只能查本部门文档） ↓ [RAG Pipeline] → 注入 user_id, role_id 到元数据中 ↓ [LLM Service] → 记录 request_id, prompt, response, latency ↓ [日志系统] → ELK / Loki + 告警规则（如：敏感词触发）

关键点：不要把权限判断放在 LLM 侧做语义过滤，那太脆弱了。必须在应用层提前做硬拦截，并在日志中打上trace-id关联全链路。

---

3. 向量数据库选型与接入

我们试过 Pinecone、Milvus、Chroma，最终选择 Pinecone Serverless，原因有三：

支持 metadata filtering（配合权限字段使用）
自动扩缩容，适合中小团队
API 简洁，快速集成

接入示例（Python + Pydantic-V2 风格）：

from pinecone import Pinecone import uuid pc = Pinecone(api_key="YOUR_API_KEY") index_name = "company-docs" if index_name not in pc.list_indexes(): pc.create_index( name=index_name, dimension=1536, # 对应 embed 维度 metric="cosine", serverless=True, cloud="aws", region="us-west-1" ) index = pc.Index(index_name) def upsert_with_permission(doc_id: str, vector: list, meta: dict): """插入带权限标记的向量""" index.upsert( vectors=[{ "id": str(uuid.uuid4()), "values": vector, "metadata": { **meta, "_doc_id": doc_id, # 用于下游去重或溯源 "allowed_roles": meta.get("roles", ["public"]), # 关键！ "dept": meta.get("department", "") } }] )

注意：allowed_roles字段将在后续检索时被用作 filter 条件，实现细粒度访问控制。

---

4. RAG 数据管道：从 Demo 到生产

Demo 阶段常犯的错误是“全量索引+无过滤”，生产环境必须引入动态权限裁剪。例如：

def search_retrieval(query: str, user_role: str, user_dept: str): query_embedding = embed(query) # 使用统一 encoder # 构建筛选条件：仅允许当前部门或公开文档 filter_dict = { "$and": [ {"allowed_roles": {"$in": ["public", user_role]}}, {"dept": {"$eq": user_dept}} if user_dept != "" else {} ] } results = index.query( top_k=5, vector=query_embedding, include_metadata=True, filter=filter_dict ) return [m["metadata"] for m in results['matches']]

这个逻辑看似简单，但决定了系统不会“泄露不该让看的信息”。同时，每条检索请求都应记录request_id、user_id、query_hash、filtered_count等字段到日志系统，方便事后审计。

---

5. 落地项目示例：一个基于 RAG 的内部问答系统

我们做了一个简单的 Flask 服务作为入口：

from flask import Flask, request, jsonify import logging app = Flask(__name__) logging.basicConfig(level=logging.INFO) @app.route("/ask", methods=["POST"]) def ask(): data = request.json token = data.get("token") query = data.get("query") # 假设有一个 auth service 解析 token 得到 user_info user_info = validate_token(token) # 返回 {id, role, dept} # 检索相关文档 docs = search_retrieval(query, user_info["role"], user_info["dept"]) if not docs: return jsonify({"error": "无匹配文档"}), 404 # 构造 prompt（简化版） context = "\n\n".join([f"[{d['title']}] {d['content']}" for d in docs]) prompt = f"以下是参考资料：\n{context}\n\n请根据以上资料回答问题：{query}" # 调用 LLM（此处省略实际调用） response = call_llm(prompt) # 记录审计日志 logging.info({ "event": "llm_query", "user_id": user_info["id"], "query_hash": hash(query), "context_len": len(context), "response_len": len(response), "model_version": "gpt-4o-mini-2024-07-18" }) return jsonify({"answer": response})

这个项目虽然不复杂，但它体现了几个关键点：