基于Graphiti的NL2SQL实现最新权威排行榜与实战测评及新手完整教程

我们团队近期落地了一个基于 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)
    # 从列信息中提取实体节点
    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'add_episode 执行完成，耗时 {(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"""你是一个精通表与列关系的助手。
根据下面的用户信息和对话历史进行回答。回复要简短、精准，始终提供有帮助的内容。

你需要了解的关于用户的信息（以便生成有用回复）：
- 需要查询的表

如果信息不足，请主动向用户询问。

用户及其对话相关事实：
{facts_string or '暂无用户及对话相关事实'}
""")
    messages = [system_message] + state['messages']
    # ...

后续规划

项目上线运行一段时间后，NL2SQL 的效果达到预期。团队内部已在广泛使用这个 demo 版。下一步计划很清晰：将 NL2SQL 封装成通用 API，再结合 SQL 执行引擎，打造一个基于对话的数据分析 Agent，让业务同学直接用自然语言提问即可获取结果。