企业官网建设流程全解析

1. 市场主流 Data Agent 的技术栈全景拆解

Data Agent 这个词在 2026 年已经不算新鲜了，但真正动手搭过一套能跑通、能落地、能扛住业务压力的 Data Agent 系统的人，其实没想象中那么多。市面上大部分所谓的“调研报告”停留在功能罗列和厂商吹水层面，真正把技术栈和技术架构掰开揉碎讲清楚的少之又少。我过去一年多时间，先后参与了三个不同规模的 Data Agent 项目——一个面向内部数据分析场景，一个面向客户侧的智能问数产品，还有一个是给某垂直行业做的数据自动化处理 Agent。踩过的坑、选错过的组件、重构过的模块，加起来够写一本小册子了。

这篇文章不打算做成厂商对比表，而是从一线开发者的视角，把 Data Agent 的技术栈分层拆开，讲清楚每一层在干什么、为什么这么选、不同选型之间的真实差异在哪里。无论你是刚开始调研 Data Agent 的技术负责人，还是正在动手搭建原型的工程师，或者只是想搞清楚这个领域技术全貌的产品经理，都能从里面找到可以直接参考的东西。

1.1 Data Agent 到底在解决什么问题

先把概念对齐。Data Agent 不是简单的“ChatBI”或者“自然语言查数”换了个名字。它和传统 BI 工具、和单纯的 LLM 对话机器人有本质区别。传统 BI 的核心是“人找数据”——人先知道要看什么指标，然后去报表里找。ChatBI 进了一步，变成“人问数据”——用自然语言提问，系统转成 SQL 去查。但 Data Agent 的核心是“数据自己动”——它能根据目标自主规划步骤、调用工具、处理中间结果、在遇到异常时调整策略，最终完成一个需要多步操作的数据任务。

举个例子。传统 BI 里你想知道“上个月华东区销售额下降的原因”，你得自己去看销售额趋势、拆区域、拆品类、拆渠道，一步步排查。ChatBI 能帮你查出“上个月华东区销售额是 X”，但“为什么下降”它答不了。Data Agent 的做法是：先查华东区上个月销售额和环比，发现确实降了；然后自动拆解到品类维度，发现是 A 品类降得最狠；再拆到渠道，发现是线上渠道出了问题；接着去查线上渠道的流量和转化率数据，发现转化率没变但流量骤降；最后关联到营销活动表，发现上个月华东区的投放预算被砍了一半。整个过程 Agent 自己规划、自己执行、自己串联因果链，最后给你一份带分析结论的报告。

这就是 Data Agent 的核心价值：把数据消费从“被动查询”变成“主动探索”。它解决的不是“查数慢”的问题，而是“不知道该查什么”和“查了也看不出问题”的问题。

1.2 技术栈分层：从底层到交互层的完整视图

一套完整的 Data Agent 系统，技术栈可以分成六层。我从下往上说，每一层都讲清楚它的职责和主流选型。

第一层：数据基础设施层。这是最底层，负责数据的存储和计算。Data Agent 再智能，也得有数据可查。这一层的选型取决于你的数据规模和现有架构。小规模场景下，MySQL + ClickHouse 的组合很常见——MySQL 存元数据和业务表，ClickHouse 做 OLAP 查询。中大规模场景下，Hive/Spark 做离线数仓，Doris/StarRocks 做实时分析，Kafka 做流式数据管道。如果已经有数据湖，Iceberg/Hudi/Delta Lake 三选一，上面套 Trino 或 Spark SQL 做查询引擎。这一层的关键是：Data Agent 需要能通过统一的 SQL 接口访问所有数据源，所以要么所有数据都在一个查询引擎下，要么有一个联邦查询层（比如 Trino）来屏蔽底层差异。

第二层：语义层与元数据层。这是 Data Agent 区别于普通 SQL 生成器的关键。语义层负责把物理表结构映射成业务语义——比如t_order_amt这个字段，语义层里叫“订单金额”，有业务定义、有计算口径、有维度归属。元数据层则存储表之间的血缘关系、字段的枚举值、指标的统计周期等。没有这一层，LLM 生成的 SQL 就是盲猜，准确率惨不忍睹。主流做法是用 dbt 做语义建模，或者用 Cube.js、MetricFlow 这类指标平台来管理。更轻量的做法是自建一套 YAML 配置，把表、字段、指标、维度、同义词都定义清楚，作为 Prompt 的一部分注入给 LLM。

第三层：Agent 编排与推理层。这是整个系统的“大脑”。核心任务是：接收用户问题，规划执行步骤，调用工具，处理中间结果，决定下一步动作。技术选型上，LangChain/LangGraph 是目前最主流的编排框架，LlamaIndex 在 RAG 场景下更顺手，AutoGen 和 CrewAI 在多 Agent 协作场景下有优势。但说实话，如果你的场景不是特别复杂，直接用原生 Python 写一个状态机 + LLM 调用，可控性反而更强。这一层的关键组件包括：任务规划器（Planner）、工具调用器（Tool Executor）、记忆模块（Memory）、反思模块（Reflection）。规划器决定“先做什么后做什么”，工具调用器负责实际执行 SQL 查询或 API 调用，记忆模块保存上下文和历史交互，反思模块在结果不符合预期时触发重新规划。

第四层：工具与执行层。Agent 要干活，得有工具。最核心的工具是 SQL 执行器——把 LLM 生成的 SQL 发到数据库执行并返回结果。除此之外，常见的工具还包括：Python 代码执行器（做复杂计算或可视化）、API 调用器（查外部数据）、文件读写器（处理上传的 Excel/CSV）、图表生成器（把结果渲染成图）。这一层的工程重点是沙箱隔离和错误处理。SQL 执行要有超时和行数限制，Python 执行要在沙箱里跑，API 调用要有重试和降级策略。我见过太多 Demo 跑得飞起、一上生产就各种超时的案例，问题都出在这一层没做好。

第五层：交互与流式输出层。用户侧的感受全看这一层。2026 年做 Data Agent，SSE 流式输出已经是标配了。用户提问后，Agent 的思考过程、中间步骤、最终结果要实时渲染出来，而不是等全部算完再一次性返回。技术实现上，后端用 FastAPI 的StreamingResponse或者 Node.js 的EventSource推送 SSE 事件，前端用EventSourceAPI 接收并逐块渲染。关键细节是：要支持 Abort 中断。用户看到 Agent 跑偏了，得能随时掐断，不然浪费 token 也浪费时间。实现上就是在后端维护一个任务 ID 到执行状态的映射，收到 Abort 请求时取消对应的异步任务。

第六层：可观测与评估层。这一层最容易被忽略，但生产环境离不开。你需要记录每一次 Agent 调用的完整链路：用户问了什么、Agent 规划了哪些步骤、每步生成了什么 SQL、执行结果是什么、最终回答是什么、耗时多少、消耗了多少 token。这些数据用来做两件事：一是排查问题，二是持续优化 Prompt 和工具。主流方案是 LangSmith 或 LangFuse 做 LLM 可观测，配合 OpenTelemetry 做全链路追踪。评估方面，需要建一套测试集，覆盖常见问题类型，定期跑回归测试，看准确率和召回率的变化。

1.3 不同规模场景下的选型差异

技术栈不是越先进越好，得看场景。我按规模分三类来说。

小规模场景（数据量 < 100GB，日活用户 < 50）。这种场景下，别搞太复杂。数据层用 PostgreSQL 或 MySQL 就够了，分析查询用 ClickHouse 单机版。语义层用 YAML 文件管理，几十张表、百来个字段，手写配置完全可行。Agent 编排直接用 Python + OpenAI SDK，不需要 LangChain 那套抽象。工具层就一个 SQL 执行器加一个 Python 执行器。交互层用 FastAPI + SSE，前端用 React 或 Vue 都行。可观测用 LangFuse 开源自部署。整套下来，一台 8C16G 的机器就能跑，成本可控，维护简单。

中规模场景（数据量 100GB-10TB，日活用户 50-500）。这时候需要认真考虑架构了。数据层用 Doris 或 StarRocks 做实时数仓，Hive/Spark 做离线补充。语义层用 dbt 管理模型，配合 Cube.js 做指标层。Agent 编排用 LangGraph，因为需要支持多轮反思和条件分支。工具层要加沙箱隔离，SQL 执行走连接池，Python 执行用 Docker 容器。交互层要支持多会话管理和历史记录。可观测用 LangSmith 商业版或自建 ELK + 自定义埋点。这一层的重点是稳定性和并发能力，需要做压力测试，确保 50 个用户同时提问时不会雪崩。

大规模场景（数据量 > 10TB，日活用户 > 500）。这种规模通常是企业级平台了。数据层是湖仓一体架构，Iceberg + Spark + Trino 或者 Hudi + Flink + Doris。语义层需要独立的元数据服务，支持版本管理和权限控制。Agent 编排要支持多 Agent 协作——一个主 Agent 做规划，多个子 Agent 分别负责 SQL 生成、代码执行、结果校验。工具层需要独立的执行服务，支持水平扩展。交互层要支持 WebSocket 和 SSE 双通道，还要有消息队列做削峰。可观测层需要全链路追踪 + 实时告警 + 自动评估。这一层的核心挑战是一致性和延迟——数据更新后语义层要同步，Agent 推理链路长了延迟会飙升，需要在架构上做取舍。

2. 核心技术架构的深度解析

上一章把技术栈分层讲了一遍，这一章深入几个关键架构决策点。这些决策点直接决定了系统的上限和下限，选错了后面很难补。

2.1 Agent 编排架构：ReAct 还是 Plan-and-Execute

Agent 编排有两种主流范式。一种是 ReAct（Reasoning + Acting），Agent 每一步都先推理再行动，根据上一步的结果决定下一步。另一种是 Plan-and-Execute，Agent 先制定完整计划，然后按计划逐步执行，执行过程中可以调整但主体计划不变。

ReAct 的优点是灵活，适合探索性任务。比如“帮我分析一下为什么这个月利润下降了”，Agent 需要一步步查数据、看结果、调整方向，ReAct 很合适。缺点是容易跑偏，而且每一步都要调 LLM，token 消耗大，延迟高。

Plan-and-Execute 的优点是效率高，计划制定一次，后续执行可以并行或串行快速推进。适合流程明确的任务，比如“生成月度经营报告”，步骤基本固定：查销售数据、查成本数据、算利润、生成图表、写总结。缺点是遇到意外情况时调整不够灵活。

实际项目中，我倾向于混合模式：先用 Plan-and-Execute 制定高层计划，每个步骤内部用 ReAct 做细粒度执行。LangGraph 的 StateGraph 很适合实现这种模式——把计划步骤定义为节点，节点内部用 ReAct 循环。这样既有全局规划的效率，又有局部执行的灵活性。

架构实现上，核心是一个状态机。状态包括：当前计划、已完成步骤、当前步骤、中间结果、上下文记忆。每个节点执行完后更新状态，然后根据条件决定下一个节点。条件判断可以用规则引擎，也可以用 LLM 做路由。我建议关键路由用规则，非关键路由用 LLM，这样可控性和灵活性兼顾。

2.2 语义层的构建方式：YAML 配置 vs 指标平台

语义层是 Data Agent 准确率的命门。我见过太多项目，LLM 能力很强，但生成的 SQL 就是不对，问题全出在语义层没做好。

最轻量的做法是 YAML 配置。每张表一个 YAML 文件，定义表名、描述、字段列表、字段类型、字段描述、枚举值、同义词。指标单独定义，包括指标名、计算逻辑、依赖字段、统计周期。这些配置在 Agent 启动时加载到内存，生成 Prompt 时动态注入相关表的配置。优点是简单直接，改起来快。缺点是表多了之后管理麻烦，没有版本控制，容易冲突。

更工程化的做法是用指标平台，比如 Cube.js 或 MetricFlow。指标平台的核心价值是统一定义、统一计算。所有指标在一个地方定义，Agent 生成的 SQL 实际上是查询指标平台，由指标平台翻译成底层 SQL。这样做的好处是：口径统一，不会出现同一个指标在不同报表里算法不一样的情况；权限可控，指标平台可以控制哪些人能查哪些指标；性能优化，指标平台可以做预聚合和缓存。

我的建议是：如果团队有数据工程能力，直接上指标平台。前期投入大一点，但后期维护成本低很多。如果团队小、表少，YAML 配置也能用，但要建立严格的 Review 机制，每次改配置都要有人审核。

2.3 流式输出的工程实现细节

SSE 流式输出听起来简单，做起来坑不少。我详细说一下实现要点。

后端用 FastAPI 的话，核心是StreamingResponse。Agent 执行过程中，每完成一个步骤就 yield 一个 SSE 事件。事件格式一般是：

event: step data: {"step": "sql_generation", "content": "SELECT ...", "status": "running"} event: step data: {"step": "sql_execution", "content": "查询完成，返回 156 行", "status": "done"} event: answer data: {"content": "根据查询结果，华东区上个月销售额..."}

前端用EventSource接收，根据 event 类型分别处理。step类型的事件用来渲染执行过程，answer类型的事件用来渲染最终回答。

关键细节一：心跳机制。SSE 连接如果长时间没有数据推送，可能会被中间层断开。需要每隔 15-30 秒发一个注释行（: heartbeat）保持连接。

关键细节二：Abort 中断。后端需要维护一个task_id -> asyncio.Task的映射。收到 Abort 请求时，调用task.cancel()取消任务，同时推送一个aborted事件给前端。注意要处理CancelledError异常，确保资源正确释放。

关键细节三：错误处理。Agent 执行过程中任何一步出错，都要推送error事件，并且要包含足够的上下文信息，方便前端展示和用户理解。不要直接把 Python 异常栈推给前端，要包装成用户能看懂的错误信息。

关键细节四：多轮对话的上下文管理。SSE 连接通常是一次性的，但用户可能连续提问。需要在后端维护会话状态，每次新问题进来时，从会话历史中加载上下文。上下文太长时要做摘要压缩，不然 token 会爆。

2.4 工具调用的安全与隔离

Agent 能执行 SQL 和 Python 代码，这是能力也是风险。安全隔离必须做好。

SQL 执行的安全措施：只允许 SELECT 语句，禁止 INSERT/UPDATE/DELETE/DROP；查询必须有 LIMIT，防止全表扫描；设置查询超时，比如 30 秒；用只读账号连接数据库；对生成的 SQL 做语法校验，防止 SQL 注入。

Python 执行的安全措施：在 Docker 容器里执行，容器无网络访问权限；限制 CPU 和内存使用；设置执行超时；禁止文件系统写操作（除了临时目录）；用 seccomp 或 AppArmor 做系统调用过滤。

我踩过的一个坑：早期版本没做 SQL 超时，有一次 Agent 生成了一个笛卡尔积查询，直接把数据库 CPU 打满了。后来加了超时和行数限制，再也没出过问题。还有一个坑是 Python 执行没隔离，Agent 生成的代码里有个死循环，把整个服务卡死了。后来改成 Docker 沙箱执行，问题解决。

3. 实操搭建：从零到一跑通一个 Data Agent

这一章我手把手带你搭一个最小可用的 Data Agent。技术栈选最简的：FastAPI + LangChain + SQLite + SSE。虽然简单，但核心链路是完整的，你可以基于这个骨架替换成生产级组件。

3.1 环境准备与项目结构

先建项目目录：

mkdir># db/models.py from sqlalchemy import create_engine, Column, Integer, String, Float, Date from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() engine = create_engine("sqlite:///./demo.db") SessionLocal = sessionmaker(bind=engine) class Sales(Base): __tablename__ = "sales" id = Column(Integer, primary_key=True) date = Column(Date) store_id = Column(Integer) product = Column(String) amount = Column(Float) quantity = Column(Integer) class Store(Base): __tablename__ = "store" id = Column(Integer, primary_key=True) name = Column(String) region = Column(String) city = Column(String)

语义层配置：

# semantic/schema.yaml tables: - name: sales description: 销售记录表，记录每笔销售交易的详细信息 columns: - name: date type: date description: 销售日期 - name: store_id type: integer description: 门店ID，关联 store 表 - name: product type: string description: 产品名称 - name: amount type: float description: 销售金额，单位元 - name: quantity type: integer description: 销售数量 - name: store description: 门店信息表 columns: - name: id type: integer description: 门店ID - name: name type: string description: 门店名称 - name: region type: string description: 所属大区，枚举值：华东、华南、华北、西南 - name: city type: string description: 所属城市 metrics: - name: 销售额 expression: SUM(sales.amount) description: 所有销售记录的金额总和 - name: 销售数量 expression: SUM(sales.quantity) description: 所有销售记录的数量总和

3.2 Agent 核心逻辑实现

Planner 负责把用户问题转成执行计划。我用一个简单的两阶段方式：先让 LLM 判断问题类型（简单查询还是复杂分析），然后生成对应的 SQL。

# agent/prompts.py SQL_GENERATION_PROMPT = """你是一个 SQL 生成助手。根据下面的表结构和用户问题，生成一条 SQLite 兼容的 SQL 查询语句。 表结构： {schema} 用户问题：{question} 要求： 1. 只生成 SELECT 语句 2. 必须加 LIMIT 1000 3. 只返回 SQL，不要有其他内容 """

# agent/planner.py import yaml from langchain_openai import ChatOpenAI from agent.prompts import SQL_GENERATION_PROMPT class Planner: def __init__(self, schema_path: str): with open(schema_path) as f: self.schema = yaml.safe_load(f) self.llm = ChatOpenAI(model="gpt-4o", temperature=0) def _format_schema(self) -> str: lines = [] for table in self.schema["tables"]: lines.append(f"表名：{table['name']} - {table['description']}") for col in table["columns"]: lines.append(f" - {col['name']} ({col['type']}): {col['description']}") for metric in self.schema.get("metrics", []): lines.append(f"指标：{metric['name']} = {metric['expression']} - {metric['description']}") return "\n".join(lines) def plan(self, question: str) -> str: prompt = SQL_GENERATION_PROMPT.format( schema=self._format_schema(), question=question ) sql = self.llm.invoke(prompt).content.strip() # 清理 markdown 代码块标记 sql = sql.replace("```sql", "").replace("```", "").strip() return sql

Executor 负责执行 SQL 并返回结果：

# agent/executor.py from sqlalchemy import text from db.models import SessionLocal class Executor: def __init__(self): self.db = SessionLocal() def execute_sql(self, sql: str) -> dict: # 安全检查 sql_upper = sql.upper().strip() if not sql_upper.startswith("SELECT"): return {"error": "只允许 SELECT 查询"} if "LIMIT" not in sql_upper: sql = sql.rstrip(";") + " LIMIT 1000" try: result = self.db.execute(text(sql)) columns = list(result.keys()) rows = [dict(zip(columns, row)) for row in result.fetchall()] return {"columns": columns, "rows": rows, "count": len(rows)} except Exception as e: return {"error": str(e)}

3.3 SSE 流式接口与前端渲染

FastAPI 的 SSE 接口：

# main.py import json import asyncio from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse from fastapi.staticfiles import StaticFiles from agent.planner import Planner from agent.executor import Executor app = FastAPI() app.mount("/static", StaticFiles(directory="static"), name="static") planner = Planner("semantic/schema.yaml") executor = Executor() # 任务管理，用于 Abort tasks = {} async def event_generator(question: str, task_id: str): try: # 步骤1：生成 SQL yield f"event: step\ndata: {json.dumps({'step': 'sql_generation', 'status': 'running', 'content': '正在生成查询语句...'})}\n\n" await asyncio.sleep(0.1) sql = planner.plan(question) yield f"event: step\ndata: {json.dumps({'step': 'sql_generation', 'status': 'done', 'content': sql})}\n\n" # 步骤2：执行 SQL yield f"event: step\ndata: {json.dumps({'step': 'sql_execution', 'status': 'running', 'content': '正在执行查询...'})}\n\n" await asyncio.sleep(0.1) result = executor.execute_sql(sql) if "error" in result: yield f"event: error\ndata: {json.dumps({'content': result['error']})}\n\n" return yield f"event: step\ndata: {json.dumps({'step': 'sql_execution', 'status': 'done', 'content': f'查询完成，返回 {result[\"count\"]} 行'})}\n\n" # 步骤3：生成回答 yield f"event: step\ndata: {json.dumps({'step': 'answer_generation', 'status': 'running', 'content': '正在整理结果...'})}\n\n" await asyncio.sleep(0.1) answer = f"查询到 {result['count']} 条记录。" if result["rows"]: answer += f" 第一条记录：{json.dumps(result['rows'][0], ensure_ascii=False)}" yield f"event: answer\ndata: {json.dumps({'content': answer})}\n\n" yield f"event: done\ndata: {json.dumps({'status': 'completed'})}\n\n" except asyncio.CancelledError: yield f"event: aborted\ndata: {json.dumps({'content': '任务已中断'})}\n\n" finally: tasks.pop(task_id, None) @app.get("/api/ask") async def ask(question: str, task_id: str): async def stream(): async for event in event_generator(question, task_id): yield event return StreamingResponse(stream(), media_type="text/event-stream") @app.post("/api/abort/{task_id}") async def abort(task_id: str): task = tasks.get(task_id) if task: task.cancel() return {"status": "aborted"} return {"status": "not_found"}

前端页面：

<!DOCTYPE html> <html> <head> <meta charset="UTF-8"> <title>Data Agent Demo</title> <style> body { font-family: sans-serif; max-width: 800px; margin: 40px auto; padding: 0 20px; } #steps { background: #f5f5f5; padding: 16px; border-radius: 8px; min-height: 100px; margin: 16px 0; } .step { margin: 8px 0; padding: 8px; border-left: 3px solid #ddd; } .step.running { border-color: #f0ad4e; } .step.done { border-color: #5cb85c; } #answer { background: #e8f4fd; padding: 16px; border-radius: 8px; margin: 16px 0; } input { width: 70%; padding: 8px; } button { padding: 8px 16px; margin-left: 8px; } </style> </head> <body> <h2>Data Agent 演示</h2> <div> <input id="question" placeholder="输入你的问题，例如：华东区销售额是多少？" /> <button onclick="ask()">提问</button> <button onclick="abort()">中断</button> </div> <div id="steps"></div> <div id="answer"></div> <script> let eventSource = null; let currentTaskId = null; function ask() { const question = document.getElementById('question').value; if (!question) return; document.getElementById('steps').innerHTML = ''; document.getElementById('answer').innerHTML = ''; currentTaskId = 'task_' + Date.now(); eventSource = new EventSource(`/api/ask?question=${encodeURIComponent(question)}&task_id=${currentTaskId}`); eventSource.addEventListener('step', (e) => { const data = JSON.parse(e.data); const div = document.createElement('div'); div.className = `step ${data.status}`; div.textContent = `[${data.step}] ${data.content}`; document.getElementById('steps').appendChild(div); }); eventSource.addEventListener('answer', (e) => { const data = JSON.parse(e.data); document.getElementById('answer').textContent = data.content; }); eventSource.addEventListener('error', (e) => { const data = JSON.parse(e.data); document.getElementById('answer').textContent = '错误：' + data.content; eventSource.close(); }); eventSource.addEventListener('done', () => { eventSource.close(); }); eventSource.addEventListener('aborted', () => { document.getElementById('answer').textContent = '任务已中断'; eventSource.close(); }); } function abort() { if (currentTaskId) { fetch(`/api/abort/${currentTaskId}`, { method: 'POST' }); } } </script> </body> </html>

3.4 测试数据与运行验证

生成测试数据：

# db/seed.py import random from datetime import date, timedelta from db.models import Base, engine, SessionLocal, Sales, Store Base.metadata.create_all(engine) db = SessionLocal() # 门店数据 stores = [ Store(id=1, name="上海旗舰店", region="华东", city="上海"), Store(id=2, name="杭州西湖店", region="华东", city="杭州"), Store(id=3, name="北京朝阳店", region="华北", city="北京"), Store(id=4, name="广州天河店", region="华南", city="广州"), Store(id=5, name="成都春熙路店", region="西南", city="成都"), ] db.add_all(stores) # 销售数据 products = ["手机", "笔记本", "平板", "耳机", "充电器"] for i in range(500): sale = Sales( date=date.today() - timedelta(days=random.randint(0, 90)), store_id=random.randint(1, 5), product=random.choice(products), amount=round(random.uniform(100, 10000), 2), quantity=random.randint(1, 10) ) db.add(sale) db.commit() print("测试数据生成完成")

运行：

python db/seed.py uvicorn main:app --reload --port 8000

打开http://localhost:8000/static/index.html，输入“华东区销售额是多少”，就能看到 Agent 一步步执行并流式返回结果。

4. 常见问题与排查技巧实录

这一章整理我在实际项目中遇到的高频问题和解决方法。这些问题在 Demo 阶段通常不会暴露，一上生产就全出来了。

4.1 SQL 生成准确率低的排查思路

问题表现：Agent 生成的 SQL 语法正确但逻辑错误，比如该用 SUM 的地方用了 COUNT，该 JOIN 的地方没 JOIN，或者 WHERE 条件写反了。

排查步骤：

第一步，检查语义层配置是否完整。字段描述是否清晰？枚举值是否列全？指标定义是否准确？我遇到过因为字段描述写得太模糊（比如“金额”没说是含税还是不含税），导致 LLM 理解偏差的情况。

第二步，检查 Prompt 里注入的 schema 是否相关。如果表很多，全量注入会干扰 LLM。应该根据用户问题做相关性检索，只注入相关的表和字段。可以用 embedding 做相似度匹配，也可以用简单的关键词匹配。

第三步，检查 Few-shot 示例是否足够。在 Prompt 里加 3-5 个高质量的问答对，能显著提升准确率。示例要覆盖常见查询模式：聚合查询、多表 JOIN、时间范围过滤、分组统计。

第四步，考虑用更结构化的方式约束输出。比如让 LLM 输出 JSON 格式的查询计划，而不是直接输出 SQL。查询计划包含：目标表、筛选条件、聚合方式、分组维度、排序方式。然后由代码根据查询计划生成 SQL。这样做的好处是可控性强，LLM 只需要做语义理解，不需要懂 SQL 语法。

避坑技巧：建一个“错误案例库”，每次发现生成错误的 SQL，就把问题、错误 SQL、正确 SQL 记录下来。定期用这些案例做 Few-shot 示例的更新，或者做微调数据。我维护了一个 200 多条的错误案例库，准确率从最初的 60% 提升到了 92%。

4.2 流式输出中断与超时的处理

问题表现：用户提问后，SSE 连接建立成功，但执行到一半突然断开，前端收不到后续事件。

排查步骤：

先看后端日志，确认是 Agent 执行出错还是连接被断开。如果是 Agent 执行出错，检查是否有未捕获的异常。如果是连接断开，检查 Nginx 或负载均衡器的超时配置。SSE 连接默认超时通常是 60 秒，如果 Agent 执行时间超过这个值，连接会被断开。

解决方案：在 Nginx 配置里把proxy_read_timeout调到 300 秒以上。同时后端要加心跳，每隔 15 秒发一个注释行，保持连接活跃。

另一个常见问题是 Abort 不生效。用户点了中断，但后端任务还在跑。检查task.cancel()是否真的取消了异步任务。如果 Agent 内部有同步阻塞操作（比如同步的数据库查询），cancel()不会立即生效。需要把阻塞操作改成异步，或者在关键步骤之间检查取消标志。

避坑技巧：给每个 Agent 任务设置最大执行时间，比如 120 秒。超时后自动取消并返回超时错误。这样即使 Abort 没生效，也不会无限期占用资源。

4.3 多轮对话上下文丢失的解决

问题表现：用户问“华东区销售额是多少”，Agent 回答了。接着问“那华南区呢”，Agent 不知道“那”指的是什么，重新生成了一个不相关的查询。

排查步骤：

检查会话管理逻辑。SSE 是无状态的，每次请求都是独立的。需要在后端维护会话状态，把历史问答存起来。下次请求进来时，把历史上下文注入 Prompt。

解决方案：用 Redis 或内存字典存储会话历史，key 是 session_id，value 是消息列表。每次新问题进来，把最近 N 轮对话拼接到 Prompt 里。N 一般取 3-5，太多会占用 token，太少会丢失上下文。

更优雅的做法是做指代消解。在生成 SQL 之前，先用 LLM 把用户问题改写成完整的问题。比如“那华南区呢”改写成“华南区销售额是多少”。这样后续处理就不依赖上下文了。

避坑技巧：上下文不是越长越好。我试过把 10 轮对话都塞进去，结果 LLM 被无关信息干扰，准确率反而下降了。后来改成只保留最近 3 轮，加上一个会话摘要（用 LLM 对更早的对话做摘要），效果最好。

4.4 常见问题速查表

4.5 生产环境部署的注意事项

最后说几个生产部署的坑。

资源隔离。Agent 执行 SQL 和 Python 代码会消耗大量 CPU 和内存。不要让 Agent 服务和数据库、其他业务服务混部。至少要用独立的容器或虚拟机。

限流降级。给每个用户设置每日提问次数上限，防止滥用。当 LLM API 不可用时，要有降级策略——比如返回缓存结果，或者提示用户稍后再试。

数据安全。Agent 能访问数据库，必须做好权限控制。用只读账号，限制可访问的表和字段。敏感字段要做脱敏处理，比如手机号、身份证号在返回结果前要掩码。

成本控制。LLM API 调用是主要成本。优化 Prompt 长度、缓存常见问题的结果、用更便宜的模型处理简单问题，都能显著降低成本。我算过一笔账，优化前每次查询平均消耗 3000 token，优化后降到 1200 token，成本降了 60%。

版本管理。语义层配置、Prompt 模板、Agent 逻辑都要做版本管理。每次变更都要记录，出问题时能快速回滚。我建议用 Git 管理所有配置和代码，每次上线打 tag。

监控告警。关键指标要监控：请求量、成功率、平均延迟、Token 消耗、错误率。设置告警阈值，比如错误率超过 5% 就告警。我用的方案是 Prometheus + Grafana，配合自定义埋点，效果不错。

这套东西搭下来，一个能用的 Data Agent 系统就成型了。从 Demo 到生产还有距离，但核心链路和关键决策点都在这里了。后面就是根据具体业务场景做调优和扩展，比如加更多工具、支持更多数据源、做多 Agent 协作。方向对了，剩下的就是迭代。