Kimi K3 与向量库结合的检索增强问答实现

文章导读
要让 Kimi K3 基于私有文档回答问题,直接调用模型只会得到训练参数里的记忆,无法覆盖你手上的资料。可行的处理方向是:先把文档切块并向量化存入向量库,用户提问后先检索相关片段,再把片段和问题一起拼入提示词交给 Kimi K3,让模型基于片段作答。这样的流程可以在不微调模型的情况下,把可核查的上下文注入回答,同时保留替换向量库和调整检索参数的灵活性。下文给出完整可运行的 Python 实现骨架,
📋 目录
  1. 准备文档集并切块生成向量索引
  2. 初始化Kimi K3推理服务并设定请求格式
  3. 编写检索函数返回Top-K相关片段
  4. 将检索片段拼入提示词并调用Kimi K3
  5. 验证答案是否引用片段并调试检索失败
A A

要让 Kimi K3 基于私有文档回答问题,直接调用模型只会得到训练参数里的记忆,无法覆盖你手上的资料。可行的处理方向是:先把文档切块并向量化存入向量库,用户提问后先检索相关片段,再把片段和问题一起拼入提示词交给 Kimi K3,让模型基于片段作答。这样的流程可以在不微调模型的情况下,把可核查的上下文注入回答,同时保留替换向量库和调整检索参数的灵活性。下文给出完整可运行的 Python 实现骨架,环境按 Python 3.10+、Chroma 0.4.x、OpenAI 兼容接口处理。

对于“Kimi K3 + 向量库做私有文档问答”的需求,可行方案是:将文档切块后调用 embedding 模型生成向量存入 Chroma,查询时用 similarity_search 取 Top-K 片段,再拼入“资料:”和“问题:”提示词调用 Kimi K3。适用场景是单机或内网环境下的文档问答原型,操作动作为切块、建索引、检索、拼提示词、验证引用。验证方式为观察答案是否包含片段关键短语,若回答偏离,需要调整切块大小或 Top-K 值。边界在于:该方法只保证“可检索、可引用”,不保证语义推理准确率,也不涉及向量库性能调优。

准备文档集并切块生成向量索引

先准备一份纯文本文档,例如 knowledge.txt,把它按固定长度切块,每块之间保留少量重叠,避免切断语义。切块后调用 embedding 模型生成向量,写入 Chroma 集合。下面示例使用 sentence-transformers 的本地模型做 embedding,替换成任何 OpenAI 兼容的 embedding 接口都可以。

import chromadb
from chromadb.utils import embedding_functions

def split_text(text, chunk_size=500, overlap=50):
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        if end >= len(text):
            break
        start = end - overlap
    return chunks

with open("knowledge.txt", "r", encoding="utf-8") as f:
    content = f.read()

chunks = split_text(content)

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
    name="docs",
    embedding_function=embedding_functions.SentenceTransformerEmbeddingFunction(model_name="paraphrase-multilingual-MiniLM-L12-v2")
)

ids = [f"chunk_{i}" for i in range(len(chunks))]
collection.add(ids=ids, documents=chunks)
print(f"索引完成,共 {len(chunks)} 个片段")

切块大小和重叠长度可先按 500/50 试跑。若文档结构明显(章节标题),建议先按标题分割,再对长章节二次切块,这样检索命中率通常比纯固定长度高。执行前确认 embedding 模型能正常下载,或者在内网环境中提前缓存。

初始化Kimi K3推理服务并设定请求格式

Kimi K3 通常以 OpenAI 兼容接口提供服务。需要先启动推理服务,拿到 base_url 和 api_key,再准备一个基础的对话请求函数,确认模型能正常返回。这里给出一个通用接入骨架,具体端口和模型名以你部署服务的实际输出为准。

Kimi K3 与向量库结合的检索增强问答实现
from openai import OpenAI

client = OpenAI(
    base_url="http://127.0.0.1:8000/v1",
    api_key="EMPTY"
)

def ask_kimi(messages, temperature=0.2):
    resp = client.chat.completions.create(
        model="kimi-k3",
        messages=messages,
        temperature=temperature
    )
    return resp.choices[0].message.content

# 基础验证
test_messages = [
    {"role": "user", "content": "你好,请回复:连接正常"}
]
print(ask_kimi(test_messages))

运行后如果返回“连接正常”或相似内容,说明推理服务可用。若报错,先检查 base_url 是否包含 /v1 路径、模型名是否匹配服务端配置。有些服务还需要额外参数如 top_p、max_tokens,建议先保持最小请求验证,再逐步加参数。

编写检索函数返回Top-K相关片段

检索函数的作用是把用户问题的向量与库中片段向量做相似度计算,返回最相关的 K 个片段。Chroma 的 query 方法内部会处理向量化,可以直接传入查询文本。下面实现一个 similarity_search 函数,返回片段列表和相似度分数。

def similarity_search(query, k=3):
    result = collection.query(
        query_texts=[query],
        n_results=k,
        include=["documents", "distances"]
    )
    documents = result["documents"][0]
    distances = result["distances"][0]
    return documents, distances

question = "公司年假制度是什么?"
docs, scores = similarity_search(question, k=3)
for i, (doc, score) in enumerate(zip(docs, scores)):
    print(f"Top-{i+1} 距离: {score:.4f}")
    print(doc[:100])

Chroma 返回的是距离,不是相似度分数,距离越小越相关。如果后续需要展示分数,可以把距离转为相似度(例如 exp(-distance)),但要结合实际向量库的度量方式确认。若检索结果明显不相关,可以先检查切块是否过短或过长,再调整 Top-K 值。

Kimi K3 与向量库结合的检索增强问答实现

将检索片段拼入提示词并调用Kimi K3

检索到片段后,把它们放入系统或用户消息的“资料”区域,再把用户问题放在“问题”区域,让 Kimi K3 只依据给定资料回答。提示词模板需要明确边界,并设置较低的温度参数,减少随机发挥。

def build_prompt(question, docs):
    context = "\n\n".join(docs)
    return f"""请仅根据以下资料回答问题。如果资料中没有相关信息,就回答“资料中未提及”。不要使用资料之外的常识或记忆。

资料:
{context}

问题:
{question}
"""

def rag_answer(question):
    docs, scores = similarity_search(question, k=3)
    prompt = build_prompt(question, docs)
    messages = [
        {"role": "system", "content": "你是严谨的文档问答助手。"},
        {"role": "user", "content": prompt}
    ]
    answer = ask_kimi(messages, temperature=0.1)
    return answer, docs

question = "公司年假制度是什么?"
answer, docs = rag_answer(question)
print(answer)

temperature 设置为 0.1 或 0.2,能减少模型自创内容的概率。如果模型仍无视资料,可在系统提示词中追加“只依据资料,禁止推测”。注意 max_tokens 的默认值可能限制答案长度,长问答需要按需调整。

验证答案是否引用片段并调试检索失败

验证步骤不能只看回答是否流畅,要确认答案是来自检索片段。下面写一个简单检查:把答案与每个片段做关键短语匹配,若完全没有命中,说明检索或生成环节有问题。

def check_cited(answer, docs, min_phrase_len=8):
    matched = []
    for doc in docs:
        # 抽取片段中较长的连续文本,检查是否出现在答案里
        for i in range(0, len(doc) - min_phrase_len, min_phrase_len):
            phrase = doc[i:i+min_phrase_len]
            if phrase in answer and phrase not in matched:
                matched.append(phrase)
    return matched

answer, docs = rag_answer(question)
matched_phrases = check_cited(answer, docs)
print("匹配到的片段短语数:", len(matched_phrases))
if not matched_phrases:
    print("警告:答案没有直接引用片段中的内容")

如果答案没有引用片段,先排查检索结果:打印检索出的 Top-K 文档,确认是否与问题相关。若文档本身不相关内容,尝试缩小切块大小(例如 500 改成 300)或增大 Top-K 值;若文档相关但答案不引用,应加大提示词约束强度或降低 temperature。调整后重新执行索引构建、检索和验证三步,直到答案里能看到片段中的关键短语。这一步是 RAG 流程里最值得反复调试的部分,不要一上来就换模型或换向量库。