Graphiti实现NL2SQL实战指南

来源：互联网 2026-07-03 06:29:12

人工智能资讯合集

基于graphiti构建数据库表结构知识图谱，融合RAG与LangGraph搭建NL2SQL智能体，通过动态提示词模板优化，将自然语言精准转为SQL查询，实现高效自动化数据分析，显著提升查询准确率与开发效率。

近期，团队成功落地了一个基于 graphiti 的 NL2SQL 项目，实际测试效果令人满意。本文将思路与实现过程整理成文档，便于后续迭代，也为有类似需求的团队提供参考。

前言

数据分析的常规流程通常包括：业务提需求、评估数据来源、数据开发、可视化、交付。这套链路周期长、重复性高。生成式 AI 的出现，尤其在自然语言理解上的突破，为自动化数据分析提供了更实际的落地方案。NL2SQL 正是这一方向的典型应用——将自然语言直接转换为 SQL 查询，缩短从需求到结果的距离。下面是我们实现的具体路径。

长期稳定更新的攒劲资源： >>>点此立即查看<<<

一、给静态大模型增加公司内部数据知识

通用大模型虽然强大，但对企业内部的数据结构并不了解。为了让模型理解业务表及字段含义，需要为其补充“内部知识”。常见方法有两种：

模型参数微调：使用内部数据构造问答对进行微调，效果不错，但副作用明显——可能影响模型在其他通用场景下的能力，甚至导致某些已有能力退化。
In-Context Learning（ICL）：通过在提示词中嵌入相关上下文（如表结构、示例 SQL），让模型现场学习。该方法不更新模型参数，对原有能力无影响。RAG（检索增强生成）是 ICL 的典型实现——先检索外部知识库，再将检索到的上下文作为输入提供给模型。

我们在公司项目中选择了 RAG 方案，将内部数据构建成知识库，作为模型生成 SQL 时的“参考资料”。

二、基于 graphiti 构建数据库表结构知识库

知识库是 RAG 系统的基础。一个可靠的知识库需要具备稳定性、可靠性和易扩展性。经过调研，构建方式主要有两类：

基于搜索引擎：通过关键字或向量相似度进行模糊搜索，再配合聚类算法实现关联检索。适合海量数据且对准确度要求不高的场景（如推荐系统）。
基于知识图谱：将实体及其关系建模为图结构，精准度较高，适合需要精确推理的场景。

以往构建知识图谱依赖人工，耗时耗力。如今借助大模型，可以自动从文本中抽取实体和关系，显著降低了门槛。graphiti 正是这样的开源工具——基于大模型自动构建知识图谱。我们在项目中采用了混合策略：先用检索找到中心节点，再通过 graphiti 构建的知识图谱进行精细化知识检索。以下是基于 graphiti 构建数据库表结构知识图谱的核心代码片段：

from graphiti_core import Graphiti
from graphiti_core.utils.maintenance.node_operations import (extract_attributes_from_nodes, extract_nodes)
from graphiti_core.utils.maintenance.edge_operations import (build_episodic_edges, resolve_extracted_edges)
from graphiti_core.utils.bulk_utils import (add_nodes_and_edges_bulk)
from graphiti_core.nodes import EpisodeType, EpisodicNode, EntityNode
from graphiti_core.edges import EntityEdge

async def add_table_column_nodes (client: Graphiti, center_node_uuid: str, schema_name: str, table_name: str, table_comment: str, columns: list):
    start = datetime.now(timezone.utc)
    now = datetime.now(timezone.utc)
    previous_episodes = []
    episode: EpisodicNode = await EpisodicNode.get_by_uuid(client.driver, center_node_uuid)
    # Extract entities as nodes
    extracted_nodes = []
    for column in columns:
        extracted_nodes.append(EntityNode(
            name = schema_name + "." + table_name + "." + column["column_name"],
            group_id = episode.group_id,
            labels = ["Column"],
            created_at = now,
            attributes = {
                "schema_name": schema_name,
                "table_name": table_name,
                "column_name": column["column_name"],
                "column_comment": column["column_comment"],
                "data_type": column["data_type"]
            }
        ))
    table_node = EntityNode(
        name = schema_name + "." + table_name,
        group_id = episode.group_id,
        labels = ["Table"],
        created_at = now,
        attributes = {
            "schema_name": schema_name,
            "table_name": table_name,
            "table_comment": table_comment
        }
    )
    nodes = [table_node] + extracted_nodes
    episodic_edges = build_episodic_edges([table_node], episode, now)
    edges = []
    for extracted_node in extracted_nodes:
        edges.append(EntityEdge(
            name = "has_column",
            fact = table_node.name + " has column " + extracted_node.name,
            source_node_uuid = table_node.uuid,
            target_node_uuid = extracted_node.uuid,
            group_id = extracted_node.group_id,
            created_at = extracted_node.created_at
        ))
    (resolved_edges, invalidated_edges) = await resolve_extracted_edges(
        client.clients, edges, episode, nodes, {}, ({('Entity','Entity'): []})
    )
    entity_edges = resolved_edges
    hydrated_nodes = await extract_attributes_from_nodes(client.clients, nodes, episode, previous_episodes, None)
    await add_nodes_and_edges_bulk(client.driver, [episode], episodic_edges, hydrated_nodes, entity_edges, client.embedder)
    end = datetime.now(timezone.utc)
    logger.info(f'Completed add_episode in {(end - start) * 1000} ms')

三、搭建 NL2SQL Agent

知识图谱构建完成后，下一步是将大模型与知识图谱串联，搭建一个 NL2SQL 智能体。核心思路是使用 LangGraph 构建一个带工具调用的 agent，工具即上一步定义的 get_table_columns。agent 在需要查询表结构时自动调用该工具，并将结果注入上下文。实现如下：

from logger import logger
from rag_graphiti_client import client, llm
from langgraph.checkpoint.memory import MemorySa ver
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode
from rag_graphiti_mcp_tools import get_table_columns

tools = [get_table_columns]
tool_node = ToolNode(tools)
llm_with_tools = llm.bind_tools(tools)

from rag_graphiti_chatbot import Chatbot, State
chatbot = Chatbot(llm=llm_with_tools, graphiti=client)

graph_builder = StateGraph(State)
memory = MemorySa ver()

async def should_continue(state, config):
    messages = state['messages']
    last_message = messages[-1]
    if not last_message.tool_calls:
        return 'end'
    else:
        return 'continue'

graph_builder.add_node('agent', chatbot.generate_chatbot_response)
graph_builder.add_node('tools', tool_node)
graph_builder.add_edge(START, 'agent')
graph_builder.add_conditional_edges('agent', should_continue, {'continue': 'tools', 'end': END})
graph_builder.add_edge('tools', 'agent')

graph = graph_builder.compile(checkpointer=memory)

from rag_graphiti_user import query_user, create_user
user_name = "tcq"
user_node_uuid = await query_user(user_name)
if user_node_uuid is None:
    user_node_uuid = await create_user(user_name)
user_state = State(user_name=user_name, user_node_uuid=user_node_uuid)

from rag_graphiti_agent import Agent, AgentRun
agent = Agent(graph)
agentRun: AgentRun = agent.run(user_state=user_state)

四、优化——基于模板的动态化提示词

NL2SQL 的准确率直接影响用户体验。我们尝试了多种优化手段，最终效果最显著的是“基于模板的动态化提示词”方案。简要说明：

用模板固定 prompt 骨架：减少每次生成的 prompt 结构大幅波动，模型理解更稳定，输出一致性显著提升。
动态注入上下文：根据用户输入，将对话历史、知识图谱检索出的表关系动态拼入模板，使模型理解当前问题的“背景故事”。

这样既保持了模板的稳定性，又提供了逻辑的灵活性。实现时，每次 agent 处理消息前，先通过 graphiti 的 search 方法根据用户输入检索相关图谱事实，然后构造包含这些事实的 system message，再拼接历史对话：

async def generate_chatbot_response(self, state: State):
    facts_string = None
    if len(state['messages']) > 0:
        last_message = state['messages'][-1]
        graphiti_query = f'{ "TableColumnQueryBot" if isinstance(last_message, AIMessage) else state["user_name"] }: {last_message.content}'
        edge_results = await asyncio.create_task(
            self.graphiti.search(
                graphiti_query,
                center_node_uuid=state['user_node_uuid'],
                num_results=5
            )
        )
        facts_string = edges_to_facts_string(edge_results)
    system_message = SystemMessage(content=f""""You are a skillfull table column relationship manager.
Review information about the user and their prior conversation below and respond accordingly.
Keep responses short and concise. And remember, always be helpful!

Things you'll need to know about the user in order to generate a helpful response:
- need query table

Ensure that you ask the user for the above if you don't already know.

Facts about the user and their conversation:
{facts_string or 'No facts about the user and their conversation'}
""")
    messages = [system_message] + state['messages']
    # ...