智能对话引擎实战 —— 多轮上下文管理与智能图表推荐 · devcfg

🎯 本章核心问题

如何构建一个真正”智能”的对话系统？

普通聊天机器人的问题：

• ❌ 无记忆：每轮对话都是独立的，无法理解”上面”、”之前”等指代
• ❌ 输出单一：只返回文字，没有可视化
• ❌ 无法总结：聊了 10 轮后无法生成整体报告

我们的解决方案：

1. ✅ 多轮上下文管理：自动维护最近 10 轮对话历史
2. ✅ 智能图表推荐：根据 SQL 特征自动选择最佳可视化方式
3. ✅ 一键报告生成：聚合多次查询结果，输出专业分析报告

---

📐 架构总览

智能对话引擎架构

核心模块一览

上下文、LLM、SQL、图表推荐与报告生成

---

💬 一、多轮上下文管理（核心亮点）

1.1 为什么需要上下文？

场景对比：

对话类型	无上下文（单轮）	有上下文（多轮）
用户输入	“按地区细分”	“上面的数据按地区再细分一下”
系统理解	❓ “什么数据？”	✅ 知道是”订单总额数据”
生成的 SQL	SELECT region, SUM(amount) FROM orders GROUP BY region	✅ 同上 + 时间过滤继承自上一轮

1.2 会话生命周期管理

app/services/chat_engine.py

async def create_session(db: AsyncSession, data: ChatSessionCreate) -> ChatSession:
    """创建新会话"""
    session = ChatSession(
        datasource_id=data.datasource_id,
        title=data.title or "New Chat",
    )
    db.add(session)
    await db.flush()
    await db.refresh(session)
    return session


async def get_session_with_messages(db: AsyncSession, session_id: int):
    """加载会话及其所有消息"""
    result = await db.execute(
        select(ChatSession).where(ChatSession.id == session_id)
    )
    session = result.scalar_one_or_none()

    if session:
        # 加载该会话的所有消息（按时间排序）
        msg_result = await db.execute(
            select(ChatMessage)
            .where(ChatMessage.session_id == session_id)
            .order_by(ChatMessage.created_at)
        )
        session.messages = list(msg_result.scalars().all())

    return session


async def delete_session(db: AsyncSession, session_id: int) -> bool:
    """删除会话及所有关联消息"""
    from sqlalchemy import delete

    # 先删消息（外键约束）
    await db.execute(
        delete(ChatMessage).where(ChatMessage.session_id == session_id)
    )
    # 再删会话
    result = await db.execute(
        delete(ChatSession).where(ChatSession.id == session_id)
    )
    return result.rowcount > 0

💡 数据模型设计

# models/chat.py

class ChatSession(Base):
    __tablename__ = "chat_sessions"

    id = Column(Integer, primary_key=True, autoincrement=True)
    datasource_id = Column(Integer, ForeignKey("datasources.id"))
    title = Column(String(200), default="New Chat")
    created_at = Column(DateTime, default=datetime.utcnow)

class ChatMessage(Base):
    __tablename__ = "chat_messages"

    id = Column(Integer, primary_key=True, autoincrement=True)
    session_id = Column(Integer, ForeignKey("chat_sessions.id"))
    role = Column(String(20))  # "user" or "assistant"
    content = Column(Text)  # 自然语言回复

    # SQL 相关字段（仅 assistant 消息有值）
    sql_query = Column(Text, nullable=True)  # 提取的 SQL
    query_result = Column(Text, nullable=True)  # JSON 格式的结果
    chart_config = Column(Text, nullable=True)  # 图表配置 JSON

    created_at = Column(DateTime, default=datetime.utcnow)

1.3 上下文窗口控制策略

app/services/chat_engine.py

async def chat(db: AsyncSession, session_id: int, user_message: str) -> ChatMessage:
    # Step 1: 加载最近的消息历史（限制为最近 10 轮）
    history_result = await db.execute(
        select(ChatMessage)
        .where(ChatMessage.session_id == session_id)
        .order_by(ChatMessage.created_at)
    )
    history = list(history_result.scalars().all())

    # ⚡ 关键：只取最近 10 条消息（约 5 轮对话）
    recent_history = history[-10:]

    # Step 2: 构建对话上下文数组
    conversation = []
    for msg in recent_history:
        conversation.append({
            "role": msg.role,
            "content": msg.content
        })

    # 追加当前用户消息
    conversation.append({"role": "user", "content": user_message})

🎯 为什么选择 10 条而不是全部？

策略	Token 消耗	上下文完整性	适用场景
全量历史	⚠️ 可能超限（>80K tokens）	✅ 完整	短对话（<20轮）
最近 N 条（当前方案）	✅ 可控（~5K tokens）	✅ 够用	长对话（50+轮）
摘要压缩	✅ 极低	❌ 可能丢失细节	超长对话（100+轮）

我们的选择：history[-10:] 平衡了成本和效果。

💡 进阶优化：滑动窗口 + 摘要

def build_smart_context(history, max_tokens=6000):
    """
    智能上下文组装策略：
    - 始终保留前 2 轮（建立初始上下文）
    - 始终保留最近 8 轮（保持连贯性）
    - 中间的用摘要替代（如果太长）
    """
    if len(history) <= 10:
        return history[-10:]

    # 保留首尾，中间摘要
    head = history[:4]  # 前 2 轮
    tail = history[-6:]  # 最近 3 轮
    middle_summary = summarize_messages(history[4:-6])

    return head + [{"role": "system", "content": f"...(中间{len(history)-10}轮已摘要): {middle_summary}"}] + tail

---

🤖 二、LLM 对话集成与 Prompt 构建

2.1 System Prompt 设计

app/services/chat_engine.py

async def chat(db: AsyncSession, session_id: int, user_message: str):
    # ... (上下文准备)

    # 获取语义模型
    semantic_model = await get_semantic_model(db, session.datasource_id)

    # 构建 System Prompt
    system_prompt = f"""你是一个专业的数据分析助手。你帮助用户从数据库中查询和分析数据。
你可以访问以下数据库结构：

{json.dumps(semantic_model, ensure_ascii=False, indent=2)}

当用户提出数据相关问题时，请按以下格式回复：
1. 简要说明你将要查询什么数据
2. 用 ```sql ``` 代码块包裹生成的 SQL 查询语句
3. 尽量提供有价值的分析和洞察

如果用户的问题与数据无关，请正常回复。"""

    # 组装完整的对话历史
    conversation_text = ""
    for msg in conversation:
        role = msg["role"]
        content = msg["content"]
        conversation_text += f"\n{role}: {content}"

    # 调用 LLM Gateway
    gateway = LLMGateway()
    assistant_content = await gateway.chat_completion([
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": conversation_text.strip()},
    ])

💡 System Prompt 的三层结构

角色定义、语义模型注入、输出格式约束

2.2 实际对话案例演示

Round 1：

👤 用户：查询最近7天的订单总额

🤖 助手：
我将为您查询最近7天（2026-05-15 至 2026-05-22）的订单总额。

```sql
SELECT DATE(created_at) as date,
       SUM(total_amount) as daily_total
FROM orders
WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
GROUP BY DATE(created_at)
ORDER BY date DESC;

📊 分析发现：

• 近7天订单总额为 ¥1,234,567
• 5月18日达到峰值（¥234,567），可能是促销活动导致
• 整体呈上升趋势，日均增长 12%

  **Round 2**（利用上下文）：

  👤 用户：上面的数据按地区再细分一下

🤖 助手：（理解”上面”= 订单数据）
好的，我将按地区细分近7天的订单销售额。

SELECT r.name as region,
       SUM(o.total_amount) as total_sales
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN regions r ON u.region_id = r.id
WHERE o.created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
GROUP BY r.id, r.name
ORDER BY total_sales DESC;

📊 地区排名：

1. 华东区：¥456,789（37%）
2. 南区：¥345,678（28%）
3. 北区：¥234,567（19%）
   …

   **Round 3**（生成报告）：

   👤 用户：帮我生成一份分析报告

🤖 助手：
正在基于我们刚才的对话数据生成专业分析报告…

📊 销售数据分析报告（2026-05-15 ~ 2026-05-22）

一、关键发现

✅ 本周销售总额突破 120 万，环比增长 15%
✅ 华东区贡献最大，占总量 37%
…

（详见下文”报告生成”章节）

---

## 🔍 三、SQL 提取与执行流程

### 3.1 从 LLM 响应中提取 SQL

`app/services/chat_engine.py`

```python
import re

async def chat(...):
    # ... (LLM 调用)

    sql_query = None
    query_result = None
    chart_config = None

    # 正则提取 SQL 代码块
    sql_match = re.search(r'```sql\s*(.*?)```', assistant_content, re.DOTALL)
    if sql_match:
        sql_query = sql_match.group(1).strip()

        # 安全校验
        if validate_sql(sql_query):
            try:
                # 执行查询
                query_result = json.dumps(
                    await execute_query(sql_query, session.datasource_id),
                    ensure_ascii=False,
                    default=str,  # 处理 datetime 等特殊类型
                )

                # 智能图表推荐
                chart_config = _suggest_chart(
                    sql_query,
                    json.loads(query_result)
                )
            except Exception as e:
                query_result = json.dumps({"error": str(e)})

3.2 错误处理策略

场景	处理方式	用户体验
响应中无 SQL	纯文本回复（如闲聊、解释性问题）	显示文字即可
SQL 验证失败	不执行，记录 WARNING 日志	提示”生成的SQL存在风险”
SQL 执行异常	捕获异常，返回错误信息	显示”查询出错：表不存在”
查询结果为空	正常返回空列表	显示”暂无数据”

---

📊 四、智能图表推荐算法（_suggest_chart）

4.1 设计思路

核心原则：根据 SQL 结构特征 和 返回数据的统计特征 自动选择最合适的图表类型。

4.2 算法实现

app/services/chat_engine.py

def _suggest_chart(sql: str, data: list[dict]) -> Optional[dict]:
    """
    智能图表推荐算法

    Args:
        sql: 生成的 SQL 语句
        data: 查询结果 [{col: val}, ...]

    Returns:
        图表配置字典，或 None（无法推荐时）
    """
    if not data or len(data) == 0:
        return None

    sql_lower = sql.lower()

    # ===== 特征提取 =====

    # 特征 1：是否有时间维度
    has_date = any(kw in sql_lower for kw in [
        "date", "time", "month", "year", "week",
        "day", "created_at", "updated_at"
    ])

    # 特征 2：是否有分组聚合
    has_group_by = "group by" in sql_lower

    # 特征 3：识别列的数据类型
    columns = list(data[0].keys())
    numeric_cols = []  # 数值型列（度量）
    string_cols = []  # 字符串型列（维度）

    for col in columns:
        values = [row.get(col) for row in data if row.get(col) is not None]
        if values and all(isinstance(v, (int, float)) for v in values):
            numeric_cols.append(col)
        else:
            string_cols.append(col)

    # ===== 规则匹配 =====

    # 规则 1：时间趋势 → 折线图
    if has_date and numeric_cols:
        return {
            "type": "line_chart",
            "x_field": string_cols[0] if string_cols else columns[0],
            "y_fields": numeric_cols,
        }

    # 规则 2：分组聚合 + 少类别 → 饼图（适合占比展示）
    elif has_group_by and numeric_cols:
        if len(data) <= 8:  # 类别数 <= 8 时用饼图
            return {
                "type": "pie_chart",
                "label_field": string_cols[0] if string_cols else columns[0],
                "value_field": numeric_cols[0],
            }

        # 规则 3：分组聚合 + 多类别 → 柱状图（适合对比）
        else:
            return {
                "type": "bar_chart",
                "x_field": string_cols[0] if string_cols else columns[0],
                "y_fields": numeric_cols,
            }

    # 规则 4：单行单数值 → 指标卡（KPI 展示）
    elif numeric_cols and len(data) == 1:
        return {
            "type": "metric_card",
            "metrics": {col: data[0].get(col) for col in numeric_cols},
        }

    # 兜底：表格展示
    return {
        "type": "table",
        "columns": columns,
    }

4.3 推荐规则矩阵

SQL 特征	数据特征	推荐图表	使用场景
包含 DATE/TIME + 数值列	时间序列	📈 line_chart	销售趋势、访问量走势
GROUP BY + ≤8 个分类	维度+度量	🥧 pie_chart	市场份额、品类占比
GROUP BY + >8 个分类	维度+度量	📊 bar_chart	各地区排名、TOP N
无 GROUP BY + 单行	单个/多个 KPI	🔢 metric_card	总额、平均值、计数
其他复杂情况	多列明细	📋 table	原始数据展示

4.4 实际案例演示

案例 1：时间趋势分析

-- 输入 SQL
SELECT DATE(created_at) as date, SUM(amount) as sales
FROM orders
WHERE created_at >= '2026-05-01'
GROUP BY DATE(created_at);

-- 返回数据
[
  {"date": "2026-05-01", "sales": 12345},
  {"date": "2026-05-02", "sales": 15678},
  ...
]

-- 推荐结果 ✅
{
  "type": "line_chart",
  "x_field": "date",
  "y_fields": ["sales"]
}

前端渲染效果：ECharts 折线图，X 轴为日期，Y 轴为销售额。

---

案例 2：品类占比分析

-- 输入 SQL
SELECT c.name as category, SUM(oi.amount) as total
FROM order_items oi
JOIN products p ON oi.product_id = p.id
JOIN categories c ON p.category_id = c.id
GROUP BY c.id;

-- 返回数据（5 个品类）
[
  {"category": "电子产品", "total": 128500},
  {"category": "服装配饰", "total": 89200},
  {"category": "食品饮料", "total": 56800},
  {"category": "家居用品", "total": 34500},
  {"category": "图书文具", "total": 12300}
]

-- 推荐结果 ✅
{
  "type": "pie_chart",
  "label_field": "category",
  "value_field": "total"
}

前端渲染效果：ECharts 饼图，显示各品类销售占比。

---

案例 3：KPI 指标卡

-- 输入 SQL
SELECT COUNT(*) as total_orders,
       AVG(total_amount) as avg_order_value,
       SUM(total_amount) as revenue
FROM orders
WHERE status = 'paid';

-- 返回数据（单行）
[{"total_orders": 1234, "avg_order_value": 256.8, "revenue": 316800}]

-- 推荐结果 ✅
{
  "type": "metric_card",
  "metrics": {
    "total_orders": 1234,
    "avg_order_value": 256.8,
    "revenue": 316800
  }
}

前端渲染效果：3 个大号数字卡片，分别显示订单数、客单价、总收入。

---

📝 五、报告生成器（generate_report_script）

5.1 应用场景

当用户经过多轮对话探索数据后，希望得到一份结构化的总结报告：

👤 用户：帮我生成一份分析报告

🤖 助手：
# 📊 销售数据分析报告
**报告周期**：2026-05-15 ~ 2026-05-22
**生成时间**：2026-05-22 14:30

## 一、🎯 核心发现
1. **本周销售总额突破 120 万**，环比增长 15%，创近半年新高
2. **华东区表现突出**，贡献了 37% 的营收，主要得益于"618预热"活动
3. **电子产品品类持续领跑**，占总销售额的 42%，其中手机配件增长最快（+28%）

## 二、📈 关键指标分析
| 指标 | 本周数值 | 环比变化 | 趋势 |
|------|---------|---------|------|
| 总销售额 | ¥1,234,567 | ↑15% | 📈 强劲增长 |
| 订单量 | 4,567 笔 | ↑8% | 📈 稳定上升 |
| 客单价 | ¥270.3 | ↑6.5% | 📈 持续优化 |
...

## 三、💡 行动建议
1. **加大华东区投入**：考虑增加该区域库存和营销预算
2. **复制成功经验**：分析华东区的爆款商品，推广到其他区域
3. **关注低增长品类**：图书文具品类连续 3 周下滑，需调查原因
...

5.2 实现原理

app/services/chat_engine.py

async def generate_report_script(
    db: AsyncSession,
    session_id: int,
    metrics: list[str],  # 用户关注的指标列表，如 ["销售额", "订单量"]
    style: str = "brief"  # 报告风格：brief / detailed / executive
) -> str:
    """
    生成数据分析报告

    Args:
        session_id: 会话 ID（用于获取历史查询结果）
        metrics: 需要重点分析的指标
        style: 报告详细程度

    Returns:
        Markdown 格式的报告文本
    """
    # Step 1: 获取最近有查询结果的 5 条消息
    msg_result = await db.execute(
        select(ChatMessage)
        .where(
            ChatMessage.session_id == session_id,
            ChatMessage.query_result.isnot(None)  # 只取有结果的
        )
        .order_by(ChatMessage.created_at.desc())
        .limit(5)
    )
    messages = list(msg_result.scalars().all())

    # Step 2: 聚合查询结果
    data = {}
    for msg in messages:
        if msg.query_result:
            try:
                data[msg.sql_query or "query"] = json.loads(msg.query_result)
            except json.JSONDecodeError:
                pass

    # Step 3: 调用 LLM 生成报告
    gateway = LLMGateway()
    report = await gateway.generate_report_script(data, metrics)

    return report

5.3 LLM Prompt 设计

app/services/llm_gateway.py

async def generate_report_script(self, data: dict, metrics: list[str]) -> str:
    system_prompt = """你是一位专业的数据报告撰写专家。根据查询结果和指定的指标，
生成一份中文 Markdown 格式的专业商业报告。

报告必须包含以下部分：
1. **关键发现**（Top 3-5 Insights）：最重要的结论，用数字支撑
2. **指标详细分析**：对每个指标进行深入解读，包含同比/环比
3. **趋势和洞察**：数据背后的原因分析和未来预测
4. **建议和行动方案**：基于数据的具体可执行建议

写作要求：
- 使用专业但易懂的语言
- 所有结论必须有数据支撑（引用具体数值）
- 使用 Markdown 格式（标题、表格、列表、加粗）
- 适当使用 emoji 增强可读性（📈 📉 🎯 💡）
- 总长度控制在 1000-2000 字"""

    user_prompt = f"""查询数据（来自用户的多次查询）：
{json.dumps(data, ensure_ascii=False, indent=2)}

需要重点分析的指标：{', '.join(metrics)}

请生成报告："""

    return await self.generate(user_prompt, system_prompt=system_prompt)

---

🔄 六、完整调用链路总结

一次典型的对话请求处理流程

从 FastAPI 路由到 SQL 执行与前端渲染

性能基准

阶段	耗时	占比
上下文加载 + 语义模型获取	50-100ms	2%
LLM 调用（含网络传输）	2000-4000ms	85%
SQL 提取 + 验证	<1ms	<1%
MySQL 查询执行	50-200ms	5%
图表推荐	<1ms	<1%
数据库写入	10-20ms	1%
总计	2.5-4.5s	100%

瓶颈在 LLM 调用（85% 时间）。后续可通过流式输出优化用户体验。

---

🎯 七、最佳实践总结

✅ 我们做到了什么

1. 真正的多轮对话：维护 10 轮上下文，理解指代关系（”上面”、”之前”）
2. 结构化输出：强制 LLM 按”说明 + SQL + 分析”三段式输出
3. 零配置可视化：_suggest_chart() 根据特征自动选择 5 种图表类型
4. 一键报告生成：聚合多次查询结果，输出专业 Markdown 报告
5. 完善的错误处理：SQL 缺失/验证失败/执行异常均有友好提示

📚 最佳实践清单

• 限制上下文窗口大小（history[-10:]）避免 Token 超限
• 每次 Prompt 都注入完整语义模型（确保字段名准确）
• 使用正则提取 SQL（兼容多种代码块格式）
• 实现 validate_sql() 双层安全验证
• _suggest_chart() 基于 SQL 特征 + 数据类型双重判断
• 报告生成使用明确的模板和约束
• 所有查询结果序列化为 JSON 存储到数据库
• 前端根据 chart_config 字段动态选择图表组件

🚀 下一步

下一篇文章将深入 数据大屏的两阶段分离架构 —— 这是整个项目最具创新性的设计！为什么大屏预览不需要调 LLM？如何实现”设计时解析，运行时执行”？配置驱动的哲学是什么？

敬请期待！🚀

---

相关代码文件：

• app/services/chat_engine.py（chat_engine.py） — 对话引擎核心逻辑
• app/models/chat.py（models/chat.py） — 会话与消息数据模型
• app/api/chat.py（api/chat.py） — RESTful API 接口
• views/chat/ChatRoom.vue — 前端对话界面