LangGraph SQL代理-MCP POC
SQL代理的自然语言 Snowflake的模型上下文协议（MCP）集成。
______________________________________________________________________
🎯 这有什么作用
AI驱动的SQL代理：
- 将自然语言问题转换为SQL查询
- 在雪花上执行安全验证
- 记住对话背景，以便后续跟进
- 通过MCP公开用于VS代码集成的工具
数据来源： 雪花TPC-H样本数据集(SNOWFLAKE_SAMPLE_DATA.TPCH_SF1)
______________________________________________________________________
📅 最近的改进（2026年2月11日）
代码质量和架构
- ✅ 模块化代理包 -有组织
src/agent/明确区分关注点：
- core.py -主代理编排 - nodes.py -6个工作流节点实现 - prompts.py -集中式LLM提示（易于更新） - graph_builder.py -LangGraph拓扑与编译
- ✅ 修复了线程问题 -SQLite内存数据库，具有用于多线程执行的持久连接
- ✅ 会话管理 -默认情况下为新会话，可选持久性
功能和可靠性
- ✅ 历史感知查询 -检测摘要/参考问题并验证历史可用性
- ✅ 智能错误处理 -在没有对话记录的情况下防止产生幻觉
- ✅ 对话记忆 -内存（默认）和基于文件的持久性模式
______________________________________________________________________
✅ 已实现的功能
会话管理
- ✅ 默认情况下为新会话 -重新启动时清除内存中的SQLite数据库
- ✅ 可选持久性 -设置
PERSIST_MEMORY=true在重启过程中保留历史记录 - ✅ 线程安全内存 -具有持久连接的多线程LangGraph执行
核心代理（6节点工作流）
- 范围检测 -过滤掉非数据问题
- SQL生成 -自然语言→ 具有架构上下文的SQL
- 安全确认 -阻止DROP/DELETE/ALTER/UPDATE操作
- 查询执行 -雪花集成重试逻辑（最多3个）
- 结果格式 -大型数据集的智能截断
- 响应生成 -用户友好的自然语言答案
高性能
- ✅ 对话记忆 -基于SQLite的历史跟踪
- ✅ 后续问题 -过去5次互动的背景
- ✅ 架构自动发现 -自动查询信息_SCHEMA
- ✅ 摘要支持 -对话历史记录的答案
- ✅ MCP集成 -将工具暴露给VS代码
______________________________________________________________________
📁 项目结构
LangGraph/
├── main.py # Interactive CLI interface
├── README.md # This file
├── .gitignore # Git ignore rules (root level)
│
├── setup/ # Configuration & dependencies
│ ├── .env # Credentials (gitignored)
│ ├── requirements.txt # Python dependencies
│ └── SETUP.md # Setup guide
│
├── data/ # Runtime data & artifacts
│ └── conversation_history.db # Conversation history (when PERSIST_MEMORY=true)
│
├── src/ # Core agent modules
│ ├── agent/ # Agent package (modularized)
│ │ ├── __init__.py # Package exports
│ │ ├── core.py # Main SQLAgent class
│ │ ├── nodes.py # 6-node workflow implementations
│ │ ├── prompts.py # All LLM prompts (centralized)
│ │ └── graph_builder.py # LangGraph construction & topology
│ ├── config.py # Environment-based configuration
│ ├── memory.py # SQLite conversation storage
│ ├── tools.py # Snowflake integration + auto schema discovery
│ └── validator.py # SQL safety validator
│
├── mcp_impl/ # MCP HTTP Server & Flask Web UI
│ ├── app.py # Flask web interface (port 8001)
│ ├── server_http.py # HTTP MCP server (port 8000)
│ ├── server_manager.py # MCP server lifecycle management
│ └── response_formatter.py # Response formatting for display
│
├── docs/ # Documentation
│ ├── PROJECT_STRUCTURE.md # Detailed codebase organization
│ ├── test_queries.md # Example queries to test the agent
│ └── COMPLETE_WORKFLOW.md # Workflow verification details
│
└── scripts/
└── run.sh # Application launcher______________________________________________________________________
🚀 快速开始
1.设置环境
# Create virtual environment
python3 -m venv .venv
source .venv/bin/activate # macOS/Linux
# Install dependencies
pip install -r setup/requirements.txt2.配置凭据
创建 setup/.env 使用您的凭据文件：
# Snowflake
SNOWFLAKE_ACCOUNT=your_account.region
SNOWFLAKE_USER=your_username
SNOWFLAKE_PASSWORD=your_password
SNOWFLAKE_DATABASE=SNOWFLAKE_SAMPLE_DATA
SNOWFLAKE_SCHEMA=TPCH_SF1
SNOWFLAKE_WAREHOUSE=COMPUTE_WH
SNOWFLAKE_ROLE=ACCOUNTADMIN
# OpenAI
OPENAI_API_KEY=sk-...
# Optional: Persistent conversation history
PERSIST_MEMORY=false # Set to true for file-based history3.运行代理
Web UI和MCP服务器：
.venv/bin/python mcp_impl/app.py应用程序将从以下时间开始：
- Web用户界面: http://localhost:8001（聊天界面）
- MCP服务器: http://localhost:8000（JSON-RPC 2.0）
______________________________________________________________________
💡 示例用法
交互模式
Enter your query: How many customers are there?
Generated SQL: SELECT COUNT(*) FROM CUSTOMER;
✓ Safety validation passed
There are 150,000 customers in the data.后续问题
Enter your query: How many orders?
Generated SQL: SELECT COUNT(*) FROM ORDERS;
✓ Safety validation passed
There are 1,500,000 orders in the database.
Enter your query: Summarize the key numbers we discussed
Using conversation history to answer...
Based on our conversation:
- Total customers: 150,000
- Total orders: 1,500,000安全特性
Enter your query: Delete all old orders
Generated SQL: DELETE FROM ORDERS WHERE...
🛑 Safety Check Failed:
❌ BLOCKED: Query contains dangerous operations: DELETE
⚠️ Only SELECT queries are allowed for safety.______________________________________________________________________
🏗️ 建筑
6节点工作流
┌─────────────┐
│ User Query │
└──────┬──────┘
│
▼
┌─────────────────┐
│ Scope Detection │ ── Filter out-of-scope questions
└────────┬────────┘
│
▼
┌──────────────────┐
│ SQL Generation │ ── NL → SQL with schema context
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Safety Validator │ ── Block dangerous operations
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Execute Query │ ── Snowflake execution (retry x3)
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Format Results │ ── Intelligent truncation
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Generate Response│ ── Natural language answer
└──────────────────┘数据流
- 输入： 自然语言问题
- 范围检查： 这与数据有关吗？
- 架构上下文： 加载表/列元数据
- SQL生成： 带有模式上下文的GPT-4
- 验证： 安全检查（只读执行）
- 执行： 使用重试逻辑查询Snowflake
- 内存： 存储在SQLite中以备后续跟进
- 答复： 自然格式化结果
______________________________________________________________________
📊 测试
快速测试套件
测试您的连接：
.venv/bin/python scripts/quick_test.py尝试示例查询： 看 docs/test_queries.md 查看示例查询列表
______________________________________________________________________
🔧 配置
环境变量(.env)
敏感凭据-永远不要提交到git
会话管理
- 默认模式：内存中的SQLite（重新启动时的新会话）
- 持久模式：设置
PERSIST_MEMORY=true用于基于文件的历史记录
架构发现
自动查询 INFORMATION_SCHEMA 首次使用时
______________________________________________________________________
📚 文档
有关更多详细信息，请参阅 文档 文件夹：
______________________________________________________________________
🎯 主要功能概述
| 功能 | 状态 | 描述 |
|---|---|---|
| 自然语言→ SQL | ✅ | GPT-4动力转换 |
| 安全验证 | ✅ | 块DROP/DELETE/ALTER |
| 架构自动发现 | ✅ | 查询信息_SCHEMA |
| 对话记忆 | ✅ | 基于SQLite的历史记录 |
| 后续问题 | ✅ | 过去5次互动的背景 |
| 重试逻辑 | ✅ | 最多3次错误尝试 |
| 新会议 | ✅ | 默认情况下为内存数据库 |
| 持久历史 | ✅ | 可选的基于文件的存储 |
| 范围检测 | ✅ | 过滤非数据问题 |
______________________________________________________________________
📝 许可证
MIT许可证-开源