昨天手搓的50行RAG检索引擎,今天接上大模型做成问答机器人:Python实战第2期
昨天那篇,50 行纯 NumPy 手搓了个 RAG 检索引擎。你问"RAG和微调有什么区别",它能挑出最相关的文档,吐出相似度分数。但光有分数还不够,用户要的是人话回答。今天把这个检索引擎接上大模型,做成命令行里能对话的问答机器人。
整个链路三步:检索(昨天代码原样复用)、拼 prompt、调大模型 API。今天的重点是后两步。先看最终效果:你问"RAG和微调有什么区别",机器人先检索出两篇最相关的文档,把它们塞进 prompt,大模型照着文档组织成自然语言回答你。检索管"找得准",大模型管"说得顺"。
环境很简单。Python 3.8 以上,一个 numpy 就够。发 HTTP 请求用标准库 urllib,不装 requests。大模型 API 用 DeepSeek,OpenAI 兼容接口,价格便宜,今天这点实验花不了几毛钱。key 别写进代码,放环境变量 DEEPSEEK_API_KEY 里。
第一步,检索。跟昨天一模一样,只是包成函数,文档向量预计算一次:
import math
import numpy as np
from collections import Counter
def char_bigrams(s):
s = s.replace(" ", "")
return [s[i:i+2] for i in range(len(s)-1)] or [s]
docs = [
"RAG是检索增强生成,先从知识库检索相关文档,再交给大模型生成回答",
"向量数据库用来存储文本的向量表示,支持快速相似度检索",
"微调是更新模型参数,RAG不需要训练模型,成本更低",
"提示词工程通过设计输入文本来引导大模型输出",
"Transformer的注意力机制让模型同时关注输入序列的不同位置",
]
vocab = sorted({t for d in docs for t in char_bigrams(d)})
idx = {t: i for i, t in enumerate(vocab)}
N = len(docs)
df = Counter()
for d in docs:
for t in set(char_bigrams(d)):
df[t] += 1
idf = np.array([math.log(N / df[t]) for t in vocab])
def tfidf_vec(text):
cnt = Counter(char_bigrams(text))
total = sum(cnt.values())
v = np.zeros(len(vocab))
for t, c in cnt.items():
if t in idx:
v[idx[t]] = (c / total) * idf[idx[t]]
return v
doc_vecs = np.stack([tfidf_vec(d) for d in docs])
def retrieve(question, top_k=2):
q = tfidf_vec(question)
denom = np.linalg.norm(doc_vecs, axis=1) * np.linalg.norm(q)
scores = np.where(denom > 0, (doc_vecs @ q) / denom, 0.0)
top = np.argsort(scores)[::-1][:top_k]
return [(docs[i], float(scores[i])) for i in top]
第二步,拼 prompt。这是 RAG 的命门。系统提示里必须写清楚:只根据参考文档回答,文档里没有就直说不知道,不许编造。这句不写,检索白做,大模型照样一本正经胡说八道。
SYSTEM = "你是一个知识库问答助手。只根据下面提供的参考文档回答问题,文档里没有的信息就直说不知道,不要编造。"
def build_prompt(question):
hits = retrieve(question)
context = "\n".join(f"文档{i+1}:{d}" for i, (d, s) in enumerate(hits))
return f"{SYSTEM}\n\n参考文档:\n{context}\n\n用户问题:{question}\n回答:"
第三步,调大模型 API。用 urllib 发 POST,body 是 OpenAI 兼容的格式。key 从环境变量读,读不到直接报错,告诉你去 export,不让你对着 401 瞎猜。
import json
import os
import urllib.request
import urllib.error
API_URL = "https://api.deepseek.com/chat/completions"
def call_llm(prompt):
api_key = os.environ.get("DEEPSEEK_API_KEY")
if not api_key:
raise RuntimeError("没读到 DEEPSEEK_API_KEY,先 export DEEPSEEK_API_KEY=你的key")
body = json.dumps({
"model": "deepseek-chat",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
}).encode("utf-8")
req = urllib.request.Request(
API_URL, data=body,
headers={"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
data = json.loads(resp.read().decode("utf-8"))
return data["choices"][0]["message"]["content"]
except urllib.error.HTTPError as e:
raise RuntimeError(f"API 返回 {e.code}:{e.read().decode('utf-8')[:200]}")
temperature 设 0.1 是有讲究的。问答场景要的是稳定输出,不是发散创作。温度越低,模型越老实照着文档说。
最后,命令行问答循环。三段拼起来:
def main():
print("RAG 问答机器人启动,输入 q 退出")
while True:
question = input("\n你:").strip()
if question.lower() == "q":
break
if not question:
continue
print("机器人:", call_llm(build_prompt(question)))
if __name__ == "__main__":
main()
验证结果。离线部分我跑了一遍:问"RAG和微调有什么区别",检索返回的两篇跟昨天一致,拼出来的 prompt 长这样:
你是一个知识库问答助手。只根据下面提供的参考文档回答问题,文档里没有的信息就直说不知道,不要编造。
参考文档:
文档1:微调是更新模型参数,RAG不需要训练模型,成本更低
文档2:RAG是检索增强生成,先从知识库检索相关文档,再交给大模型生成回答
用户问题:RAG和微调有什么区别
回答:
API 那步需要你自己的 key。我用本地 mock 服务验了整条链路:正常返回能解析出回答文本,key 错误返回 401 会把服务端信息打出来。上面贴的代码和输出都是真实跑出来的,没有编造。
说三个最常见的坑。第一个,key 硬编码进代码,顺手 push 到公开仓库。公开仓库里因为硬编码泄露 key 的事一点不少见,刷出来的账单够心疼。环境变量是底线。
第二个,prompt 里不加"不知道就说不知道"。大模型的出厂设置是永远给你一个答案,文档里没有它就现编。RAG 防幻觉,一半靠检索,一半靠这句 prompt。
第三个,top_k 设太大。十篇八篇文档全塞进 prompt,context 容易爆,token 钱也哗哗的。我的经验:2 到 3 篇最划算,再多收益递减,还稀释注意力。
收藏这篇,下次你要做知识库问答,检索、prompt、API 调用三段代码直接抄过去改,省至少两小时。
下期是 Python 实战第 3 期:给机器人加"记忆",做多轮对话。历史消息怎么管理才不爆 token,上下文窗口怎么取舍。关注我,不错过。
你接大模型 API 的时候,踩过最大的坑是什么?评论区聊聊。
署名:Cyrus
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐


所有评论(0)