LangGraph+DuckDB+ReActAgent实战：Excel问答助手开发指南（超详细）从入门到精通，收藏这篇就够了！

本文介绍了一个基于LangGraph+DuckDB+ReActAgent架构的Excel问答助手系统。该系统能够将Excel文件自动映射为数据库表结构，通过自然语言生成SQL查询并智能推荐可视化图表。采用会话隔离设计支持多用户并发，使用DuckDB作为内存数据库实现高性能分析，并通过LangGraph实现工作流编排。系统支持多文件、多Sheet的统一分析，并提供流式响应和链路追踪功能，使非技术人员也能通过自然语言进行复杂数据分析。

基于 LangGraph + DuckDB + ReActAgent 的 Excel 问答助手

unsetunset前言unsetunset

在数据分析场景中，Excel 文件是最常见的数据载体之一。但传统的 Excel 数据分析往往需要用户具备一定的技术能力，比如熟悉公式、透视表或者 SQL 查询。能不能让用户用自然语言直接提问，系统自动完成数据分析并给出可视化结果？

本文将详细介绍一个基于LangGraph + DuckDB + ReActAgent架构的 Excel 问答系统，它能够：

• 支持多文件、多 Sheet 的统一分析
• 自动将 Excel 数据映射为数据库表结构
• 根据自然语言问题生成 SQL 查询
• 智能推荐可视化图表类型
• 实时流式返回思考过程和分析结果

unsetunset👉 欢迎关注我的开源项目unsetunset

🎬 运行效果预览

📌 如果您这个项目感兴趣，欢迎「付费点赞」支持一下！

👉 下方点击「喜欢作者」，金额随意，心意无价。让我们在技术路上彼此赋能，少走弯路，高效落地！

付费后，请务必添加我的微信（微信号：weber812）并发送支付凭证，我将第一时间拉您进入专属「技术支持群」。

进群后您能获得哪些福利？👉 完整代码具体请见下方说明！

unsetunset一、系统架构设计unsetunset

1.1 整体架构

整个系统采用LangGraph作为编排引擎，将 Excel 问答流程拆解为多个节点，每个节点负责特定的任务。

用户问题 → Excel解析 → SQL生成 → SQL执行 → 结果总结 → 数据渲染

1.2 技术栈选型

1.3 核心设计理念

会话隔离：每个chat_id拥有独立的 DuckDB 实例，避免多用户数据混淆。

流式响应：采用 SSE（Server-Sent Events）实时推送思考过程，提升用户体验。

智能推荐：不仅生成 SQL，还根据查询逻辑推荐最合适的图表类型（折线图、柱状图、表格等）。

unsetunset二、核心模块详解unsetunset

2.1 状态管理（ExcelAgentState）

LangGraph 的核心是状态流转。我们定义了ExcelAgentState来管理整个问答流程的状态：

class ExcelAgentState(TypedDict): user_query: str # 用户问题 file_list: list # 文件列表 chat_id: Optional[str] # 聊天ID（会话隔离） file_metadata: Dict[str, FileInfo] # 文件元数据 sheet_metadata: Dict[str, SheetInfo] # Sheet元数据 db_info: list[dict] # 表结构信息 catalog_info: Dict[str, str] # Catalog映射 generated_sql: Optional[str] # 生成的SQL chart_url: Optional[str] # 图表URL chart_type: Optional[str] # 图表类型 apache_chart_data: Optional[Dict[str, Any]] # ECharts数据 execution_result: Optional[ExecutionResult] # SQL执行结果 report_summary: Optional[str] # 分析报告

设计亮点：

• 使用TypedDict确保类型安全
• 区分文件级元数据（file_metadata）和表级元数据（sheet_metadata）
• 支持多 Catalog 管理，每个文件对应一个独立的 Catalog

2.2 DuckDB 连接管理

2.2.1 为什么选择 DuckDB？

2.2.2 会话级连接管理

为了支持多用户并发，我们设计了两层管理器：

# 第一层：单个 Excel 文件的 DuckDB 管理器class ExcelDuckDBManager: def __init__(self): self._connection = None # 延迟初始化 self._registered_catalogs = {} # Catalog 注册表 self._registered_tables = {} # 表注册表 def register_excel_file(self, file_path, file_name): """注册 Excel 文件到 DuckDB""" # 1. 读取所有 Sheet # 2. 清理列名（去除特殊字符） # 3. 创建 Catalog 和 Table # 4. 返回表结构信息# 第二层：聊天级别的管理器（会话隔离）class ChatDuckDBManager: def __init__(self): self._chat_managers = {} # {chat_id: ExcelDuckDBManager} self._session_timeout = 36000# 10小时超时 def get_manager(self, chat_id): """为每个 chat_id 创建独立的管理器""" if chat_id notin self._chat_managers: self._chat_managers[chat_id] = ExcelDuckDBManager() return self._chat_managers[chat_id]

核心优势：

• 会话隔离：不同用户的数据互不干扰
• 资源复用：同一会话内多次查询共享连接
• 自动清理：超时会话自动释放资源

2.2.3 数据注册流程

def register_excel_file(self, file_path: str, file_name: str): # 1. 生成唯一的 Catalog 名称 catalog_name = self._get_unique_catalog_name(file_name) # 2. 读取 Excel 所有 Sheet excel_file_data = pd.ExcelFile(file_path) dataframes = [(sheet, pd.read_excel(file_path, sheet_name=sheet)) for sheet in excel_file_data.sheet_names] # 3. 清理列名（去除特殊字符、中文支持） for sheet_name, df in dataframes: df.columns = [self._sanitize_column_name(col) for col in df.columns] # 4. 注册到 DuckDB conn.execute(f"CREATE SCHEMA IF NOT EXISTS {catalog_name}") conn.execute(f'CREATE TABLE "{catalog_name}"."{table_name}" AS SELECT * FROM df') # 5. 提取表结构信息 columns_info = {col: {"comment": col, "type": self._map_pandas_dtype_to_sql(df[col].dtype)} for col in df.columns} return catalog_name, registered_tables

关键技术点：

• 列名清理：支持中文列名，自动处理特殊字符
• 类型映射：Pandas dtype → SQL type（如int64→BIGINT）
• 样本数据：保存前 5 行数据用于 SQL 生成时的上下文参考

2.3 SQL 生成节点

这是整个系统的核心，需要将自然语言转换为精确的 SQL 查询。

2.3.1 Prompt 设计

prompt = ChatPromptTemplate.from_template("""你是一位专业的数据库管理员（DBA），任务是根据提供的数据库结构、表关系以及用户需求，生成优化的DUCKDB SQL查询语句，并推荐合适的可视化图表。## 任务 - 根据用户问题生成一条优化的SQL语句。 - 根据查询生成逻辑从**图表定义**中选择最合适的图表类型。## 生成SQL要求： - 你必须仅生成一条合法、可执行的SQL查询语句 ， 不得包含解释、Markdown、注释或额外文本。 - **必须直接且完整地使用所提供的表结构和表关系来生成SQL语句**，表的格式必须是 catalog_name.table_name 的形式，生成的SQL语句中的catalog、table、column 必须用 "" 符号包裹。 - 你必须严格遵守数据类型、外键关系及表结构中定义的约束。当没有表的关联关系时，你可以根据样例数据进行推断。 - 不得假设表结构中未明确定义的列或表。 - 显式声明所有JOIN条件（禁止自然连接）；注意join性能，尽可能的在join之间先进行'group by'操作以减少join的数据量。 - 当遇到复杂查询时，使用WITH CTE分层组织复杂逻辑。 - 若问题涉及时序，请合理使用提供的"当前时间"上下文（例如用于相对日期计算）。 - 如果用户问题模糊或者缺乏足够的信息以生成正确的查询，请返回：`NULL` - 当用户明确要求查看明细数据且未指定具体数量时，应适当限制返回结果数量（如LIMIT 50）以提高查询性能，但如果用户指定了具体数量则按照用户要求执行 - 对于聚合查询或统计类查询，不应随意添加LIMIT子句## 提供的信息 - 表结构：{db_schema} - 用户提问：{user_query} - 当前时间：{current_time}## 图表定义 - generate_line_chart: 用于显示数据随时间或连续变量的趋势 - generate_bar_chart: 用于比较不同类别的值，适合横向对比 - generate_pie_chart: 用于显示数据的占比关系 - generate_table: 生成结构化表格，用于展示明细数据 - ... （共20+种图表类型）## 输出格式 - 你**必须且只能**输出一个符合以下结构的 **纯 JSON 对象**： ```json {{ "sql_query": "生成的SQL语句字符串", "chart_type": "推荐的图表类型字符串，如 \"generate_area_chart\"" }}

Prompt 设计要点：

2.3.2 表结构上下文构建

db_schema = [ { "table_name": "销售数据_Sheet1", "catalog_name": "销售数据", "columns": { "日期": {"comment": "日期", "type": "VARCHAR(255)"}, "产品": {"comment": "产品", "type": "VARCHAR(255)"}, "销售额": {"comment": "销售额", "type": "BIGINT"} }, "sample_data": [ {"日期": "2024-01-01", "产品": "A", "销售额": 1000}, ... ], "table_comment": "销售数据.xlsx - Sheet1" }]

关键信息：

• 列注释：保留原始列名作为注释（支持中文）
• 样本数据：帮助 LLM 理解数据分布
• 表注释：标注数据来源（文件名 + Sheet名）

2.4 SQL 执行节点

def exe_sql_excel_query(state): # 1. 获取 DuckDB 管理器 duckdb_manager = get_duckdb_manager(chat_id=state.get("chat_id")) # 2. 清理 SQL（移除反引号） sql = state["generated_sql"].replace("`", "") # 3. 执行查询 columns, data = duckdb_manager.execute_sql(sql) # 4. 封装结果 state["execution_result"] = ExecutionResult( success=True, columns=columns, data=data ) return state

执行结果结构：

class ExecutionResult(BaseModel): success: bool # 是否成功 columns: List[str] # 列名列表 data: Optional[List[Dict[str, Any]]] # 查询结果（字典列表） error: Optional[str] # 错误信息

2.5 数据渲染节点

根据图表类型选择不同的渲染方式：

def data_render_condition(state: ExcelAgentState) -> str: chart_type = state.get("chart_type") # 表格类型使用 ECharts 渲染 if chart_type.lower() in ["mcp-server-chart-generate_table"]: return "data_render_apache" # 其他图表使用 AntV MCP 渲染 return "data_render"

2.5.1 表格渲染（ECharts）

def excel_data_render_apache(state): # 1. 解析 SQL 获取列信息 is_select_all = check_if_select_all(generated_sql) if is_select_all: # SELECT * 查询：从表结构中获取所有列 column_comments = get_all_column_comments_for_tables(table_names, db_info) else: # 普通查询：解析 SELECT 子句 column_comments = extract_select_columns_with_comments(generated_sql, db_info) # 2. 构建表格数据 table_data = { "llm": {"type": "response_table"}, "data": { "column": column_comments, # 列标题（中文） "result": [ {column_comments[i]: row[actual_columns[i]] for i in range(len(columns))} for row in data_result.data ] } } return state

技术难点：

• 列名映射：将数据库列名映射为中文注释
• SELECT * 处理：需要从表结构中提取所有列信息
• 别名处理：聚合函数的别名优先级高于列注释

2.5.2 图表渲染（AntV MCP）

对于非表格类型的图表，调用 AntV MCP 服务生成可视化：

def data_render_ant(state): # 1. 准备图表数据 chart_data = { "type": state["chart_type"], "data": state["execution_result"].data } # 2. 调用 MCP 服务生成图表 chart_url = mcp_client.generate_chart(chart_data) state["chart_url"] = chart_url return state

2.6 流式响应设计

为了提升用户体验，系统采用 SSE 实时推送每个节点的执行进度：

async def run_excel_agent(self, query, response, chat_id, ...): # 1. 异步流式执行 LangGraph asyncfor chunk_dict in graph.astream(input=initial_state, stream_mode="updates"): step_name, step_value = next(iter(chunk_dict.items())) # 2. 发送步骤标题（折叠面板） if step_name notin ["summarize", "data_render"]: await response.write(f'<details><summary>{step_name}...</summary>') # 3. 发送步骤内容 content = self._format_step_content(step_value) await response.write(content) # 4. 关闭折叠面板 if step_name notin ["summarize", "data_render"]: await response.write('</details>')

前端渲染效果：

▼ excel_parsing... 文件解析结果 文件列表： 1. file_name: 销售数据.xlsx | Catalog: 销售数据 | Sheets: 2 | 上传时间: 2024-11-28▼ sql_generator... SELECT "销售数据"."Sheet1"."产品", SUM("销售数据"."Sheet1"."销售额") AS 总销售额 FROM "销售数据"."Sheet1" GROUP BY "销售数据"."Sheet1"."产品"▼ sql_executor... ✅ 查询执行成功！返回 5 行数据，2 列分析报告：根据查询结果，产品A的总销售额为50000元，产品B为38000元...[表格/图表渲染]

unsetunset三、关键技术实现unsetunset

3.1 多文件多 Sheet 支持

3.1.1 Catalog 隔离机制

# 文件1：销售数据.xlsx（2个Sheet）Catalog: 销售数据 - Table: Sheet1 - Table: Sheet2# 文件2：库存数据.csvCatalog: 库存数据 - Table: 库存数据# SQL 查询示例SELECT s."产品", s."销售额", i."库存量"FROM "销售数据"."Sheet1" sJOIN "库存数据"."库存数据" i ON s."产品" = i."产品"

优势：

• 避免表名冲突（不同文件可能有同名 Sheet）
• 清晰的数据来源标识
• 支持跨文件关联查询

3.1.2 表名清理规则

def _sanitize_table_name(self, sheet_name: str) -> str: # 1. 替换非法字符（保留中文） table_name = re.sub(r'[^\w\u4e00-\u9fa5]', '_', sheet_name) # 2. 移除首尾下划线 table_name = table_name.strip('_') # 3. 确保不以数字开头 if table_name and table_name[0].isdigit(): table_name = f'table_{table_name}' return table_name or 'unknown_sheet'

处理示例：

3.2 SQL 解析与列名映射

使用SQLGlot解析 SQL 语句，提取列信息并映射为中文注释。

3.2.1 提取表名

def extract_table_names_sqlglot(sql: str) -> list: expression = parse(sql)[0] tables = set() for table in expression.find_all(sqlglot.exp.Table): tables.add(table.name) return list(tables)

3.2.2 提取列名并映射注释

def extract_select_columns_with_comments(sql, schema_inspector): expressions = parse(sql) column_info_list = [] for expression in expressions: selects = expression.find_all(sqlglot.exp.Select) for select in selects: for proj in select.expressions: # 处理别名 if isinstance(proj, sqlglot.exp.Alias): column_info_list.append({ "name": proj.this.name, "alias": proj.alias, "is_aggregate": isinstance(proj.this, (sqlglot.exp.Sum, ...)) }) # 处理普通列 elif isinstance(proj, sqlglot.exp.Column): column_info_list.append({ "name": proj.name, "table": proj.table }) # 映射为中文注释 result_columns = [] for col_info in column_info_list: if col_info.get("is_aggregate") or col_info.get("alias"): # 聚合函数保留别名 result_columns.append(col_info["alias"]) else: # 普通列查找注释 comment = find_column_comment(col_info["name"], col_info["table"], schema_inspector) result_columns.append(comment or col_info["name"]) return result_columns

处理示例：

3.3 链路追踪（Langfuse）

集成 Langfuse 记录每次对话的完整执行链路：

if self.ENABLE_TRACING: langfuse = get_client() with langfuse.start_as_current_observation( input=query, as_type="agent", name="表格问答" ) as rootspan: rootspan.update_trace(session_id=chat_id, user_id=user_id) # 执行 LangGraph async for chunk in graph.astream(...): # 每个节点的执行会自动记录到 Langfuse ...

追踪信息包括：

• 每个节点的输入输出
• LLM 调用的 token 消耗
• 执行耗时
• 错误堆栈

3.4 任务取消机制

支持用户中途取消长时间运行的任务：

class ExcelAgent: def __init__(self): self.running_tasks = {} # {task_id: {"cancelled": False}} asyncdef run_excel_agent(self, ...): task_id = user_dict["id"] self.running_tasks[task_id] = {"cancelled": False} asyncfor chunk in graph.astream(...): # 检查是否已取消 if self.running_tasks[task_id]["cancelled"]: await response.write("> 这条消息已停止") break # 处理节点输出 ... asyncdef cancel_task(self, task_id: str): if task_id in self.running_tasks: self.running_tasks[task_id]["cancelled"] = True

unsetunset四、实战案例分析unsetunset

案例1：单文件简单查询

用户问题：销售额最高的前5个产品是哪些？

执行流程：

1. excel_parsing - 注册文件：销售数据.xlsx - Catalog: 销售数据 - Table: Sheet1 (100行, 3列)2. sql_generator - 生成SQL: SELECT "销售数据"."Sheet1"."产品", SUM("销售数据"."Sheet1"."销售额") AS 总销售额 FROM "销售数据"."Sheet1" GROUP BY "销售数据"."Sheet1"."产品" ORDER BY 总销售额 DESC LIMIT 5 - 推荐图表: generate_bar_chart3. sql_executor - 返回 5 行数据4. summarize - 生成报告：根据销售数据分析，销售额最高的5个产品分别是...5. data_render - 渲染柱状图

案例2：多文件关联查询

用户问题：哪些产品的库存量低于销售量的10%？

执行流程：

1. excel_parsing - 注册文件1：销售数据.xlsx → Catalog: 销售数据 - 注册文件2：库存数据.csv → Catalog: 库存数据2. sql_generator - 生成SQL: SELECT s."产品", SUM(s."销售额") AS 总销售额, i."库存量" FROM "销售数据"."Sheet1" s JOIN "库存数据"."库存数据" i ON s."产品" = i."产品" GROUP BY s."产品", i."库存量" HAVING i."库存量" < SUM(s."销售额") * 0.1 - 推荐图表: generate_table3. sql_executor - 返回 3 行数据4. summarize - 生成报告：有3个产品的库存量低于销售量的10%，分别是...5. data_render_apache - 渲染表格

unsetunset五、性能优化与最佳实践unsetunset

5.1 性能优化策略

5.2 错误处理

try: # 执行 SQL columns, data = duckdb_manager.execute_sql(sql) state["execution_result"] = ExecutionResult(success=True, ...)except Exception as e: # 记录错误 logger.error(f"SQL执行失败: {e}") state["execution_result"] = ExecutionResult( success=False, error=str(e) )

常见错误类型：

• SQL 语法错误（列名不存在、表名错误）
• 类型转换错误（字符串与数字比较）
• 内存溢出（大数据集未 LIMIT）

5.3 安全性考虑

unsetunset六、总结unsetunset

6.1 核心优势

6.2 技术亮点

1. 会话隔离：每个 chat_id 独立的 DuckDB 实例，支持多用户并发
2. 智能映射：自动将 Excel 列名映射为中文注释
3. 流式响应：SSE 实时推送思考过程，提升用户体验
4. 图表推荐：根据查询逻辑智能推荐 20+ 种图表类型
5. SQL 优化：Prompt 引导生成高性能 SQL（先 GROUP BY 再 JOIN）

unsetunset📚 完整代码unsetunset

项目地址:https://github.com/apconw/sanic-web

🌈 项目亮点

• ✅ 集成 MCP 多智能体架构
• ✅ 支持 Dify / LangChain / LlamaIndex / Ollama / vLLM /Neo4j
• ✅ 前端采用 Vue3 + TypeScript + Vite5，现代化交互体验
• ✅ 内置 ECharts / AntV 图表问答 + CSV 表格问答
• ✅ 支持对接主流 RAG 系统 与 Text2SQL 引擎
• ✅ 轻量级 Sanic 后端，适合快速部署与二次开发
• ✅项目已被蚂蚁官方推荐收录

如何学习AI大模型 ？

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。【保证100%免费】🆓

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取【保证100%免费】

读者福利：👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈

对于0基础小白入门：

如果你是零基础小白，想快速入门大模型是可以考虑的。

一方面是学习时间相对较短，学习内容更全面更集中。
二方面是可以根据这些资料规划好学习计划和方向。

👉1.大模型入门学习思维导图👈

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

对于从来没有接触过AI大模型的同学，我们帮你准备了详细的学习成长路线图&学习规划。可以说是最科学最系统的学习路线，大家跟着这个大的方向学习准没问题。（全套教程文末领取哈）

👉2.AGI大模型配套视频👈

很多朋友都不喜欢晦涩的文字，我也为大家准备了视频教程，每个章节都是当前板块的精华浓缩。

👉3.大模型实际应用报告合集👈

这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。（全套教程文末领取哈）

👉4.大模型实战项目&项目源码👈

光学理论是没用的，要学会跟着一起做，要动手实操，才能将自己的所学运用到实际当中去，这时候可以搞点实战项目来学习。（全套教程文末领取哈）

👉5.大模型经典学习电子书👈

随着人工智能技术的飞速发展，AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型，如GPT-3、BERT、XLNet等，以其强大的语言理解和生成能力，正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。（全套教程文末领取哈）

👉6.大模型面试题&答案👈

截至目前大模型已经超过200个，在大模型纵横的时代，不仅大模型技术越来越卷，就连大模型相关的岗位和面试也开始越来越卷了。为了让大家更容易上车大模型算法赛道，我总结了大模型常考的面试题。（全套教程文末领取哈）

为什么分享这些资料?

只要你是真心想学AI大模型，我这份资料就可以无偿分享给你学习，我国在这方面的相关人才比较紧缺，大模型行业确实也需要更多的有志之士加入进来，我也真心希望帮助大家学好这门技术，如果日后有什么学习上的问题，欢迎找我交流，有技术上面的问题，我是很愿意去帮助大家的！

这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取【保证100%免费】

读者福利：👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈