使用 Azure OpenAI、LangChain 与 Function Calling 构建安全的 AI 数据库助手

Source: Dev.to

从原始 CSV 到可投入生产的 AI 助手，安全查询数据——避免产生幻觉式 SQL。

在本文中，我将逐步讲解如何使用以下技术构建一个 AI 驱动的数据分析师：

• Azure OpenAI
• LangChain
• LangGraph
• Function Calling
• SQLite

该助手能够：

• 使用 pandas 分析 CSV 数据
• 安全查询 SQL 数据库
• 自动选择预定义的后端函数
• 清晰解释结果
• 防止幻觉和不安全的 SQL

🧩 问题陈述

当使用 AI 模型和数据库时，常见的问题包括：

• ❌ 虚构的 SQL 查询
• ❌ 不安全的 eval 或原始 SQL 执行
• ❌ 无法控制模型可以访问的内容
• ❌ 未提供结果计算方式的解释

Goal: Build an AI assistant that:

• 回答关于 COVID 数据的分析性问题
• 仅使用允许的工具
• 永不猜测
• 对每个答案进行解释

📊 数据集

我们使用 COVID all‑states history 数据集，其中包括：

• state
• date
• hospitalizedIncrease
• positiveIncrease
• …以及更多

该数据集首先被用作：

• 一个 pandas DataFrame
• 一个 SQLite 数据库

🧱 架构概览

User Question
     ↓
Azure OpenAI (Assistant / LangChain)
     ↓
Tool Selection (Function / SQL / DataFrame)
     ↓
Safe Backend Execution
     ↓
Result
     ↓
Final Explanation

关键理念：

• 模型决定 做什么。
• 你的后端决定 如何 实现。

Source:

🔹 第 1 部分：通过 LangChain 与 Azure OpenAI 对话

使用 AzureChatOpenAI 连接到 Azure OpenAI：

llm = AzureChatOpenAI(
    azure_endpoint="https://<your-resource-name>.cognitiveservices.azure.com/",
    api_key="YOUR_API_KEY",
    api_version="2024-12-01-preview",
    model="gpt-4o-mini"
)

基本检查

response = llm.invoke([
    HumanMessage(content="Hello, Azure OpenAI via LangChain!")
])
print(response.content)

🔹 Part 2: DataFrame Agent (CSV Analysis)

将 CSV 加载到 pandas 中，并通过工具公开 受控计算。

DataFrame Tool

@tool
def run_df(query: str) -> str:
    """Run Python code on the global dataframe `df` and return the result."""
    return str(eval(query))

⚠️ 注意： 在生产环境中，请将 eval 替换为受限执行层。

强制使用工具

llm_with_tools = llm.bind_tools([run_df])

该提示 强制模型：

• 使用工具
• 执行实际的 pandas 计算
• 解释结果

🔹 第三部分：从 CSV → SQL（SQLite）

将 CSV 转换为 SQLite：

engine = create_engine("sqlite:///./db/test.db")

df.to_sql(
    name="all_states_history",
    con=engine,
    if_exists="replace",
    index=False
)

现在可以通过 SQL 查询相同的数据集。

🔹 第4部分：使用 LangGraph 的 SQL 代理

Create a ReAct agent using LangGraph:

agent_executor_SQL = create_react_agent(
    model=llm,
    tools=toolkit.get_tools()
)

The system prompt enforces:

• 仅限有效表
• 仅限特定列
• 不允许虚构值
• 仅输出 Markdown​

🔹 第5部分：函数调用（不使用原始 SQL）

与其让模型直接生成 SQL，不如定义 预先批准的后端函数。

示例函数

def get_hospitalized_increase_for_state_on_date(state_abbr, specific_date):
    ...

def get_positive_cases_for_state_on_date(state_abbr, specific_date):
    ...

函数注册表（关键！）

FUNCTION_MAP = {
    "get_hospitalized_increase_for_state_on_date": get_hospitalized_increase_for_state_on_date,
    "get_positive_cases_for_state_on_date": get_positive_cases_for_state_on_date,
}

这可以确保：

• ✅ 仅运行允许的函数
• ❌ 防止任意代码执行

🔹 第6部分：Azure OpenAI 函数调用（无 Assistant API）

使用 Chat Completions + functions：

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=messages,
    functions=functions,
    function_call="auto"
)

如果模型调用函数：

1. 提取参数
2. 通过 FUNCTION_MAP 路由
3. 执行后端逻辑
4. 将结果返回
5. 获取最终的有依据的答案

🔹 第7部分：Assistant API（持久上下文）

创建 Assistant

assistant = client.beta.assistants.create(
    name="Covid Data Assistant",
    model="gpt-4o-mini",
    tools=[{"type": "function", "function": fn} for fn in functions]
)

Assistant 循环（关键概念）

while True:
    run_status = client.beta.threads.runs.retrieve(...)

    if run_status.status == "requires_action":
        # extract function name
        # dispatch via FUNCTION_MAP
        # submit tool output
    elif run_status.status == "completed":
        break

Assistant 记住对话上下文，同时仍然强制严格的工具使用和安全执行。

🧠 关键要点

• ✅ 此设计解决了什么
  • 防止 SQL 幻觉
  • 强化后端安全
  • 让 AI 回答基于数据
  • 随工具增长平稳扩展

🧩 心智模型

🎯 何时使用何种方式？

🚀 最后思考

这种方法反映了 真实生产 AI 系统 的构建方式：

• AI 决定 什么
• 后端控制 如何

关键好处：

• 数据保持权威性
• 解释保持透明

与我联系

让我们一起学习并构建酷炫的数据科学和 AI 项目！

• LinkedIn:
• GitHub:

📩 欢迎在下方留言！