使用 langchain_deepseek 实现自然语言转数据库查询SQL

Github

• https://github.com/langchain-ai/langchain

官网

• https://python.langchain.com/docs/introduction/

简介

LangChain 是一个用于构建 LLM 驱动的应用程序的框架。它可以帮助您将可互操作的组件和第三方集成链接在一起，以简化 AI 应用程序开发 - 同时随着底层技术的发展做出面向未来的决策。

LangChain 通过模型、嵌入、向量存储等的标准接口帮助开发人员构建由 LLM 支持的应用程序。

LangChain 的用途：

• 实时数据增强。利用 LangChain 庞大的集成库（包含模型提供商、工具、向量存储、检索器等），轻松将 LLM 连接到各种数据源和外部 / 内部系统。
• 模型互操作性。您的工程团队可以进行实验，以找到最适合您应用程序需求的选择，从而交换模型。随着行业前沿的发展，快速适应 — LangChain 的抽象概念让您不断前进，而不会失去动力。

LangChain 的生态：

• LangSmith - 有助于代理评估和可观察性。调试性能不佳的 LLM 应用程序运行，评估代理轨迹，获得生产中的可见性，并随着时间的推移提高性能。
• LangGraph - 使用我们的低级代理编排框架 LangGraph 构建能够可靠地处理复杂任务的代理。 LangGraph 提供可定制的架构、长期记忆和人机交互工作流程，并受到 LinkedIn、Uber、Klarna 和 GitLab 等公司在生产中的信任。
• LangGraph 平台- 使用专门为长期运行、有状态的工作流构建的部署平台轻松部署和扩展代理。在团队之间发现、重用、配置和共享代理 — 并使用 LangGraph Studio中的可视化原型快速迭代。

腾讯云DeepSeek API

• https://console.cloud.tencent.com/lkeap/api

DEEPSEEK_API_KEY = "sk-xxxx"
DEEPSEEK_API_BASE = "https://api.lkeap.cloud.tencent.com/v1"

注：API_KEY 申请后免费增送1000000token。

DeepSeek API

• https://platform.deepseek.com/usage

DEEPSEEK_API_KEY = "sk-xxxx"
DEEPSEEK_API_BASE = "https://api.deepseek.com"

注：API_KEY 申请，申请后可以充个10块钱。没充值API接口用不了。

ChatDeepSeek

• https://python.langchain.com/api_reference/deepseek/chat_models/langchain_deepseek.chat_models.ChatDeepSeek.html

安装相关库

pip install langchain
pip install langchain-core
pip install langchain_deepseek
pip install --upgrade --quiet langchain-community langchainhub langgraph
pip install  python-dotenv

创建 .env 文件

DEEPSEEK_API_KEY = "sk-xxxx"
DEEPSEEK_API_BASE = "https://api.deepseek.com"

验证 API 接口

import os
from dotenv import load_dotenv, dotenv_values
from langchain_deepseek import ChatDeepSeek

load_dotenv()

env_vars = dotenv_values()
print("[Custom .env variables]")
for key, value in env_vars.items():
    print(f"{key}={value}")

def Demo1():
    client = ChatDeepSeek(
        model="deepseek-chat"
    )
    prompt_messages = [
        {"role": "system", "content": "You are a helpful assistant, and you only speak Chinese."},
        {"role": "user", "content": "你好"}
    ]
    # 调用模型
    response = client.invoke(prompt_messages)
    print(response)
    content = response.content
    print(content)

if __name__ == "__main__":
    Demo1()

• 输出

content='你好！很高兴见到你，有什么我可以帮忙的吗？' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 11, 'prompt_tokens': 16, 'total_tokens': 27, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 16}, 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_3a5770e1b4_prod0225', 'id': '10b16ac4-4342-42fe-8109-7e0c04f59ccf', 'finish_reason': 'stop', 'logprobs': None} id='run-e0b68259-63a6-4aee-a6e6-cf634b675861-0' usage_metadata={'input_tokens': 16, 'output_tokens': 11, 'total_tokens': 27, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
你好！很高兴见到你，有什么我可以帮忙的吗？

生成数据库查询SQL

获取测试用数据库

curl -s https://raw.githubusercontent.com/lerocha/chinook-database/master/ChinookDatabase/DataSources/Chinook_Sqlite.sql | sqlite3 Chinook.db

验证数据库查询

from langchain_community.utilities import SQLDatabase

db = SQLDatabase.from_uri("sqlite:///Chinook.db")
print(db.dialect)
print(db.get_usable_table_names())
db.run("SELECT * FROM Artist LIMIT 10;")

生成数据库查询SQL

import os
from dotenv import load_dotenv, dotenv_values
from typing import TypedDict, Annotated
from langchain import hub
from langchain_deepseek import ChatDeepSeek
from langchain_community.utilities import SQLDatabase

load_dotenv()

env_vars = dotenv_values()
print("[Custom .env variables]")
for key, value in env_vars.items():
    print(f"{key}={value}")

class QueryOutput(TypedDict):
    """Generated SQL query."""
    query: Annotated[str, ..., "Syntactically valid SQL query."]

def Demo1():
    client = ChatDeepSeek(
        model="deepseek-chat"
    )
    # 加载数据库
    db = SQLDatabase.from_uri("sqlite:///Chinook.db")
    # 加载 SQL 查询提示词模板
    query_prompt_template = hub.pull("langchain-ai/sql-query-system-prompt")
    prompt = query_prompt_template.invoke(
        {
            "dialect": db.dialect,
            "top_k": 10,
            "table_info": db.get_table_info(),
            "input": "列出年龄大于30岁的员工的名字和姓氏。"
        }
    )
    print(prompt)
    llm = client.with_structured_output(QueryOutput)
    response = llm.invoke(prompt)

    print(response)
    query = response['query']
    print(query)
    
    results = db.run(query)
    print(results)
    
if __name__ == "__main__":
    Demo1()

• 输出

{'query': "SELECT FirstName, LastName, Title FROM Employee WHERE (strftime('%Y', 'now') - strftime('%Y', BirthDate)) > 30 LIMIT 10;"}

SELECT FirstName, LastName, Title FROM Employee WHERE (strftime('%Y', 'now') - strftime('%Y', BirthDate)) > 30 LIMIT 10;

[('Andrew', 'Adams', 'General Manager'), ('Nancy', 'Edwards', 'Sales Manager'), ('Jane', 'Peacock', 'Sales Support Agent'), ('Margaret', 'Park', 'Sales Support Agent'), ('Steve', 'Johnson', 'Sales Support Agent'), ('Michael', 'Mitchell', 'IT Manager'), ('Robert', 'King', 'IT Staff'), ('Laura', 'Callahan', 'IT Staff')]