1. 本地下载模型

ollama pull gemma4:12b
ollama pull bge-m3

2 创建文本，作为知识库

• knowledge.txt

==================================================
飞流游戏平台（Feiliu Game Platform）SDK 接入规范 V3.6
发布时间：2026年5月
==================================================

【核心概述】
飞流游戏平台是面向全球开发者的跨系统游戏分发平台。第三方游戏必须通过接入本 SDK 来实现用户鉴权、支付和数据上报。

【环境要求】
- 目标运行环境：Android 12+ / iOS 16+
- 支持开发框架：React Native (New Architecture / JSI 驱动)
- 本地编译工具：建议使用 Pixi 进行 Python 自动化环境管理

【初始化配置参数】
第三方游戏在启动时，必须在主入口初始化飞流引擎，核心初始化参数如下：
- 引擎初始化端点 (App Endpoint)：https://api.feiliu-games.com/v3/init
- 沙箱环境 AppID：fl_sandbox_99824
- 生产环境 AppID：fl_prod_88311
- 默认高并发心跳检测间隔 (Heartbeat Interval)：15秒

【核心安全鉴权机制 - 签名生成算法】
所有从游戏客户端发往飞流服务器的请求，必须在 HTTP Header 中携带 `X-Feiliu-Sign`。
签名生成步骤如下：
1. 将请求体(Body)的 JSON 字符串进行标准的 MD5 加密，生成 32 位小写字符串。
2. 在该字符串末尾拼接平台分配的私钥（盐值），2026年测试私钥固定为：`FeiliuSalt2026_Secure!`。
3. 将拼接后的字符串再次进行 SHA-256 加密，最终得到的 64 位字符即为 `X-Feiliu-Sign`。

【常见错误码与排查指南】
- 错误码 4001：AppID 不匹配或未激活。请检查 `fl_sandbox_99824` 拼写。
- 错误码 4005：签名校验失败。通常是因为未在末尾拼接私钥盐值，或者 MD5 未转为小写。
- 错误码 5002：心跳超时。请确认心跳间隔是否严格设置为了 15 秒。

3. 将文本写入向量数据库

import os
# 【安全防护】：强行绕过可能干扰本地环回 127.0.0.1 通信的代理设置
os.environ["no_proxy"] = "localhost,127.0.0.1"
os.environ["NO_PROXY"] = "localhost,127.0.0.1"

import chromadb
from langchain_text_splitters import RecursiveCharacterTextSplitter
import json
import urllib.request
import sys
import shutil

# ==========================================
# 📊 基础常数配置（须与问答端保持完全一致）
# ==========================================
OLLAMA_EMBED_URL = "http://localhost:11434/api/embed"
EMBED_MODEL = "bge-m3"  # 1024维阿里中文向量专家

DB_PATH = "./feiliu_vector_db_prod" 
COLLECTION_NAME = "feiliu_docs_prod"

def get_embedding(text):
    """请求本地 Ollama 将文本块编码为高维向量"""
    data = json.dumps({"model": EMBED_MODEL, "input": text}).encode("utf-8")
    req = urllib.request.Request(OLLAMA_EMBED_URL, data=data, headers={'Content-Type': 'application/json'})
    try:
        # 针对本地大模型可能存在的冷启动情况，放宽超时时间至 30 秒
        with urllib.request.urlopen(req, timeout=30) as response:
            res = json.loads(response.read().decode("utf-8"))
            return res["embeddings"][0]
    except Exception as e:
        print(f"\n❌ 向量模型通信失败，请检查 Ollama 是否开启或模型是否下载。错误: {e}")
        sys.exit(1)

def main():
    print("🚀 [数据治理服务启动] 开始全量知识库构建程序...")
    
    # 1. 严格审查本地文本源是否存在
    if not os.path.exists("knowledge.txt"):
        print("❌ [阻断异常] 未能在当前目录下找到 'knowledge.txt' 文本源！")
        return

    # 2. 读取源知识数据
    with open("knowledge.txt", "r", encoding="utf-8") as f:
        text = f.read()

    # 3. 执行物理强擦逻辑：无条件抹除本地旧的向量库文件夹，防止 SQLite 锁死或维度冲突
    if os.path.exists(DB_PATH):
        print(f"🧹 监测到已有向量库残留，正在强制物理擦除旧路径: {DB_PATH} ...")
        try:
            shutil.rmtree(DB_PATH)
            print("   -> 旧数据库清理完毕，当前处于无污染纯净状态。")
        except Exception as e:
            print(f"   -> ⚠️ 清理坏库失败（可能被系统进程占用锁死）: {e}")

    # 4. 智能文本结构化切片
    print(f"📖 正在将长文本按【300字窗口 + 50字语义重叠】切碎...")
    splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
    chunks = splitter.split_text(text)
    print(f"   -> 文本已切分完成，共计产出 {len(chunks)} 个独立特征豆腐块。")

    # 5. 建立全新的持久化 ChromaDB 数据库实体
    chroma_client = chromadb.PersistentClient(path=DB_PATH)
    collection = chroma_client.get_or_create_collection(name=COLLECTION_NAME)

    # 6. 全量转化为向量坐标并写入底层数据库
    print(f"🧠 正在调用模型 [{EMBED_MODEL}] 逐块进行分布式/本地向量特征提取...")
    for i, chunk in enumerate(chunks):
        print(f"   [数据注入] 正在编码并塞入第 {i+1}/{len(chunks)} 块...")
        vector = get_embedding(chunk)
        collection.add(ids=[f"doc_{i}"], embeddings=[vector], documents=[chunk])
    
    # 7. 打印收工状态
    print(f"\n🎉 [全量同步成功] 恭喜！当前库内共计安全保存了 {collection.count()} 条 1024 维特征索引数据。")
    print("💾 数据库已进入硬盘持久化状态，现在可以安全运行问答端代码了。")

if __name__ == "__main__":
    main()

3.1 运行

pixi run python ingest.py

3.2 输出

shen@deMacBook-Air feiliu_rag_system % pixi run python ingest.py
🚀 [数据治理服务启动] 开始全量知识库构建程序...
📖 正在将长文本按【300字窗口 + 50字语义重叠】切碎...
   -> 文本已切分完成，共计产出 5 个独立特征豆腐块。
🧠 正在调用模型 [bge-m3] 逐块进行分布式/本地向量特征提取...
   [数据注入] 正在编码并塞入第 1/5 块...
   [数据注入] 正在编码并塞入第 2/5 块...
   [数据注入] 正在编码并塞入第 3/5 块...
   [数据注入] 正在编码并塞入第 4/5 块...
   [数据注入] 正在编码并塞入第 5/5 块...

🎉 [全量同步成功] 恭喜！当前库内共计安全保存了 5 条 1024 维特征索引数据。
💾 数据库已进入硬盘持久化状态，现在可以安全运行问答端代码了。

4.让大模型根据知识库的数据开始回答问题

import os
# 【安全防护】：强行绕过可能干扰本地环回 127.0.0.1 通信的代理设置
os.environ["no_proxy"] = "localhost,127.0.0.1"
os.environ["NO_PROXY"] = "localhost,127.0.0.1"

import chromadb
import json
import urllib.request
import sys

# ==========================================
# 📊 基础常数配置（须与入库端保持完全一致）
# ==========================================
OLLAMA_EMBED_URL = "http://localhost:11434/api/embed"
OLLAMA_GEN_URL = "http://localhost:11434/api/generate"

EMBED_MODEL = "bge-m3"        # 向量模型（需用它来将用户的【提问】转化为相同维度的坐标）
CHAT_MODEL = "gemma4:12b"     # 推理对话大脑

DB_PATH = "./feiliu_vector_db_prod" 
COLLECTION_NAME = "feiliu_docs_prod"
DISTANCE_THRESHOLD = 0.7     # 超过此值， 证明小抄与问题风马牛不相及

def get_embedding(text):
    """将提问文本编码为向量，以便去数据库检索"""
    data = json.dumps({"model": EMBED_MODEL, "input": text}).encode("utf-8")
    req = urllib.request.Request(OLLAMA_EMBED_URL, data=data, headers={'Content-Type': 'application/json'})
    try:
        with urllib.request.urlopen(req, timeout=30) as response:
            res = json.loads(response.read().decode("utf-8"))
            return res["embeddings"][0]
    except Exception as e:
        print(f"\n❌ 连接 Ollama 向量化失败: {e}")
        sys.exit(1)

def query_knowledge(collection, query):
    """核心双轨分流检索引擎"""
    query_vector = get_embedding(query)
    
    # 去只读的本地表内打捞相似度最高的前 4 个知识点
    results = collection.query(query_embeddings=[query_vector], n_results=4)
    
    valid_docs = []
    if results and results['documents'] and results['distances']:
        docs = results['documents'][0]
        distances = results['distances'][0]
        
        # 语义紧密关联度硬卡阈值判断
        for doc, dist in zip(docs, distances):
            if dist <= DISTANCE_THRESHOLD and doc.strip():
                valid_docs.append(doc)

    print("\n" + "="*60)
    
    # 🌟 双轨制分流
    if valid_docs:
        context = "\n---\n".join(valid_docs)
        print("📚 [状态：命中私有数据] 正在拉取本地小抄实施 RAG 强约束问答...")
        print(context)
        print("="*60)
        
        prompt = f"""你是一个经过严格安全审计的私有知识库助手。请完全基于以下给出的【参考资料】来回答用户的提问。

【行为准则】：
1. 答案必须严格遵循【参考资料】中的事实，禁止掺杂任何你已知的外部知识。
2. 如果【参考资料】中找不到明确答案，请直接回答“抱歉，在本地参考资料中未检索到相关内容”。

【参考资料】：
{context}

【用户的提问】：
{query}

【你的严谨回答】：
"""
    else:
        print("💡 [状态：未命中库] 相似度过低，已切换为大模型原生通用大脑兜底解答！")
        print("="*60)
        
        prompt = f"""你是一个无所不知、视野开阔的通用 AI 助手。
由于本地知识库里完全没有包含相关的技术方案，请调用你原本所学的庞大知识库，礼貌详细地为用户解惑。

【用户的提问】：
{query}

【你的回答】：
"""

    print(f"🤖 {CHAT_MODEL} 正在深度推理并流式输出：\n")
    data = json.dumps({"model": CHAT_MODEL, "prompt": prompt, "stream": True}).encode("utf-8")
    req = urllib.request.Request(OLLAMA_GEN_URL, data=data, headers={'Content-Type': 'application/json'})
    
    try:
        with urllib.request.urlopen(req) as response:
            # 改进：引入局部变量缓存切片，防止极端情况下 urllib 流式产生断包粘包
            buffer = ""
            while True:
                chunk = response.read(1024)  # 每次读取固定的安全字节块
                if not chunk:
                    break
                buffer += chunk.decode("utf-8")
                
                # 按照换行符拆分成熟的 JSON 行
                while "\n" in buffer:
                    line, buffer = buffer.split("\n", 1)
                    if line.strip():
                        res_json = json.loads(line)
                        sys.stdout.write(res_json["response"])
                        sys.stdout.flush()
        print("\n" + "_"*40)
    except Exception as e:
        print(f"\n❌ 与本地生成模型内核交互中断: {e}")

def main():
    # 🌟 核心改进：服务启动时只做“唯读检查”，不承担任何初始化写盘逻辑
    if not os.path.exists(DB_PATH):
        print("❌ [服务启动失败] 没发现已初始化的本地向量库文件夹！")
        print("💡 解决方案：请在终端首先运行 `pixi run python ingest.py` 生成数据索引。")
        return

    try:
        chroma_client = chromadb.PersistentClient(path=DB_PATH)
        collection = chroma_client.get_collection(name=COLLECTION_NAME)
        print(f"💡 [热启动成功] 服务已挂载本地向量库，检测到共有 {collection.count()} 条可用索引数据。")
    except Exception as e:
        print(f"❌ [读取表异常] 向量数据库文件损坏或缺失表结构 ({e})。请重新执行 ingest.py 构建流程。")
        return

    # 命令行单次提问拓展支持
    if len(sys.argv) > 1:
        query_knowledge(collection, sys.argv[1])
        return

    # 无边界交互式 REPL 循环终端
    print("\n✨ 欢迎接入飞流智能生产级知识库系统！输入 'exit' 或 'quit' 安全退出。")
    while True:
        try:
            user_input = input("\n👤 研发人员提问: ").strip()
            if not user_input: continue
            if user_input.lower() in ['exit', 'quit']:
                print("👋 服务已安全解挂。再见！")
                break
            query_knowledge(collection, user_input)
        except KeyboardInterrupt:
            print("\n👋 安全断开。")
            break

if __name__ == "__main__":
    main()

4.1 运行

pixi run python app.py

4.2 结果

💡 [热启动成功] 服务已挂载本地向量库，检测到共有 5 条可用索引数据。

✨ 欢迎接入飞流智能生产级知识库系统！输入 'exit' 或 'quit' 安全退出。

👤 研发人员提问: 签名生成算法

============================================================
📚 [状态：命中私有数据] 正在拉取本地小抄实施 RAG 强约束问答...
【核心安全鉴权机制 - 签名生成算法】
所有从游戏客户端发往飞流服务器的请求，必须在 HTTP Header 中携带 `X-Feiliu-Sign`。
签名生成步骤如下：
1. 将请求体(Body)的 JSON 字符串进行标准的 MD5 加密，生成 32 位小写字符串。
2. 在该字符串末尾拼接平台分配的私钥（盐值），2026年测试私钥固定为：`FeiliuSalt2026_Secure!`。
3. 将拼接后的字符串再次进行 SHA-256 加密，最终得到的 64 位字符即为 `X-Feiliu-Sign`。
============================================================
🤖 gemma4:12b 正在深度推理并流式输出：

所有从游戏客户端发往飞流服务器的请求，必须在 HTTP Header 中携带 `X-Feiliu-Sign`。其生成算法步骤如下：

1. 将请求体(Body)的 JSON 字符串进行标准的 MD5 加密，生成 32 位小写字符串。
2. 在该字符串末尾拼接平台分配的私钥（盐值），其中 2026 年测试私钥固定为：`FeiliuSalt2026_Secure!`。
3. 将拼接后的字符串再次进行 SHA-256 加密，最终得到的 64 位字符即为 `X-Feiliu-Sign`。
________________________________________

👤 研发人员提问: 签名生成算法

============================================================
📚 [状态：命中私有数据] 正在拉取本地小抄实施 RAG 强约束问答...
【核心安全鉴权机制 - 签名生成算法】
所有从游戏客户端发往飞流服务器的请求，必须在 HTTP Header 中携带 `X-Feiliu-Sign`。
签名生成步骤如下：
1. 将请求体(Body)的 JSON 字符串进行标准的 MD5 加密，生成 32 位小写字符串。
2. 在该字符串末尾拼接平台分配的私钥（盐值），2026年测试私钥固定为：`FeiliuSalt2026_Secure!`。
3. 将拼接后的字符串再次进行 SHA-256 加密，最终得到的 64 位字符即为 `X-Feiliu-Sign`。
============================================================
🤖 gemma4:12b 正在深度推理并流式输出：

所有从游戏客户端发往飞流服务器的请求，必须在 HTTP Header 中携带 `X-Feiliu-Sign`。

签名生成步骤如下：
1. 将请求体(Body)的 JSON 字符串进行标准的 MD5 加密，生成 32 位小写字符串。
2. 在该字符串末尾拼接平台分配的私钥（盐值），其中2026年测试私钥固定为：`FeiliuSalt2026_Secure!`。
3. 将拼接后的字符串再次进行 SHA-256 加密，最终得到的 64 位字符即为 `X-Feiliu-Sign`。
________________________________________

👤 研发人员提问: 今天天气怎么样

============================================================
💡 [状态：未命中库] 相似度过低，已切换为大模型原生通用大脑兜底解答！
============================================================
🤖 gemma4:12b 正在深度推理并流式输出：

您好！作为一个致力于为您提供全面服务的 AI 助手，我很希望能为您提供最准确的信息。

由于系统出于对您的**隐私保护**的考虑，我无法自动获取您当前的精确地理位置。因此，为了能准确回答您“今天天气怎么样”，**请您告知我您目前所在的城市或具体的区域名称。**

一旦您提供了地点信息，我不仅可以为您查询到以下实时的天气数据：
1.  **基础气象：** 实时温度、体感温度、湿度以及风力等级。
2.  **降水预警：** 今日是否有降雨、降雪或雷暴等极端天气提醒。
3.  **紫外线与空气质量：** 今天的 UV 指数高低以及 PM2.5 等空气质量指数，方便您规划户外活动。
4.  **穿衣建议：** 根据当下的气温变化，为您提供合理的着装建议（例如：是需要加衣防寒还是轻便透气）。

此外，如果您对天气背后的**科学原理**感兴趣（例如：为什么这一带经常出现局部阵雨？某种气候现象的成因是什么？），我也非常乐意从气象学的专业角度为您进行深度剖析。

**请告诉我您所在的城市，我将立即为您细致地查询并解答！**
________________________________________

👤 研发人员提问: