不搞复杂向量库，不折腾深度学习框架，50行代码跑通第一个AI应用

引言

在AI应用开发的浪潮中，很多人被Transformer、Attention、向量数据库等技术名词吓退，迟迟不敢动手。其实对于应用层开发者来说，完全可以把大模型当作一种新型“数据库”来用——你不需要理解B+树索引怎么实现，也能写出高效的SQL查询。

今天，我们就用不到50行Python代码，搭建一个完整的RAG（检索增强生成）系统。整个实现只依赖openai库，核心逻辑清晰到你可以用“查表+填空”来理解。

一、整体设计：RAG就是“查表+填空”

我们先把整个代码跑起来，再拆解每个部分。最终效果如下：

# 完整代码见文末
questions = [
    "我想退货，运费谁出？",
    "白银会员需要消费多少？",
    "你们什么时候发货？"
]
for q in questions:
    print(ask_ai(q))

输出示例：

问题: 我想退货，运费谁出？
回答: 我们支持7天无理由退货，运费由买家承担。如果商品质量问题，运费由我们承担。

问题: 白银会员需要消费多少？
回答: 普通会员消费满1000元升级为白银会员。

问题: 你们什么时候发货？
回答: 现货商品24小时内发货，预售商品以页面标注时间为准。

二、代码拆解：四个步骤构建RAG

1. 知识库 —— 你的“数据表”

knowledge_base = {
    "退货政策": "我们支持7天无理由退货，运费由买家承担。如果商品质量问题，运费由我们承担。",
    "发货时间": "现货商品24小时内发货，预售商品以页面标注时间为准。",
    "会员等级": "普通会员消费满1000元升级为白银会员，满5000元升级为黄金会员。"
}

这里我们把领域知识直接写成字典，相当于一个只有三行记录的“表”。在实际项目中，这些数据可能来自文档、数据库或爬虫。

2. 检索函数 —— 你的“SQL查询”

def retrieve(query):
    keywords = {
        "退货": "退货政策",
        "发货": "发货时间",
        "会员": "会员等级"
    }
    for key, topic in keywords.items():
        if key in query:
            return knowledge_base[topic]
    return knowledge_base["退货政策"]   # 默认返回

这是超简化版的检索逻辑：根据用户问题中的关键词，直接命中对应的知识条目。真实场景中，这里会换成向量检索（如FAISS、ChromaDB），但核心思想完全一样：从知识库中找到最相关的片段。

你可以把这一步想象成 SELECT * FROM knowledge_base WHERE content LIKE '%关键词%'。

3. 提示构建 —— 你的“SQL查询优化”

def build_prompt(query, context):
    return f"""基于以下已知信息回答问题。如果无法从信息中找到答案，就说不知道。

已知信息：
{context}

问题：{query}
回答："""

这里我们把检索到的知识和用户问题“拼”成一个自然语言指令，告诉模型如何使用这些信息。这就是Prompt Engineering的本质——就像给SQL加索引提示一样，让执行引擎更高效。

4. 模型调用 —— 你的“数据库执行引擎”

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key",   # 请替换为你自己的密钥
    base_url="https://api.deepseek.com"
)

def ask_ai(query):
    context = retrieve(query)
    prompt = build_prompt(query, context)
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "You are a helpful assistant"},
            {"role": "user", "content": prompt}
        ],
        stream=False
    )
    return response.choices[0].message.content

调用大模型API就像执行一条SQL语句：发送输入，等待返回结果。这里的 temperature 参数可以调整回答的创造性，类似SQL中的 OPTIMIZE 提示。

三、运行你的第一个RAG

环境准备

pip install openai

获取API密钥

• 访问 DeepSeek平台 注册并获取API Key
• 将代码中的 "your-api-key" 替换为真实密钥（或使用环境变量）

运行脚本

保存代码为 rag_demo.py，执行：

python rag_demo.py

你会看到三条问题的自动回答。恭喜，你已经跑通了一个完整的RAG系统！

四、从玩具到工具：改进方向

这个超简版本虽然能工作，但离生产环境还有距离。你可以沿着以下路径逐步升级：

1. 向量检索替代关键词匹配

# 使用 chromadb 做真实检索
import chromadb
client = chromadb.Client()
collection = client.create_collection("my_kb")
collection.add(documents=[...], embeddings=[...])
results = collection.query(query_texts=[query], n_results=1)

2. 增加流式输出

response = client.chat.completions.create(
    stream=True,
    ...
)
for chunk in response:
    print(chunk.choices[0].delta.content, end="")

3. 添加成本监控与异常重试

def ask_ai_with_retry(query, retries=3):
    for i in range(retries):
        try:
            # 调用逻辑
            print(f"本次消耗 tokens: {response.usage}")
            return answer
        except Exception as e:
            print(f"重试 {i+1}/{retries}: {e}")
            time.sleep(2)
    return "服务暂时不可用"

4. 生产化封装

• 用FastAPI包装成HTTP服务
• 用Java/Go重写高并发路径（可选）
• 引入环境变量管理密钥，避免硬编码

五、总结

我们把AI应用开发比作数据库开发：

• 知识库 = 数据表
• 检索 = SQL查询
• 提示构建 = 查询优化
• 模型调用 = 执行引擎

这种类比能帮助应用开发者快速建立心智模型，避开底层的Transformer原理，专注于数据与业务逻辑。

记住：动手比看课重要100倍。花3小时跑通一个RAG，胜过花3天看理论。 现在，拿起代码，去构建你的第一个AI应用吧！

---

附：完整代码

import os
from openai import OpenAI

knowledge_base = {
    "退货政策": "我们支持7天无理由退货，运费由买家承担。如果商品质量问题，运费由我们承担。",
    "发货时间": "现货商品24小时内发货，预售商品以页面标注时间为准。",
    "会员等级": "普通会员消费满1000元升级为白银会员，满5000元升级为黄金会员。"
}

def retrieve(query):
    keywords = {"退货": "退货政策", "发货": "发货时间", "会员": "会员等级"}
    for key, topic in keywords.items():
        if key in query:
            return knowledge_base[topic]
    return knowledge_base["退货政策"]

def build_prompt(query, context):
    return f"""基于以下已知信息回答问题。如果无法从信息中找到答案，就说不知道。

已知信息：
{context}

问题：{query}
回答："""

client = OpenAI(
    api_key=os.environ.get("DEEPSEEK_API_KEY"),  # 建议使用环境变量
    base_url="https://api.deepseek.com"
)

def ask_ai(query):
    context = retrieve(query)
    prompt = build_prompt(query, context)
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[
            {"role": "system", "content": "You are a helpful assistant"},
            {"role": "user", "content": prompt}
        ],
        stream=False
    )
    return response.choices[0].message.content

if __name__ == "__main__":
    questions = [
        "我想退货，运费谁出？",
        "白银会员需要消费多少？",
        "你们什么时候发货？"
    ]
    for q in questions:
        print(f"\n问题: {q}")
        print(f"回答: {ask_ai(q)}")