示例代码

Kinger2026/7/23大约 4 分钟

import os

# 1. 禁用 HuggingFace 联网检查（解决连不上国外服务器导致的卡顿）
os.environ["TRANSFORMERS_OFFLINE"] = "1"
os.environ["HF_HUB_OFFLINE"] = "1"

# 2. 消除 Windows 下烦人的符号链接警告
os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"

# 3. 告诉 Python 以后只去本地缓存找模型
os.environ["SENTENCE_TRANSFORMERS_HOME"] = "./models/model_cache"
from langchain_openai import ChatOpenAI
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from config.settings import settings

# 1. 加载本地 Embedding (这是基础包，肯定有)
embeddings = HuggingFaceEmbeddings(
    model_name="shibing624/text2vec-base-chinese",
    cache_folder="./models/model_cache"
)

# 2. 加载数据库
vectorstore = Chroma(persist_directory="./vector_db", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 3. 初始化 Qwen
llm = ChatOpenAI(
    model=settings.MODEL_NAME,
    api_key=settings.API_KEY,
    base_url=settings.BASE_URL,
    temperature=settings.TEMPERATURE,
)

# 4. 定义 Prompt
template = """你是一个 AIOps 专家。请根据以下上下文回答问题。
上下文: {context}
问题: {question}
回答:"""
prompt = ChatPromptTemplate.from_template(template)

# 5. 【核心】LCEL 表达式 (这是现代写法，不依赖 langchain.chains)
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 这一行就是你的“大脑回路”：找资料 -> 格式化 -> 喂给 Prompt -> 给 LLM -> 转成文字
rag_chain = (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
)

def chat():
    print("🤖 AIOps 助手已上线！(流式输出模式)")
    while True:
        user_input = input("\n运维问题 >> ")
        if user_input == "exit": break

        print("\n💡 建议方案：", end="", flush=True)

        # 将 .invoke() 替换为 .stream()
        # 这会让数据像水流一样，出一块打一块
        for chunk in rag_chain.stream(user_input):
            print(chunk, end="", flush=True)
        print("\n")  # 换行

if __name__ == "__main__":
    chat()

1. 环境“防卡顿”配置

代码开头的 os.environ 是为了解决你在 Windows 下开发常遇到的痛点：

离线模式：TRANSFORMERS_OFFLINE 告诉程序：“模型已经在本地了，别再去连 HuggingFace 的服务器检查更新了。”这能显著提升启动速度，避免因为网络抖动卡在加载界面。
路径锁定：SENTENCE_TRANSFORMERS_HOME 明确告诉 Python 你的 Embedding 模型藏在 ./models/model_cache 里。

2. 知识库加载（检索层）

这里是 AI 的“长期记忆”来源：

**HuggingFaceEmbeddings**：这是你的“翻译官”。它负责把用户输入的文字翻译成 AI 能理解的向量数字。
**Chroma**：这是你的“图书馆”。persist_directory="./vector_db" 指定了数据库文件就在本地。
**retriever**：这是你的“图书管理员”。search_kwargs={"k": 3} 表示它每次收到提问，都会去库里抓取最相关的 3 段内容 出来。

3. LCEL 管道表达式（逻辑神经回路）

这是你代码中最精彩的部分，采用了 LCEL (LangChain Expression Language)。它通过管道符 | 把数据流串起来：

**{"context": retriever | format_docs, "question": RunnablePassthrough()}**：
context：调用管理员找资料，并用 format_docs 把多段资料拼成一个大长段。
question：原封不动地传下用户的提问。
**| prompt**：把找来的资料和问题塞进你定义的 template（模版）里，变成一段完整的指令。
**| llm**：把指令发给 Qwen 大脑进行推理。
**| StrOutputParser()**：把 AI 返回的复杂数据包，精简提取成我们能看懂的纯文本字符串。

4. 流式输出交互（表现层）

在 chat() 函数中，你使用了 .stream() 模式：

for chunk in rag_chain.stream(user_input):
    print(chunk, end="", flush=True)

**chunk**：表示 AI 生成的一小块内容（通常是一个词或一个字）。
**flush=True**：强迫 Python 立即把字吐到屏幕上，而不是攒够了再输出。
效果：这让 AI 看起来像是在“实时思考并打字”，极大缓解了用户等待云端 API 返回时的焦虑感。

总结：这个 RAG 到底是怎么跑的？

输入：你问“K8s 怎么重启 Pod？”
检索：ChromaDB 翻出《K8s 运维手册》第 5 页关于重启的说明。
组合：Prompt 变成：“你是一个专家，请根据《手册第 5 页内容》，回答‘K8s 怎么重启 Pod？’”。
生成：Qwen 收到指令，开始一个字一个字地吐出操作步骤。

你的下一个优化方向

你的代码现在已经很稳健了。如果想让它更像一个**“运维助手”**，可以考虑：

增加溯源：在输出答案后，让 AI 告诉你是从哪个 .txt 文件里找到的（利用 doc.metadata）。
异常捕获：加上 try...except，防止 Qwen API 偶尔波动导致整个程序崩溃退出。

你想试试让 AI 在回答完问题后，自动显示出它是参考了哪份文件吗？ 我可以帮你微调一下 rag_chain 的结构。