AI智能体开发实战：RAG与SQL智能调用

帆1122

444人浏览 · 2026-03-27 21:14:23

帆1122 · 2026-03-27 21:14:23 发布

从零实现一个多工具智能体：RAG + SQL + 计算器（Python + LangGraph）

本文首发于掘金，同步发布于 CSDN。这是我学习 AI Agent 开发第一周的实战记录，实现一个能自动判断问题类型、调用不同工具（RAG、SQL、计算器）的智能体。

一、环境准备

1.1 安装 Python 包

pip install chromadb langgraph requests python-dotenv

1.2 配置 API Key
创建 .env 文件：

DEEPSEEK_API_KEY=你的API密钥

1.3 准备测试数据
运行以下脚本创建 SQLite 数据库：

import sqlite3
conn = sqlite3.connect("sales.db")
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS sales (
    id INTEGER PRIMARY KEY,
    product TEXT,
    amount INTEGER,
    date TEXT
)
''')
sales_data = [
    ("笔记本", 30, "2024-03-01"),
    ("鼠标", 50, "2024-03-02"),
    ("键盘", 20, "2024-03-03"),
    ("笔记本", 40, "2024-03-04"),
    ("鼠标", 60, "2024-03-05"),
]
cursor.executemany("INSERT INTO sales (product, amount, date) VALUES (?, ?, ?)", sales_data)
conn.commit()
conn.close()

二、完整代码
2.1 导入库和初始化

import chromadb
import sqlite3
import requests
import os
import math
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from typing import TypedDict

load_dotenv()

# 初始化 Chroma（RAG 向量数据库）
client = chromadb.PersistentClient(path="./chroma_data")
collection = client.get_or_create_collection("test_collection")

# 添加示例文档到知识库
collection.add(
    documents=["AI Agent 是智能体程序", "Python 是一门编程语言"],
    ids=["doc1", "doc2"]
)

2.2 通用 API 调用函数

def call_deepseek(prompt):
    """调用 DeepSeek API"""
    api_key = os.getenv("DEEPSEEK_API_KEY")
    url = "https://api.deepseek.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    data = {
        "model": "deepseek-chat",
        "messages": [{"role": "user", "content": prompt}]
    }
    response = requests.post(url, headers=headers, json=data)
    return response.json()["choices"][0]["message"]["content"]

2.3 RAG 检索函数

def search_rag(query):
    results = collection.query(query_texts=[query], n_results=1)
    return results['documents'][0][0] if results['documents'][0] else ""

2.4 定义状态（数据包）

class AgentState(TypedDict):
    question: str    # 用户问题
    intent: str      # 意图：knowledge / data / calculator
    answer: str      # 最终答案

2.5 意图判断节点

def classify_node(state: AgentState):
    prompt = f"""判断问题类型，只回答"knowledge"、"data"或"calculator"。
    知识类：问概念、定义、解释
    数据类：问统计、数量、金额
    计算类：数学计算
    问题：{state["question"]}
    类型："""
    result = call_deepseek(prompt).strip().lower()
    if "knowledge" in result:
        intent = "knowledge"
    elif "data" in result:
        intent = "data"
    else:
        intent = "calculator"
    return {"intent": intent}

2.6 各工具节点
知识节点（RAG）：

def knowledge_node(state: AgentState):
    context = search_rag(state["question"])
    prompt = f"基于资料回答问题。资料：{context}\n问题：{state['question']}"
    return {"answer": call_deepseek(prompt)}

SQL 节点：

def get_schema():
    conn = sqlite3.connect("sales.db")
    cursor = conn.cursor()
    cursor.execute("SELECT sql FROM sqlite_master WHERE type='table'")
    schema = cursor.fetchall()
    conn.close()
    return "\n".join([s[0] for s in schema if s[0]])

def execute_sql(sql):
    conn = sqlite3.connect("sales.db")
    cursor = conn.cursor()
    try:
        cursor.execute(sql)
        result = cursor.fetchall()
        conn.close()
        return str(result)
    except Exception as e:
        conn.close()
        return str(e)

def sql_node(state: AgentState):
    schema = get_schema()
    prompt = f"""数据库结构：{schema}
字段：product(产品), amount(数量), date(日期)
问题：{state['question']}
SQL："""
    sql = call_deepseek(prompt).strip()
    # 清理 Markdown 标记
    sql = sql.replace("```sql", "").replace("```", "").strip()
    result = execute_sql(sql)
    explain_prompt = f"查询结果：{result}\n用自然语言回答"
    return {"answer": call_deepseek(explain_prompt)}

计算器节点：

def calculator_node(state: AgentState):
    prompt = f"提取表达式：{state['question']}"
    expr = call_deepseek(prompt).strip().lower()
    allowed_names = {"sqrt": math.sqrt, "pi": math.pi, "e": math.e}
    try:
        result = eval(expr, {"__builtins__": {}}, allowed_names)
        return {"answer": f"{expr} = {result}"}
    except:
        return {"answer": "无法计算该表达式"}

2.7 路由与图构建

def route_by_intent(state: AgentState):
    if state["intent"] == "knowledge":
        return "knowledge"
    elif state["intent"] == "data":
        return "sql"
    else:
        return "calculator"

builder = StateGraph(AgentState)
builder.add_node("classify", classify_node)
builder.add_node("knowledge", knowledge_node)
builder.add_node("sql", sql_node)
builder.add_node("calculator", calculator_node)
builder.set_entry_point("classify")
builder.add_conditional_edges("classify", route_by_intent, {
    "knowledge": "knowledge",
    "sql": "sql",
    "calculator": "calculator"
})
builder.add_edge("knowledge", END)
builder.add_edge("sql", END)
builder.add_edge("calculator", END)
graph = builder.compile()

2.8 运行主程序

if __name__ == "__main__":
    while True:
        question = input("\n请输入问题（q 退出）：")
        if question.lower() == 'q':
            break
        result = graph.invoke({"question": question, "intent": "", "answer": ""})
        print(f"答案：{result['answer']}")