LangGraph + MCP на практике: строим ChatBI-агента, который ищет данные по запросу на естественном языке
Автор собрал ChatBI-агента, который понимает человеческую формулировку и сам возвращает результат из базы: оркестрацию процесса он построил на LangGraph, а работу с базой обернул в MCP. Пайплайн такой: определение намерения, получение схемы таблиц, генерация SQL, проверка SQL, выполнение запроса и разбор результата. MCP-сервер при этом открывает только три инструмента — list_tables, get_table_schema и execute_sql — и проверяет, что запросы только читают данные.
Полный путь реализации ChatBI на LangGraph с обёрткой базы через MCP: двойная проверка SQL и условия выхода при повторных попытках. Можно брать и адаптировать под себя.
LangGraph + MCP на практике: строим ChatBI-агента для запросов к данным на естественном языке
В предыдущих статьях мы делали агента поддержки, мультиагентную коллаборацию и агента для ревью кода. Теперь поднимаемся на корпоративный уровень: оркестрируем процесс через LangGraph + подключаем базу данных через MCP и делаем ChatBI-агента, который понимает человеческую речь и сам возвращает результат из базы. I. Общая архитектура Текст
Редактировать
Пользователь задаёт вопрос → LangGraph распознаёт намерение → получает схему таблиц → генерирует SQL → проверяет SQL → выполняет запрос → интерпретирует результат → выдаёт ответ Ключевое разделение: LangGraph: отвечает за оркестрацию workflow, определяет «кто, когда и что делает», управляет переходами состояний и откатами при ошибках. MCP: оборачивает операции с базой данных в стандартные инструменты, агент вызывает их по единому протоколу — не нужно писать отдельную адаптацию под каждую базу. II. Шаг первый: устанавливаем зависимости bash
Редактировать
pip install langgraph langchain-openai langchain-mcp-adapters fastmcp pymysql sqlalchemy
三、第二步：构建 MCP Database Server
MCP Server 是 Agent 和数据库之间的"薄代理层"，本身不写业务逻辑，只负责协议转换和安全校验。
python
Редактировать
# mcp_db_server.py
import os
import json
from fastmcp import FastMCP
import pymysql
from contextlib import contextmanager
mcp = FastMCP("DatabaseServer")
DB_CONFIG = {
"host": os.getenv("DB_HOST", "localhost"),
"user": os.getenv("DB_USER", "root"),
"password": os.getenv("DB_PASSWORD", ""),
"database": os.getenv("DB_NAME", "sales"),
"charset": "utf8mb4",
}
@contextmanager
def get_db():
conn = pymysql.connect(**DB_CONFIG)
try:
yield conn
finally:
conn.close()
@mcp.tool()
def list_tables() -> str:
"""列出数据库中所有可用的表名"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute("SHOW TABLES")
tables = [row[0] for row in cursor.fetchall()]
return json.dumps(tables)
@mcp.tool()
def get_table_schema(table_name: str) -> str:
"""获取指定表的字段结构（列名、类型、注释）"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute(f"DESCRIBE `{table_name}`")
columns = cursor.fetchall()
schema = []
for col in columns:
schema.append({
"field": col[0],
"type": col[1],
"nullable": col[2],
"key": col[3],
"comment": col[5] if len(col) > 5 else ""
})
return json.dumps(schema, ensure_ascii=False)
@mcp.tool()
def execute_sql(sql: str) -> str:
"""执行只读 SQL 查询并返回结果。仅允许 SELECT 语句。"""
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT"):
return json.dumps({"error": "只允许 SELECT 查询，拒绝执行"})
if any(kw in sql_upper for kw in ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]):
return json.dumps({"error": "检测到危险操作，拒绝执行"})
try:
with get_db() as conn:
cursor = conn.cursor(pymysql.cursors.DictCursor)
cursor.execute(sql)
rows = cursor.fetchall()
return json.dumps(rows, default=str, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": str(e)})
if __name__ == "__main__":
mcp.run()
启动服务：
bash
编辑
python mcp_db_server.py
四、第三步：定义 State
python
编辑
from typing import TypedDict, Optional, List
from langchain_core.messages import HumanMessage, AIMessage
class ChatBIState(TypedDict):
messages: list[HumanMessage | AIMessage]
user_query: str # 用户原始问题
tables: Optional[str] # 可用表列表
schema: Optional[str] # 相关表结构
generated_sql: Optional[str] # 生成的 SQL
sql_valid: bool # SQL 是否通过校验
query_result: Optional[str] # 查询结果
final_answer: Optional[str] # 最终回答
error: Optional[str] # 错误信息
retry_count: int # 重试次数
五、第四步：连接 MCP 工具
使用 langchain-mcp-adapters 把 MCP Server 暴露的工具转换为 LangChain 标准 Tool 对象。
python
编辑
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
async def get_mcp_tools():
client = MultiServerMCPClient({
"database": {
"command": "python",
"args": ["mcp_db_server.py"],
"transport": "stdio"
}
})
tools = await client.get_tools()
return tools
六、第五步：编写核心 Node
. Schema 发现节点：先搞清楚有哪些表、哪些字段
python
编辑
async def schema_discovery_node(state: ChatBIState) -> dict:
"""获取可用表和关键字段信息，为 SQL 生成提供上下文"""
tools = await get_mcp_tools()
list_tables_tool = next(t for t in tools if t.name == "list_tables")
tables_str = await list_tables_tool.ainvoke({})
tables = json.loads(tables_str)
# 根据用户问题，智能选择可能相关的表（简单版：取前3个表的结构）
# 生产环境可用向量检索做语义匹配
schema_info = []
get_schema_tool = next(t for t in tools if t.name == "get_table_schema")
for table in tables[:3]:
schema_str = await get_schema_tool.ainvoke({"table_name": table})
schema_info.append(f"表名: {table}\n{schema_str}\n")
return {
"tables": tables_str,
"schema": "\n".join(schema_info)
}
. SQL 生成节点：让 LLM 结合表结构生成 SQL
python
编辑
def sql_generation_node(state: ChatBIState) -> dict:
"""基于用户问题和表结构生成 SQL"""
query = state["user_query"]
schema = state["schema"]
prompt = f"""你是一个 SQL 专家。请根据以下数据库结构和用户问题，生成一条可执行的 MySQL 查询语句。
数据库表结构：
{schema}
用户问题：{query}
要求：
. 只使用上述表结构中的表和字段
. 只生成 SELECT 语句
. 结果默认 LIMIT 10，除非用户明确要求更多
. 只输出 SQL，不要任何解释
生成的 SQL："""
response = llm.invoke(prompt)
sql = response.content.strip().strip("```sql").strip("```").strip()
return {"generated_sql": sql}
. SQL 校验节点：双重保险
python
编辑
def sql_validation_node(state: ChatBIState) -> dict:
"""校验 SQL 的安全性（只读）和语法合理性"""
sql = state["generated_sql"]
sql_upper = sql.upper()
# 基础安全检查
dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE", "TRUNCATE"]
for kw in dangerous_keywords:
if kw in sql_upper:
return {
"sql_valid": False,
"error": f"SQL 包含危险操作: {kw}"
}
if not sql_upper.startswith("SELECT"):
return {"sql_valid": False, "error": "只允许 SELECT 查询"}
# 让 LLM 做一次语义校验
schema = state["schema"]
check_prompt = f"""请校验以下 SQL 是否使用了正确的表和字段。
表结构：{schema}
SQL：{sql}
如果正确，只输出：VALID
如果有问题，输出：INVALID + 原因"""
response = llm.invoke(check_prompt)
if "INVALID" in response.content:
return {
"sql_valid": False,
"error": f"语义校验失败: {response.content}"
}
return {"sql_valid": True}
. SQL 执行节点：通过 MCP 工具查数据库
python
编辑
async def sql_execution_node(state: ChatBIState) -> dict:
"""通过 MCP 工具执行 SQL 查询"""
tools = await get_mcp_tools()
execute_tool = next(t for t in tools if t.name == "execute_sql")
result_str = await execute_tool.ainvoke({"sql": state["generated_sql"]})
result = json.loads(result_str)
if isinstance(result, dict) and "error" in result:
return {
"query_result": None,
"error": result["error"]
}
return {
"query_result": json.dumps(result, ensure_ascii=False, indent=2)
}
. 结果解读节点：把数据翻译成自然语言
python
编辑
def answer_generation_node(state: ChatBIState) -> dict:
"""将查询结果转化为自然语言回答"""
query = state["user_query"]
result = state["query_result"]
prompt = f"""你是一个数据分析师。用户问：{query}
查询结果如下：
{result}
请用简洁的中文回答用户，突出关键数据。如果结果为空，如实告知。"""
response = llm.invoke(prompt)
return {"final_answer": response.content}
. SQL 修正节点：校验或执行失败时重新生成
python
编辑
def sql_correction_node(state: ChatBIState) -> dict:
"""根据错误信息重新生成 SQL"""
query = state["user_query"]
schema = state["schema"]
error = state.get("error", "")
old_sql = state.get("generated_sql", "")
prompt = f"""之前的 SQL 执行失败，请修正。
用户问题：{query}
表结构：{schema}
原 SQL：{old_sql}
错误信息：{error}
请生成修正后的 SELECT 语句，只输出 SQL。"""
response = llm.invoke(prompt)
sql = response.content.strip().strip("```sql").strip("```").strip()
return {
"generated_sql": sql,
"retry_count": state.get("retry_count", 0) + 1
}
七、第六步：组装图并编译
python
编辑
from langgraph.graph import StateGraph, START, END
from typing import Literal
def route_after_validation(state: ChatBIState) -> Literal["execute_sql", "correct_sql"]:
if state["sql_valid"]:
return "execute_sql"
return "correct_sql"
def route_after_execution(state: ChatBIState) -> Literal["answer", "correct_sql", "__end__"]:
if state.get("error"):
if state.get("retry_count", 0) < 3:
return "correct_sql"
return "__end__" # 超过重试次数，输出错误
return "answer"
workflow = StateGraph(ChatBIState)
workflow.add_node("discover_schema", schema_discovery_node)
workflow.add_node("generate_sql", sql_generation_node)
workflow.add_node("validate_sql", sql_validation_node)
workflow.add_node("execute_sql", sql_execution_node)
workflow.add_node("correct_sql", sql_correction_node)
workflow.add_node("generate_answer", answer_generation_node)
# 主流程
workflow.add_edge(START, "discover_schema")
workflow.add_edge("discover_schema", "generate_sql")
workflow.add_edge("generate_sql", "validate_sql")
# 校验后分流
workflow.add_conditional_edges("validate_sql", route_after_validation)
# 执行后分流
workflow.add_conditional_edges("execute_sql", route_after_execution)
# 修正后重新校验
workflow.add_edge("correct_sql", "validate_sql")
# 生成答案后结束
workflow.add_edge("generate_answer", END)
app = workflow.compile()
八、第七步：运行测试
python
编辑
import asyncio
async def main():
result = await app.ainvoke({
"messages": [],
"user_query": "统计每个产品类别的销售额，按销售额降序排列",
"retry_count": 0
})
print("用户问题:", result["user_query"])
print("生成 SQL:", result.get("generated_sql"))
print("查询结果:", result.get("query_result"))
print("最终回答:", result.get("final_answer"))
if result.get("error"):
print("错误:", result["error"])
asyncio.run(main())
九、进阶：多 MCP Server 协同
生产环境中，Agent 往往需要同时访问多个系统。LangGraph MCP Adapters 支持同时连接多个 MCP Server：
python
编辑
async def get_multi_tools():
client = MultiServerMCPClient({
"database": {
"command": "python",
"args": ["mcp_db_server.py"],
"transport": "stdio"
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"transport": "stdio"
},
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/data"],
"transport": "stdio"
}
})
return await client.get_tools()
Так агент может одновременно с запросами к базе читать файлы и искать по репозиторию кода, самостоятельно выбирая подходящую цепочку инструментов под вопрос пользователя. X. Рекомендации для продакшена Безопасность SQL — это базовая линия: на уровне MCP Server обязательно нужна проверка на режим только чтения, аккаунт базы данных должен иметь минимальные права (пользователь только для чтения), не давайте агенту права на запись. Обнаружение схемы должно быть умным: при большом числе таблиц нельзя загружать всё сразу — можно использовать векторный поиск (Qdrant/ES) для семантического сопоставления и подтягивать только структуру релевантных таблиц. Добавьте условие выхода из цикла: цикл «генерация SQL → проверка → выполнение → исправление» должен ограничивать максимальное число попыток, чтобы не уйти в бесконечный цикл. Подключите LangSmith для трассировки: записывайте весь процесс генерации, проверки и выполнения каждого SQL — так проще разбираться, «почему сгенерировался неправильный SQL». Разворачивайте MCP Server по HTTP: в продакшене используйте транспорт streamable-http и выносите MCP Server в отдельный микросервис — так удобнее централизованно управлять аутентификацией и ограничением нагрузки. Итог Ключевая идея этой архитектуры: LangGraph управляет процессом, MCP управляет инструментами. LangGraph с помощью конечного автомата выстраивает «обнаружение схемы → генерация SQL → проверка → выполнение → интерпретация» в управляемый конвейер; MCP стандартизирует операции с базой данных, агент вызывает инструменты по единому протоколу — и при смене базы или структуры таблиц код агента менять не нужно. По сравнению с тем, когда LLM просто пишет SQL напрямую, здесь добавляются две страховки — обнаружение схемы и двойная проверка, — и точность, и безопасность становятся более управляемыми.
Источник: Juejin · Популярное об ИИ за неделю · juejin.cn