🎯 Core Questions of This Chapter

How to build a truly “intelligent” conversational system?

Problems with ordinary chatbots:

  • ❌ No memory: Each conversation round is independent, cannot understand references like “above”, “previously”
  • ❌ Single output: Only returns text, no visualization
  • ❌ Cannot summarize: After 10 rounds, cannot generate an overall report

Our solution:

  1. ✅ Multi-turn context management: Automatically maintain the history of the last 10 rounds
  2. ✅ Intelligent chart recommendation: Automatically select the best visualization based on SQL characteristics
  3. ✅ One-click report generation: Aggregate results from multiple queries, output professional analysis reports

📐 Architecture Overview

Smart Chat Engine Architecture

Smart Chat Engine Architecture

Core Modules at a Glance

Chat Engine Core Modules

Context, LLM, SQL, Chart Recommendation, and Report Generation

💬 I. Multi-turn Context Management (Key Highlight)

1.1 Why Context Matters?

Scenario Comparison:

Dialog Type No Context (Single Turn) With Context (Multi-turn)
User Input “Break down by region” “Break down the above data further by region”
System Understanding ❓ “What data?” ✅ Knows it’s “order total amount data”
Generated SQL SELECT region, SUM(amount) FROM orders GROUP BY region ✅ Same + time filter inherited from previous round

1.2 Session Lifecycle Management

app/services/chat_engine.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
async def create_session(db: AsyncSession, data: ChatSessionCreate) -> ChatSession:
    """Create a new session"""
    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):
    """Load session along with all its messages"""
    result = await db.execute(
        select(ChatSession).where(ChatSession.id == session_id)
    )
    session = result.scalar_one_or_none()

    if session:
        # Load all messages for this session (sorted by time)
        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:
    """Delete session and all associated messages"""
    from sqlalchemy import delete

    # Delete messages first (foreign key constraint)
    await db.execute(
        delete(ChatMessage).where(ChatMessage.session_id == session_id)
    )
    # Then delete session
    result = await db.execute(
        delete(ChatSession).where(ChatSession.id == session_id)
    )
    return result.rowcount > 0

💡 Data Model Design

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 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)  # Natural language reply

    # SQL related fields (only for assistant messages)
    sql_query = Column(Text, nullable=True)  # Extracted SQL
    query_result = Column(Text, nullable=True)  # JSON formatted result
    chart_config = Column(Text, nullable=True)  # Chart configuration JSON

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

1.3 Context Window Control Strategy

app/services/chat_engine.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
async def chat(db: AsyncSession, session_id: int, user_message: str) -> ChatMessage:
    # Step 1: Load recent message history (limit to last 10 rounds)
    history_result = await db.execute(
        select(ChatMessage)
        .where(ChatMessage.session_id == session_id)
        .order_by(ChatMessage.created_at)
    )
    history = list(history_result.scalars().all())

    # ⚡ Key: Take only the last 10 messages (~5 conversation rounds)
    recent_history = history[-10:]

    # Step 2: Build conversation context array
    conversation = []
    for msg in recent_history:
        conversation.append({
            "role": msg.role,
            "content": msg.content
        })

    # Append current user message
    conversation.append({"role": "user", "content": user_message})

🎯 Why 10 Messages Instead of All?

Strategy Token Consumption Context Completeness Use Case
Full History ⚠️ May exceed limit (>80K tokens) ✅ Complete Short conversations (<20 rounds)
Last N items (current) ✅ Controllable (~5K tokens) ✅ Sufficient Long conversations (50+ rounds)
Summary Compression ✅ Very low ❌ May lose details Very long conversations (100+ rounds)

Our choice: history[-10:] balances cost and effectiveness.

💡 Advanced Optimization: Sliding Window + Summary

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def build_smart_context(history, max_tokens=6000):
    """
    Smart context assembly strategy:
    - Always keep the first 2 rounds (establish initial context)
    - Always keep the last 8 rounds (maintain coherence)
    - Replace middle rounds with summary (if too long)
    """
    if len(history) <= 10:
        return history[-10:]

    # Keep head and tail, summarize middle
    head = history[:4]  # First 2 rounds
    tail = history[-6:]  # Last 3 rounds
    middle_summary = summarize_messages(history[4:-6])

    return head + [{"role": "system", "content": f"...(middle {len(history)-10} rounds summarized): {middle_summary}"}] + tail

🤖 II. LLM Conversation Integration and Prompt Construction

2.1 System Prompt Design

app/services/chat_engine.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
async def chat(db: AsyncSession, session_id: int, user_message: str):
    # ... (context preparation)

    # Get semantic model
    semantic_model = await get_semantic_model(db, session.datasource_id)

    # Build System Prompt
    system_prompt = f"""You are a professional data analysis assistant. You help users query and analyze data from databases.
You have access to the following database structure:

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

When the user asks data-related questions, please reply in the following format:
1. Briefly explain what data you are going to query
2. Wrap the generated SQL query statement in ```sql ``` code blocks
3. Provide valuable analysis and insights whenever possible

If the user's question is not data-related, reply normally."""

    # Assemble complete conversation history
    conversation_text = ""
    for msg in conversation:
        role = msg["role"]
        content = msg["content"]
        conversation_text += f"\n{role}: {content}"

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

💡 Three Layers of System Prompt

System Prompt Three Layers

Role Definition, Semantic Model Injection, Output Format Constraints

2.2 Actual Conversation Example

Round 1:

1
2
3
4
5
6
7
8
9
10
11
12
👤 User: Query the total order amount for the last 7 days

🤖 Assistant:
I will query the total order amount for the last 7 days (2026-05-15 to 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;

📊 Analysis Findings:

  • Total orders in the last 7 days: ¥1,234,567
  • Peak on May 18 (¥234,567), possibly due to promotional events
  • Overall upward trend with an average daily growth of 12%
    1
    2
    3

    **Round 2** (leveraging context):

    👤 User: Break down the above data further by region

🤖 Assistant: (understands “above” = order data)
Okay, I will break down the order sales of the last 7 days by region.

1
2
3
4
5
6
7
8
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;

📊 Regional Ranking:

  1. East China: ¥456,789 (37%)
  2. South Region: ¥345,678 (28%)
  3. North Region: ¥234,567 (19%)
    …
    1
    2
    3

    **Round 3** (generating report):

    👤 User: Help me generate an analysis report

🤖 Assistant:
Generating a professional analysis report based on our previous conversation data…

📊 Sales Analysis Report (2026-05-15 ~ 2026-05-22)

I. Key Findings

✅ Weekly total sales exceeded 1.2 million, up 15% MoM
✅ East China contributed the most, accounting for 37% of total
…

(See “Report Generation” section below for details)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

---

## 🔍 III. SQL Extraction and Execution Flow

### 3.1 Extracting SQL from LLM Responses

`app/services/chat_engine.py`

```python
import re

async def chat(...):
    # ... (LLM call)

    sql_query = None
    query_result = None
    chart_config = None

    # Extract SQL code block with regex
    sql_match = re.search(r'```sql\s*(.*?)```', assistant_content, re.DOTALL)
    if sql_match:
        sql_query = sql_match.group(1).strip()

        # Safety check
        if validate_sql(sql_query):
            try:
                # Execute query
                query_result = json.dumps(
                    await execute_query(sql_query, session.datasource_id),
                    ensure_ascii=False,
                    default=str,  # Handle special types like datetime
                )

                # Intelligent chart recommendation
                chart_config = _suggest_chart(
                    sql_query,
                    json.loads(query_result)
                )
            except Exception as e:
                query_result = json.dumps({"error": str(e)})

3.2 Error Handling Strategy

Scenario Handling Approach User Experience
No SQL in response Plain text reply (e.g., chitchat, explanatory questions) Display text only
SQL validation fails Do not execute, log WARNING Prompt “Generated SQL poses a risk”
SQL execution exception Catch exception, return error message Display “Query error: table does not exist”
Empty query result Return empty list normally Display “No data available”

📊 IV. Intelligent Chart Recommendation Algorithm (_suggest_chart)

4.1 Design Philosophy

Core principle: Automatically select the most appropriate chart type based on SQL structural characteristics and statistical features of returned data.

4.2 Algorithm Implementation

app/services/chat_engine.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def _suggest_chart(sql: str, data: list[dict]) -> Optional[dict]:
    """
    Intelligent chart recommendation algorithm

    Args:
        sql: Generated SQL statement
        data: Query results [{col: val}, ...]

    Returns:
        Chart configuration dictionary, or None (when unable to recommend)
    """
    if not data or len(data) == 0:
        return None

    sql_lower = sql.lower()

    # ===== Feature Extraction =====

    # Feature 1: Contains time dimension
    has_date = any(kw in sql_lower for kw in [
        "date", "time", "month", "year", "week",
        "day", "created_at", "updated_at"
    ])

    # Feature 2: Has group-by aggregation
    has_group_by = "group by" in sql_lower

    # Feature 3: Identify column data types
    columns = list(data[0].keys())
    numeric_cols = []  # Numeric columns (measures)
    string_cols = []   # String columns (dimensions)

    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)

    # ===== Rule Matching =====

    # Rule 1: Time trend → Line chart
    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,
        }

    # Rule 2: Group-by aggregation + few categories → Pie chart (suitable for proportion display)
    elif has_group_by and numeric_cols:
        if len(data) <= 8:  # Use pie chart when categories ≤ 8
            return {
                "type": "pie_chart",
                "label_field": string_cols[0] if string_cols else columns[0],
                "value_field": numeric_cols[0],
            }

        # Rule 3: Group-by aggregation + many categories → Bar chart (suitable for comparison)
        else:
            return {
                "type": "bar_chart",
                "x_field": string_cols[0] if string_cols else columns[0],
                "y_fields": numeric_cols,
            }

    # Rule 4: Single row, single numeric value → Metric card (KPI display)
    elif numeric_cols and len(data) == 1:
        return {
            "type": "metric_card",
            "metrics": {col: data[0].get(col) for col in numeric_cols},
        }

    # Fallback: Table display
    return {
        "type": "table",
        "columns": columns,
    }

4.3 Recommendation Rule Matrix

SQL Feature Data Feature Recommended Chart Use Case
Contains DATE/TIME + numeric columns Time series 📈 line_chart Sales trends, traffic patterns
GROUP BY + ≤8 categories Dimension + measure 🥧 pie_chart Market share, category proportions
GROUP BY + >8 categories Dimension + measure 📊 bar_chart Regional rankings, TOP N
No GROUP BY + single row Single/multiple KPIs 🔢 metric_card Total amount, average, count
Other complex cases Multiple columns detail 📋 table Raw data display

4.4 Practical Examples

Example 1: Time Trend Analysis

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
-- Input SQL
SELECT DATE(created_at) as date, SUM(amount) as sales
FROM orders
WHERE created_at >= '2026-05-01'
GROUP BY DATE(created_at);

-- Returned data
[
  {"date": "2026-05-01", "sales": 12345},
  {"date": "2026-05-02", "sales": 15678},
  ...
]

-- Recommended result ✅
{
  "type": "line_chart",
  "x_field": "date",
  "y_fields": ["sales"]
}

Frontend rendering: ECharts line chart, X-axis: date, Y-axis: sales amount.

Example 2: Category Proportion Analysis

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
-- Input 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;

-- Returned data (5 categories)
[
  {"category": "Electronics", "total": 128500},
  {"category": "Clothing & Accessories", "total": 89200},
  {"category": "Food & Beverage", "total": 56800},
  {"category": "Home & Living", "total": 34500},
  {"category": "Books & Stationery", "total": 12300}
]

-- Recommended result ✅
{
  "type": "pie_chart",
  "label_field": "category",
  "value_field": "total"
}

Frontend rendering: ECharts pie chart, showing sales proportions by category.

Example 3: KPI Metric Card

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
-- Input SQL
SELECT COUNT(*) as total_orders,
       AVG(total_amount) as avg_order_value,
       SUM(total_amount) as revenue
FROM orders
WHERE status = 'paid';

-- Returned data (single row)
[{"total_orders": 1234, "avg_order_value": 256.8, "revenue": 316800}]

-- Recommended result ✅
{
  "type": "metric_card",
  "metrics": {
    "total_orders": 1234,
    "avg_order_value": 256.8,
    "revenue": 316800
  }
}

Frontend rendering: 3 large number cards displaying order count, average order value, and total revenue respectively.

📝 V. Report Generator (generate_report_script)

5.1 Use Case

After multiple rounds of data exploration, users may want a structured summary report:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
👤 User: Help me generate an analysis report

🤖 Assistant:
# 📊 Sales Analysis Report
**Report Period**: 2026-05-15 ~ 2026-05-22
**Generated At**: 2026-05-22 14:30

## I. 🎯 Key Findings
1. **Weekly total sales exceeded 1.2 million**, up 15% MoM, hitting a six-month high
2. **East China region performed exceptionally**, contributing 37% of revenue, largely due to the "618 Preheating" campaign
3. **Electronics category continues to lead**, accounting for 42% of total sales, with phone accessories growing the fastest (+28%)

## II. 📈 Key Metrics Analysis
| Metric | This Week Value | MoM Change | Trend |
|--------|-----------------|------------|-------|
| Total Sales | ¥1,234,567 | ↑15% | 📈 Strong growth |
| Order Volume | 4,567 orders | ↑8% | 📈 Steady rise |
| Average Order Value | ¥270.3 | ↑6.5% | 📈 Continuous improvement |
...

## III. 💡 Action Recommendations
1. **Increase investment in East China**: Consider adding inventory and marketing budget to this region
2. **Replicate successful practices**: Analyze best-selling products in East China and promote them to other regions
3. **Monitor low-growth categories**: Books & Stationery has been declining for 3 consecutive weeks; investigate reasons
...

5.2 Implementation Principle

app/services/chat_engine.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
async def generate_report_script(
    db: AsyncSession,
    session_id: int,
    metrics: list[str],  # User-specified metrics, e.g., ["sales", "order volume"]
    style: str = "brief"  # Report style: brief / detailed / executive
) -> str:
    """
    Generate data analysis report

    Args:
        session_id: Session ID (used to retrieve historical query results)
        metrics: Metrics to focus on
        style: Report detail level

    Returns:
        Report text in Markdown format
    """
    # Step 1: Get the last 5 messages with query results
    msg_result = await db.execute(
        select(ChatMessage)
        .where(
            ChatMessage.session_id == session_id,
            ChatMessage.query_result.isnot(None)  # Only those with results
        )
        .order_by(ChatMessage.created_at.desc())
        .limit(5)
    )
    messages = list(msg_result.scalars().all())

    # Step 2: Aggregate query results
    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: Call LLM to generate report
    gateway = LLMGateway()
    report = await gateway.generate_report_script(data, metrics)

    return report

5.3 LLM Prompt Design

app/services/llm_gateway.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
async def generate_report_script(self, data: dict, metrics: list[str]) -> str:
    system_prompt = """You are a professional data report writing expert. Based on the query results and specified metrics,
generate a professional business report in Chinese Markdown format.

The report must include the following sections:
1. **Key Findings** (Top 3-5 Insights): Most important conclusions supported by numbers
2. **Detailed Metric Analysis**: In-depth interpretation of each metric, including YoY/MoM comparisons
3. **Trends and Insights**: Analysis of underlying reasons and future predictions
4. **Recommendations and Action Plan**: Concrete, actionable suggestions based on data

Writing requirements:
- Use professional yet understandable language
- All conclusions must be supported by data (cite specific values)
- Use Markdown formatting (headings, tables, lists, bold)
- Use emojis appropriately to enhance readability (📈 📉 🎯 💡)
- Total length between 1000-2000 characters"""

    user_prompt = f"""Query data (from user's multiple queries):
{json.dumps(data, ensure_ascii=False, indent=2)}

Metrics to focus on: {', '.join(metrics)}

Please generate the report:"""

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

🔄 VI. Complete Call Chain Summary

A Typical Conversation Request Processing Flow

Typical Conversation Request Processing Flow

From FastAPI route to SQL execution and frontend rendering

Performance Benchmark

Stage Time Percentage
Context loading + semantic model retrieval 50-100ms 2%
LLM call (including network transmission) 2000-4000ms 85%
SQL extraction + validation <1ms <1%
MySQL query execution 50-200ms 5%
Chart recommendation <1ms <1%
Database write 10-20ms 1%
Total 2.5-4.5s 100%

Bottleneck is LLM call (85% of time). Can be optimized later with streaming output to improve user experience.

🎯 VII. Best Practices Summary

✅ What We Achieved

  1. True multi-turn dialogue: Maintain 10 rounds of context, understand referential relationships (“above”, “previously”)
  2. Structured output: Force LLM to output in “explanation + SQL + analysis” three-part format
  3. Zero-config visualization: _suggest_chart() automatically selects from 5 chart types based on features
  4. One-click report generation: Aggregate results from multiple queries, output professional Markdown reports
  5. Comprehensive error handling: Friendly prompts for missing SQL, validation failure, execution exceptions

📚 Best Practices Checklist

  • Limit context window size (history[-10:]) to avoid token overflow
  • Inject full semantic model into every prompt (ensure field name accuracy)
  • Use regex to extract SQL (compatible with multiple code block formats)
  • Implement validate_sql() with two-layer security validation
  • _suggest_chart() based on dual judgment of SQL features + data types
  • Use explicit templates and constraints for report generation
  • Serialize all query results as JSON and store in database
  • Frontend dynamically selects chart components based on chart_config field

🚀 Next Steps

The next article will dive deep into the two-stage separation architecture of the data dashboard — the most innovative design of this entire project! Why doesn’t dashboard preview need to call the LLM? How to achieve “parse at design time, execute at runtime”? What is the philosophy behind configuration-driven design?

Stay tuned! 🚀

Related code files:

  • app/services/chat_engine.py - Core logic of the chat engine
  • app/models/chat.py - Session and message data models
  • app/api/chat.py - RESTful API endpoints
  • views/chat/ChatRoom.vue - Frontend chat interface