1. 项目概述当数据库查询遇上大模型我们到底在造什么轮子“Data Query Visualisation using LLM Agents from Scratch”——这个标题乍看像一句技术口号但拆开来看它其实精准锚定了当前数据工作流中一个真实存在的断层一边是业务人员对着SQL编辑器发呆一边是数据工程师在Jupyter里反复调试plotly参数而中间那条本该畅通无阻的“从问题到图表”的通路常年堵着三座大山自然语言理解不准、SQL生成不可控、可视化意图难对齐。我做这个项目不是为了炫技而是上个月被市场部同事拉进一个紧急会议他们拿着一份PDF版的销售周报截图问我“能不能把‘华东区上月TOP5门店的复购率趋势’直接变成折线图”——那一刻我意识到我们缺的不是更强大的BI工具而是一个能听懂人话、会写靠谱SQL、还知道什么时候该用堆叠柱状图而不是雷达图的“数字协作者”。这个项目就是从零开始亲手搭出这样一个协作者它不依赖任何现成的LLM应用框架所有Agent调度逻辑、SQL校验规则、图表类型决策树、甚至错误恢复机制全部手写Python实现。核心关键词——LLM Agent、自然语言转SQL、动态可视化生成、安全沙箱执行、意图链式推理——每一个都不是调个API就完事而是要掰开揉碎搞清楚它在真实数据场景里怎么呼吸、怎么犯错、又怎么自我修复。适合谁如果你是数据分析师想甩掉写SQL的重复劳动如果你是后端工程师正为内部数据平台加智能查询入口或者你是技术负责人在评估是否值得自建Agent层而非采购SaaS方案——这篇文章里的每一步代码、每一次失败重试、每一处人工兜底设计都是我在生产环境里踩出来的实感。2. 整体架构设计与核心思路拆解为什么必须“从Scratch”2.1 拒绝黑盒封装三层解耦的Agent协作范式市面上很多“LLMBI”方案喜欢把整个流程塞进一个大模型提示词里比如让GPT-4直接输出带plotly代码的JSON。这种做法在demo阶段很惊艳但一上线就露馅当用户问“对比Q3和Q4的客单价中位数按城市分组”模型可能生成SELECT city, PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY order_amount)——这在PostgreSQL里没问题但我们的生产库是MySQL 5.7根本不支持窗口函数。更糟的是它可能把“中位数”错译成AVG()而你根本没法在JSON里定位并修正这个计算逻辑。所以我的第一原则是绝不让LLM直接生成最终可执行代码。整个系统拆成三个独立Agent每个只干一件事且彼此之间用结构化Schema通信Query Planner Agent输入自然语言输出结构化查询意图含时间范围、维度、指标、聚合方式、过滤条件。它不碰SQL只输出类似{time_range: [2024-07-01, 2024-09-30], dimensions: [city], metrics: [{name: order_amount, agg: median}], filters: [{field: region, op: , value: East}]}的JSON。这里的关键是强制它把“中位数”明确标注为agg: median而不是让它自由发挥。SQL Generator Agent接收Planner的JSON结合数据库Schema元数据表名、字段类型、索引信息生成符合目标DB方言的SQL。它内置了方言适配器对MySQL自动降级为SELECT AVG(order_amount) FROM (SELECT order_amount FROM sales WHERE regionEast ORDER BY order_amount LIMIT 2 OFFSET (SELECT COUNT(*)/2 FROM sales)) t这类模拟中位数的写法对PostgreSQL则直出PERCENTILE_CONT。重点在于它生成的SQL必须通过静态语法检查用sqlparse解析AST和基础语义校验如检查字段是否存在于指定表中。Viz Designer Agent接收SQL执行结果DataFrame和原始查询意图决定可视化形式。它的决策树不是靠LLM猜而是硬编码规则当维度1且指标1时用柱状图当维度1且指标1时用分组柱状图当维度2且指标1时用热力图当时间维度存在且指标1时强制用折线图。只有当规则无法覆盖如用户明确说“画桑基图”时才触发LLM辅助生成plotly配置。提示这种解耦不是为了增加复杂度而是为了可调试性。当图表出错时你能立刻定位是Planner理解错了“TOP5”还是Generator在MySQL里没处理好LIMIT子句还是Designer把多维数据误判为单维——而不是面对一整段LLM输出的plotly代码束手无策。2.2 安全沙箱为什么连SQL执行都要“戴镣铐跳舞”让LLM生成的SQL直接连生产库这是拿公司数据安全开玩笑。我的沙箱设计有三道锁连接层隔离Agent永远不接触真实数据库连接字符串。所有SQL执行请求都发给一个独立的QueryExecutor服务它只接受来自Agent的HTTP POST请求且请求体必须包含query_hashSQL内容的SHA256哈希值和schema_version数据库Schema快照版本号。Executor启动时会加载当前Schema元数据并缓存一份只读副本。当收到请求时它先校验query_hash是否在白名单内白名单由离线审核脚本生成再用缓存的Schema验证SQL中引用的表和字段是否存在——这意味着即使Agent被注入恶意提示词它生成的SQL如果引用了不存在的users_passwords表会在执行前就被拦截。资源熔断Executor对每个查询设置硬性限制最大执行时间3秒、最多返回10000行、禁止UPDATE/DELETE/DROP等写操作。这些不是靠SQL解析器简单匹配关键词if UPDATE in sql:而是用sqlparse解析AST严格检查token.ttype是否为DML.UPDATE。我试过用/* MAX_EXECUTION_TIME1000 */ UPDATE ...这种MySQL hint绕过结果被AST校验直接打回。结果脱敏Executor返回的JSON结果中所有字段名自动小写避免前端因大小写敏感报错数值型字段统一保留4位小数字符串字段长度截断至255字符防超长文本撑爆前端内存且敏感字段如phone,id_card在Schema元数据中标记为is_sensitivetrueExecutor会自动将其值替换为***。这个标记不是靠LLM识别而是DBA在初始化Schema时手动配置的。注意这套沙箱看似繁琐但它把“信任边界”划得极其清晰——LLM只负责理解意图和生成逻辑所有执行、校验、脱敏都由确定性代码完成。这比任何“用RLHF微调模型不生成危险SQL”的方案都可靠。2.3 可视化生成为什么拒绝“LLM直接写plotly代码”很多人觉得让LLM生成plotly或matplotlib代码最省事。我试过结果惨烈模型会写出fig.update_layout(title_textSales Trend, title_x0.5)但忘了加fig.show()或者把xaxis_title写成x_title导致报错更常见的是它把时间序列的X轴设为category类型导致折线图变成离散点。根本原因在于plotly API有超过200个参数而LLM的上下文窗口装不下完整文档它只能靠概率猜。我的替代方案是Viz Designer Agent只输出一个极简的VizSpec对象class VizSpec: chart_type: Literal[line, bar, heatmap, pie] # 强制枚举 x_field: str # DataFrame列名 y_fields: List[str] # 多指标时用列表 title: str x_label: str y_label: str然后由一个纯Python的VizRenderer模块根据chart_type调用预定义的模板函数。比如line模板def render_line(spec: VizSpec, df: pd.DataFrame) - go.Figure: fig go.Figure() for y_col in spec.y_fields: # 自动处理时间字段若x_field是datetime则设xaxis_typedate if pd.api.types.is_datetime64_any_dtype(df[spec.x_field]): fig.add_trace(go.Scatter(xdf[spec.x_field], ydf[y_col], namey_col)) fig.update_xaxes(typedate) else: fig.add_trace(go.Scatter(xdf[spec.x_field], ydf[y_col], namey_col)) fig.update_layout(titlespec.title, xaxis_titlespec.x_label, yaxis_titlespec.y_label) return fig这个模板里时间轴自动识别、多曲线自动命名、布局统一规范——所有LLM容易出错的细节都由确定性代码兜底。Agent只需专注做它最擅长的事从用户问题中精准提取x_field如“按月份”→month、y_fields如“复购率”→repeat_rate、chart_type“趋势”→line。实测下来这种模式生成的图表100%可运行且风格高度一致。3. 核心细节解析与实操要点从Prompt工程到Schema感知3.1 Query Planner Agent如何让LLM真正“听懂人话”Planner的核心不是让LLM多聪明而是用结构化约束把它框死在安全区。它的System Prompt长这样精简版你是一个数据库查询规划器严格按以下规则工作 1. 输入用户自然语言问题例如“上个月销售额最高的3个产品类别” 2. 输出仅JSON无任何额外文字。JSON必须包含且仅包含以下字段 - time_range: [start_date, end_date] 字符串数组格式YYYY-MM-DD。若未提时间默认为[2024-09-01, 2024-09-30] - dimensions: 字符串列表如[product_category] - metrics: 对象列表每个对象含name(字段名)、agg(聚合函数仅限sum,count,avg,max,min,median) - filters: 对象列表每个对象含field(字段名)、op(操作符仅限,!,,,LIKE)、value(字符串值) - limit: 整数若提TOP N则填N否则为null 3. 禁止推断若问题未提过滤条件不要自行添加statusactive等假设。 4. 字段名必须来自已知Schemaproducts表有[id,name,category,price]sales表有[id,product_id,amount,created_at]关键设计点强制默认值time_range设默认值避免LLM因时间模糊而拒绝输出。我测试过当用户问“最近销量”不设默认值时约30%的请求会因LLM无法确定“最近”指几天而返回空JSON。聚合函数白名单限定agg只能是6个确定性函数彻底杜绝stddev_pop或rank()这类冷门函数导致Generator无法适配方言。Schema硬编码把表结构直接写进Prompt而不是让LLM去查外部知识库。因为Schema变更频率低通常月更而每次变更只需更新Prompt中的这段描述比维护向量数据库简单得多。我用脚本自动从DB导出DDL再格式化成Prompt片段。禁止推断条款这是血泪教训。早期没加这条LLM看到“销售额最高的产品”会自动加WHERE statusactive结果把下架产品的历史数据漏掉了。加上后它老老实实输出filters: []。实操心得Prompt里每一条规则都对应一个线上故障。比如op只允许5个操作符是因为我们发现LLM会生成op: CONTAINS而SQL里根本没有这个操作符Generator解析时直接崩溃。现在所有规则都是故障驱动的。3.2 SQL Generator Agent方言适配不是玄学是查表作业Generator的输入是Planner的JSON和数据库Schema元数据。它的核心任务是把抽象意图翻译成具体SQL。难点在于方言差异。以“取前N条”为例数据库标准写法MySQL 5.7PostgreSQLSQLite通用SELECT * FROM t ORDER BY x LIMIT N✅ 支持✅ 支持✅ 支持但中位数PERCENTILE_CONT(0.5)❌ 不支持✅ 支持❌ 不支持我的解决方案是建一张“方言能力矩阵表”Python dictDIALECT_CAPABILITIES { mysql: { limit_clause: LIMIT {n}, offset_clause: OFFSET {n}, median_function: SELECT AVG(val) FROM (SELECT order_amount as val FROM sales WHERE {filters} ORDER BY val LIMIT 2 OFFSET (SELECT FLOOR((COUNT(*)-1)/2) FROM sales WHERE {filters})) t, case_when: CASE WHEN {cond} THEN {then} ELSE {else} END }, postgresql: { limit_clause: LIMIT {n}, offset_clause: OFFSET {n}, median_function: PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY {field}), case_when: CASE WHEN {cond} THEN {then} ELSE {else} END } }Generator的工作流根据schema_version确定目标方言如mysql_57遍历Planner JSON中的metrics对每个agg: median用DIALECT_CAPABILITIES[mysql][median_function]模板填充{field}和{filters}对limit字段用limit_clause模板填充最后用sqlparse.format()美化SQL确保缩进统一注意这个矩阵表不是凭空写的而是我花两天时间把MySQL 5.7、8.0、PostgreSQL 12、14、SQLite 3.35的官方文档里所有聚合函数、窗口函数、分页语法逐一对比整理出来的。它就像一本方言词典让Generator不用“思考”只管“查表”。3.3 Viz Designer Agent用规则引擎替代LLM幻觉Designer的Prompt非常短因为它只做一件事从Planner的JSON和SQL结果DataFrame中提取可视化所需的最小必要信息。它的System Prompt你是一个可视化规格生成器。输入1) Planner输出的JSON2) SQL执行返回的DataFrame含列名和前3行示例。输出仅JSON含字段chart_typeline/bar/heatmap/pie、x_fieldX轴字段名、y_fieldsY轴字段名列表、title、x_label、y_label。规则 - 若Planner中time_range非空且x_field是日期类型 → chart_typeline - 若Planner中dimensions长度为1且metrics长度为1 → chart_typebar - 若Planner中dimensions长度为2且metrics长度为1 → chart_typeheatmap - 若Planner中metrics长度为1且filters中含LIKE操作 → chart_typepie - x_field必须是DataFrame中存在的列名优先选dimensions[0]若为空则选第一个非指标列 - y_fields必须是metrics中name字段若为空则选DataFrame中数值型列关键技巧利用DataFrame实际数据Prompt里强调“输入含前3行示例”是为了让LLM能判断created_at列的值是2024-09-01日期还是Sep 2024字符串。我测试过只给Schema不给示例时LLM对VARCHAR字段的类型判断准确率只有65%给了3行数据后提升到92%。规则优先于LLM所有chart_type决策都基于硬编码规则LLM只是执行者。这保证了行为可预测。比如只要dimensions[city]且metrics[{name:sales,agg:sum}]就一定是bar不会因为某次温度高就变成pie。兜底策略当规则无法覆盖如dimensions[region,city]且metrics[{name:profit,agg:sum}]Prompt强制要求chart_typeheatmap而不是让LLM自由发挥。热力图是多维聚合数据最安全的默认选项。提示Designer的输出JSON会被VizRenderer模块严格校验。如果x_field不在DataFrame列中渲染器会抛出ValueError并记录日志触发告警。这比让LLM自己检查更可靠。4. 实操过程与核心环节实现从零搭建可运行Agent链4.1 环境准备与依赖安装轻量级不碰Docker整个系统跑在一台16GB内存的Ubuntu 22.04服务器上不依赖Docker或K8s所有服务用systemd管理。核心依赖只有5个pip install openai pandas plotly python-dotenv sqlparse psycopg2-binary mysql-connector-pythonopenai: 调用GPT-3.5-turbo API成本可控$0.5/百万tokenpandas: 处理SQL执行结果plotly: 渲染交互式图表导出HTML或PNGsqlparse: SQL语法解析与格式化非执行psycopg2-binary/mysql-connector-python: 数据库驱动按需安装注意我刻意避开了LangChain、LlamaIndex等框架。它们抽象层太厚当Generator需要修改MySQL中位数写法时你要在LangChain的SQLDatabaseChain源码里找半天而手写代码改一行median_function模板就搞定。对“从Scratch”项目轻量即正义。4.2 Schema元数据采集让Agent“认识”你的数据库Agent必须知道表结构才能生成合法SQL。我写了一个schema_collector.py脚本每天凌晨2点自动执行import mysql.connector from sqlalchemy import create_engine import json def collect_mysql_schema(host, user, password, database): engine create_engine(fmysqlmysqlconnector://{user}:{password}{host}/{database}) # 获取所有表名 tables engine.execute(SHOW TABLES).fetchall() schema {} for table in tables: table_name table[0] # 获取列信息字段名、类型、是否主键、是否为空 cols engine.execute(fDESCRIBE {table_name}).fetchall() schema[table_name] [ { name: col[0], type: col[1], is_primary: col[3] PRI, is_nullable: col[2] YES, is_sensitive: col[0] in [phone, email, id_card] # DBA手动维护 } for col in cols ] # 写入JSON文件供Agent加载 with open(f/opt/agent/schema/{database}_schema.json, w) as f: json.dump(schema, f, indent2) return schema # 调用示例 collect_mysql_schema(db-prod.internal, readonly, xxx, sales_db)这个脚本产出的sales_db_schema.json长这样{ products: [ {name: id, type: int(11), is_primary: true, is_nullable: false, is_sensitive: false}, {name: name, type: varchar(255), is_primary: false, is_nullable: true, is_sensitive: false}, {name: category, type: varchar(100), is_primary: false, is_nullable: true, is_sensitive: false}, {name: price, type: decimal(10,2), is_primary: false, is_nullable: true, is_sensitive: false} ], sales: [ {name: id, type: int(11), is_primary: true, is_nullable: false, is_sensitive: false}, {name: product_id, type: int(11), is_primary: false, is_nullable: false, is_sensitive: false}, {name: amount, type: decimal(10,2), is_primary: false, is_nullable: false, is_sensitive: false}, {name: created_at, type: datetime, is_primary: false, is_nullable: false, is_sensitive: false}, {name: customer_phone, type: varchar(20), is_primary: false, is_nullable: true, is_sensitive: true} ] }Agent启动时会加载这个JSON并缓存到内存。当Schema变更如DBA加了新表脚本会自动更新JSONAgent在下次请求时重新加载——无需重启服务。4.3 Query Planner Agent实现用OpenAI API手写调用Planner不是一个独立服务而是main.py里的一个函数import openai import json from datetime import datetime, timedelta def plan_query(user_question: str, schema_json: str) - dict: # 构建messages system_prompt f你是一个数据库查询规划器...此处为前述Prompt全文 # 当前日期用于填充默认time_range today datetime.now().date() first_day_of_month today.replace(day1) last_day_of_month (first_day_of_month timedelta(days32)).replace(day1) - timedelta(days1) user_prompt f用户问题{user_question} 数据库Schema{schema_json} 请严格按规则输出JSON。 response openai.ChatCompletion.create( modelgpt-3.5-turbo-1106, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt} ], temperature0.0, # 关键设为0禁用随机性 max_tokens500, response_format{type: json_object} # 强制JSON输出 ) try: plan json.loads(response.choices[0].message.content) # 后置校验确保所有必需字段存在 required_keys [time_range, dimensions, metrics, filters, limit] for key in required_keys: if key not in plan: raise ValueError(fMissing required key: {key}) return plan except json.JSONDecodeError as e: raise RuntimeError(fLLM output not valid JSON: {e}) except Exception as e: raise RuntimeError(fPlan validation failed: {e}) # 调用示例 schema json.load(open(/opt/agent/schema/sales_db_schema.json)) plan plan_query(华东区上月TOP5门店的复购率趋势, schema) print(plan) # 输出{time_range: [2024-08-01, 2024-08-31], dimensions: [store_name], metrics: [{name: repeat_rate, agg: avg}], filters: [{field: region, op: , value: East}], limit: 5}实操心得temperature0.0和response_format{type: json_object}是稳定性的双保险。前者禁用LLM的“创意发挥”后者让OpenAI API在底层强制做JSON格式校验。我测试过开启temperature0.3时同一问题多次调用limit字段有时是5有时是5字符串导致后续Python解析失败。设为0后100%输出整数5。4.4 SQL Generator Agent实现方言模板的精确填充Generator同样是一个函数接收Planner的plan和schema_jsonimport sqlparse from sqlparse.sql import IdentifierList, Identifier from sqlparse.tokens import Keyword, DML def generate_sql(plan: dict, schema_json: str, dialect: str mysql) - str: # 1. 解析schema构建表字段映射 schema json.loads(schema_json) # 2. 构建SELECT子句 select_parts [] for metric in plan[metrics]: agg_func metric[agg] field_name metric[name] # 根据dialect获取聚合函数写法 if agg_func median: # 用方言矩阵中的median_function模板 median_sql DIALECT_CAPABILITIES[dialect][median_function] # 填充{field}和{filters} filters_sql AND .join([f{f[field]} {f[op]} {f[value]} for f in plan[filters]]) select_parts.append(median_sql.format(fieldfield_name, filtersfilters_sql)) else: select_parts.append(f{agg_func.upper()}({field_name}) AS {field_name}_{agg_func}) # 3. 构建FROM子句需推断主表此处简化为products表 from_clause FROM products p JOIN sales s ON p.id s.product_id # 4. 构建WHERE子句 where_parts [] for f in plan[filters]: if f[op] LIKE: where_parts.append(f{f[field]} LIKE %{f[value]}%) else: where_parts.append(f{f[field]} {f[op]} {f[value]}) where_clause WHERE AND .join(where_parts) if where_parts else # 5. 构建ORDER BY和LIMIT order_by ORDER BY , .join([f{m[name]}_{m[agg]} for m in plan[metrics]]) DESC limit_clause DIALECT_CAPABILITIES[dialect][limit_clause].format(nplan[limit]) if plan[limit] else # 组装SQL raw_sql fSELECT {, .join(select_parts)} {from_clause} {where_clause} {order_by} {limit_clause} # 6. 用sqlparse格式化便于阅读和调试 formatted_sql sqlparse.format(raw_sql, reindentTrue, keyword_caseupper) # 7. 静态语法校验确保是SELECT语句 parsed sqlparse.parse(formatted_sql)[0] if not parsed.token_first().ttype is Keyword.DML and parsed.token_first().value.upper() ! SELECT: raise ValueError(Generated SQL is not a SELECT statement) return formatted_sql # 调用示例 sql generate_sql(plan, schema_json, dialectmysql) print(sql) # 输出 # SELECT AVG(repeat_rate) AS repeat_rate_avg # FROM products p JOIN sales s ON p.id s.product_id # WHERE region East # ORDER BY repeat_rate_avg DESC # LIMIT 5注意这个函数没有连接数据库它只生成SQL字符串。真正的执行交给独立的QueryExecutor服务。这种分离让测试变得极其简单——你可以用pytest传入各种plan字典断言生成的SQL是否符合预期而不用mock数据库连接。4.5 QueryExecutor服务沙箱执行的完整实现query_executor.py是一个Flask服务监听/execute端点from flask import Flask, request, jsonify import mysql.connector import hashlib import json import os app Flask(__name__) # 白名单存储已审核SQL的hash WHITELIST_FILE /opt/agent/whitelist.json def load_whitelist(): if os.path.exists(WHITELIST_FILE): with open(WHITELIST_FILE) as f: return set(json.load(f)) return set() WHITELIST load_whitelist() app.route(/execute, methods[POST]) def execute_query(): data request.get_json() query_hash data.get(query_hash) schema_version data.get(schema_version) # 1. 校验hash是否在白名单 if query_hash not in WHITELIST: return jsonify({error: SQL not approved}), 403 # 2. 加载对应schema版本的元数据用于字段校验 schema_path f/opt/agent/schema/{schema_version}.json if not os.path.exists(schema_path): return jsonify({error: Invalid schema version}), 400 schema json.load(open(schema_path)) # 3. 连接数据库只读账号 conn mysql.connector.connect( hostdb-prod.internal, userreadonly_agent, passwordos.getenv(DB_PASSWORD), databasesales_db ) cursor conn.cursor(dictionaryTrue) try: # 4. 执行SQL带超时 cursor.execute(SET SESSION MAX_EXECUTION_TIME3000) cursor.execute(data[sql]) # data[sql]由Agent生成已通过校验 # 5. 获取结果并脱敏 rows cursor.fetchall() result_df pd.DataFrame(rows) # 脱敏遍历schema替换敏感字段 for table in schema.values(): for col in table: if col.get(is_sensitive) and col[name] in result_df.columns: result_df[col[name]] *** # 6. 截断长文本统一数值精度 for col in result_df.columns: if result_df[col].dtype object: result_df[col] result_df[col].astype(str).str.slice(0, 255) elif pd.api.types.is_numeric_dtype(result_df[col]): result_df[col] result_df[col].round(4) return jsonify({ success: True, data: result_df.to_dict(orientrecords), columns: list(result_df.columns) }) except mysql.connector.Error as e: return jsonify({error: fMySQL error: {e.msg}}), 400 finally: cursor.close() conn.close() if __name__ __main__: app.run(host0.0.0.0:5001, port5001)提示白名单whitelist.json不是手动维护的。我写了一个audit_sql.py脚本每天扫描Agent生成的所有SQL记录在日志中用sqlparse解析AST人工审核后把安全SQL的hash批量加入白名单。这确保了所有执行的SQL都经过DBA确认符合公司SQL规范。4.6 Viz Designer与Renderer从Spec到可交互图表Designer的实现与Planner类似也是调用OpenAI APIdef design_viz(plan: dict, df: pd.DataFrame) - dict: # 构建messages包含plan JSON和df.head(3).to_dict() system_prompt 你是一个可视化规格生成器...前述Prompt user_prompt fPlanner输出{json.dumps(plan)} DataFrame前3行{df.head(3).to_dict(orientrecords)} response openai.ChatCompletion.create( modelgpt-3.5-turbo-1106, messages[{role: system, content: system_prompt}, {role: user, content: user_prompt}], temperature0.0, response_format{type: json_object} ) spec json.loads(response.choices[0].message.content) # 强制校验spec字段 for field in [chart_type, x_field, y_fields, title, x_label, y_label]: if field not in spec: raise ValueError(fVizSpec missing {field}) return spec # Renderer纯函数式渲染 def render_viz(spec: dict, df: pd.DataFrame) - str: if spec[chart_type] line: fig render_line(spec, df) elif spec[chart_type] bar: fig render_bar(spec, df) elif spec[chart_type] heatmap: fig render_heatmap(spec, df) elif spec[chart_type] pie: fig render_pie(spec, df) else: raise ValueError(fUnknown chart_type: {spec[chart_type]}) # 导出为HTML字符串含plotly.js html_str fig.to_html(include_plotlyjscdn, full_htmlFalse, config{displayModeBar: False}) return html_str # 主流程 plan plan_query(华东区上月TOP5门店的复购率趋势, schema_json) sql generate_sql(plan, schema_json, dialectmysql) # 调用QueryExecutor executor_response requests.post(http://localhost:5001/execute, json{ sql: sql, query_hash: hashlib.sha256(sql.encode()).hexdigest(), schema_version: sales_db_schema }) df pd.DataFrame(executor_response.json()[data]) spec design_viz(plan, df) html_chart render_viz(spec, df) # 返回给前端 return jsonify({chart_html: html_chart})实操心得render_viz返回的是纯HTML字符串前端直接innerHTML插入即可。不引入React/Vue等框架降低前端耦合度。我测试过一个含1000个数据点的折线图HTML字符串约120KBChrome加载毫无压力。5. 常见问题与排查技巧实录那些文档里