> **本文是「架构实战·第二季」第十篇**。前面我们讲了 Agent 和多 Agent 协作，本篇进入 AI Agent 最常见、最刚需的形态——对话系统。

---

## 引子：客服 AI 的"第三次进化"

过去五年，客服 AI 经历了三代演进：

1. **规则机器人**（2018 前）：写大量 if-else，效果差、维护成本高；
2. **检索式客服**（2019-2022）：BM25 + FAQ 匹配，能回答 60% 常见问题；
3. **LLM 智能助手**（2023+）：能理解复杂问题、调系统、做操作。

第三代核心架构是什么？这就是这一篇要讲的。

---

## 一、对话系统的本质

### 1.1 一个 AI 对话系统包含什么

```
用户输入
   ↓
┌──────────────┐
│  输入理解     │ NLU / Intent / Slot Filling
└──────┬───────┘
       ↓
┌──────────────┐
│  对话状态     │ DST（Dialog State Tracker）
└──────┬───────┘
       ↓
┌──────────────┐
│  决策/Policy  │ 该回答、调工具、还是转人工
└──────┬───────┘
       ↓
┌──────────────┐
│  回应生成     │ NLG / Agent 调用
└──────┬───────┘
       ↓
用户输出
```

传统对话系统每一步是独立模块，LLM 时代则可以**用一个 LLM 统一承接**所有步骤。

### 1.2 现代 LLM 对话系统的架构

```
┌────────────────────────────────────────┐
│         LLM 对话引擎                   │
│  ┌──────────────────────────────────┐  │
│  │ System Prompt（含所有规则）      │  │
│  └──────────────────────────────────┘  │
└────────────────────────────────────────┘
         ↑              ↓
    历史消息         工具调用
         ↓
┌────────────────────────────────────────┐
│            记忆系统                     │
│  · 短期（session）· 长期（user-level） │
│  · 摘要 · 向量检索                      │
└────────────────────────────────────────┘
         ↑              ↓
┌────────────────────────────────────────┐
│         工具与知识                      │
│  · RAG · Function Calling · 数据库     │
└────────────────────────────────────────┘
```

---

## 二、对话状态管理

### 2.1 短期记忆（Session Context）

最直接的做法：把整个对话历史塞到 messages 里。

```python
class SessionManager:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.MAX_HISTORY = 20  # 保留最近 20 轮
    
    def get_messages(self, session_id: str):
        history = self.redis.lrange(f"session:{session_id}", 0, -1)
        return [json.loads(m) for m in history]
    
    def add_message(self, session_id: str, role: str, content: str):
        msg = json.dumps({"role": role, "content": content})
        self.redis.rpush(f"session:{session_id}", msg)
        self.redis.ltrim(f"session:{session_id}", -self.MAX_HISTORY, -1)
        self.redis.expire(f"session:{session_id}", 3600)  # 1h 过期
```

### 2.2 长期记忆（User Preference）

用户跨 session 的偏好，例如：

- "用户喜欢简洁的回答"
- "用户是 HR 角色"
- "用户的语言风格偏专业"

```python
class LongTermMemory:
    def __init__(self, db):
        self.db = db
    
    def get_user_profile(self, user_id):
        # 加载用户档案
        return UserProfile.from_db(user_id)
    
    def update_preference(self, user_id, pref_type, content):
        """用户反馈或自动分析得到的偏好"""
        UserPreference.create(
            user_id=user_id,
            type=pref_type,
            content=content,
            confidence=0.7,
            source="auto"
        )
```

调用时，把长期记忆注入系统 prompt：

```python
SYSTEM_PROMPT = f"""你是 {{company}} 的客服助手。

【用户偏好】
{format_preferences(user_profile.preferences)}

【当前对话历史】
{format_messages(session_history)}

【用户问题】
{user_query}"""
```

### 2.3 上下文压缩

session 长时上下文会爆炸，需要压缩：

**压缩策略**：

```python
class ContextCompressor:
    def __init__(self, llm):
        self.llm = llm
    
    def compress(self, messages):
        # 1. 保留最近 N 轮详细
        recent = messages[-10:]
        
        # 2. 总结更早的
        older = messages[:-10]
        if older:
            summary = self.llm.complete(f"""请总结以下对话的关键信息：
{format_messages(older)}
            
输出格式：
- 主要讨论了 X
- 用户的核心诉求是 Y
- 已经达成的结论是 Z""")
        else:
            summary = ""
        
        return summary, recent
```

调用时：

```python
def build_messages(query, session_id, user_id):
    history = session_mgr.get_messages(session_id)
    summary, recent = compressor.compress(history)
    
    profile = long_term.get_user_profile(user_id)
    
    messages = [
        {"role": "system", "content": build_system_prompt(profile, summary)},
        *recent,
        {"role": "user", "content": query}
    ]
    return messages
```

---

## 三、工具调用与意图路由

### 3.1 何时调工具

对话系统的核心决策："这个问题是**信息查询**，还是**业务操作**？"

**用 LLM 自判断**：

```python
SYSTEM_PROMPT = """你是客服助手。请判断用户意图属于以下哪种：

1. simple_qa - 简单知识问答
2. lookup - 需要查订单、商品等业务数据
3. action - 需要做退款、改地址等操作
4. complaint - 用户投诉，需要转人工
5. chitchat - 闲聊

【输出格式】
JSON: {"intent": "...", "confidence": 0.95, "entities": {...}}"""
```

### 3.2 路由到具体工具

```python
class ConversationEngine:
    def __init__(self, llm, tools, rag_engine):
        self.llm = llm
        self.tools = tools
        self.rag = rag_engine
    
    def handle(self, query, session_id, user_id):
        # 1. 路由
        intent = self.classify_intent(query)
        
        if intent["intent"] == "simple_qa":
            # RAG 召回 + 回答
            docs = self.rag.retrieve(query)
            answer = self.llm.complete(build_prompt(query, docs))
        elif intent["intent"] == "lookup":
            # 调业务查询
            answer = self.call_lookup_tool(query, intent["entities"])
        elif intent["intent"] == "action":
            # 先确认 + 再执行
            answer = self.confirm_and_execute(query, intent["entities"])
        elif intent["intent"] == "complaint":
            # 转人工
            answer = self.transfer_to_human(query)
        else:
            answer = self.llm.complete(query)  # 通用对话
        
        return answer
```

### 3.3 敏感操作的人机协作

退款、改地址这类操作，必须二次确认：

```python
async def confirm_and_execute(self, query, entities):
    """用户每次做敏感操作前确认"""
    
    # 1. 把操作翻译给用户看
    confirmation = f"""
您即将执行以下操作：
- 操作类型：退款
- 订单号：{entities.get('order_id')}
- 退款金额：¥{entities.get('amount')}

请回复"确认"执行，或"取消"放弃。
"""
    
    # 2. 等待用户确认
    user_response = await wait_for_user_input()
    
    if user_response.strip() == "确认":
        # 3. 执行 + 通知
        result = await self.tools["apply_refund"].func(**entities)
        return f"退款已成功，预计 3-5 工作日到账。"
    else:
        return "操作已取消。"
```

---

## 四、流式响应与中断处理

### 4.1 流式响应

LLM 一次完整输出可能 10-30 秒，必须流式：

```python
async def stream_response(query, session_id, user_id):
    messages = build_messages(query, session_id, user_id)
    
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        stream=True,
    )
    
    full_text = ""
    async for chunk in response:
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            full_text += token
            yield {"type": "token", "content": token}
    
    yield {"type": "done", "full": full_text}
    session_mgr.add_message(session_id, "assistant", full_text)
```

前端 SSE 接入：

```javascript
const eventSource = new EventSource("/api/chat");
eventSource.onmessage = (e) => {
    const data = JSON.parse(e.data);
    if (data.type === "token") {
        appendToChat(data.content);
    } else if (data.type === "done") {
        finalizeChat(data.full);
    }
};
```

### 4.2 用户中断

用户可能 AI 还没说完就发新问题，必须支持中断：

```python
class InterruptibleStream:
    def __init__(self):
        self.interrupt_flag = False
    
    async def stream_with_interrupt(self, response_stream, user_id):
        async for chunk in response_stream:
            if self.check_interrupt(user_id):
                self.interrupt_flag = True
                break
            yield chunk
    
    def check_interrupt(self, user_id):
        """检查用户是否发了新消息"""
        new_msg = redis.lpop(f"interrupt:{user_id}")
        return bool(new_msg)
```

前端实现：

```javascript
sendBtn.onclick = () => {
    userInput.disabled = false;
    // 中断上一轮
    if (currentStream) currentStream.close();
    // 发送新消息
    startNewStream();
};
```

---

## 五、对话质量管理

### 5.1 评测维度

| 维度 | 说明 |
|------|------|
| 任务完成率 | 用户目标达成 |
| 回答准确率 | 内容是否正确 |
| 一次解决率 | 是否需要转人工 |
| 平均对话轮次 | 效率 |
| 用户满意度 | CSAT |
| 重复率 | AI 重复说同一句 |

### 5.2 自动化评测

```python
def evaluate_conversation(messages, expected_intent):
    """基于脚本的自动评测"""
    result = {
        "task_completed": False,
        "rounds": len([m for m in messages if m["role"] == "user"]),
        "transferred": False,
    }
    
    for msg in messages:
        if msg["role"] == "system" and "transferred" in msg["content"]:
            result["transferred"] = True
        if msg.get("intent") == expected_intent:
            result["task_completed"] = True
    
    return result
```

### 5.3 LLM-as-a-Judge

```python
JUDGE_PROMPT = """你是客服质量评估员，请根据以下对话评估质量。

【评估维度】（每项 0-10）
- 任务完成度：用户问题是否被解决
- 回答准确性：信息是否正确
- 沟通友好度：语气是否专业、亲切
- 转人工判断：是否该转时未转、不该转时乱转

【对话记录】
{conversation}

【用户问题】
{user_query}

【用户满意度】
{csat_score}

请输出 JSON：{"scores": {...}, "issues": ["..."], "suggestions": ["..."]}"""

def llm_judge(conversation):
    return json.loads(llm.complete(JUDGE_PROMPT.format(
        conversation=format_conversation(conversation)
    )))
```

---

## 六、踩坑提示

### 坑 1：上下文窗口爆

**症状**：session 跑了 50 轮后，模型开始"胡说"。

**解决**：每 10 轮做一次摘要，保留摘要 + 最近 5 轮原文。

### 坑 2：长期记忆注入过多

**症状**：把所有用户偏好都塞进 prompt，模型被淹没。

**解决**：只注入相关偏好。用 RAG 召回偏好：

```python
# 把偏好也做向量化
relevant_prefs = preference_vector.search(query, top_k=3)
```

### 坑 3：用户画像被注入污染

**症状**：用户说"我不在乎省钱"，被记入偏好，下次回答变成不推荐省钱方案。

**解决**：偏好来源必须可审计 + 可撤销：

```python
UserPreference:
    - id
    - user_id
    - content
    - confidence (0-1)
    - source (auto/manual/user)
    - created_at
    - expires_at  # 时效
```

### 坑 4：流式响应解析报错

**症状**：中文 UTF-8 跨 chunk 切割时乱码。

**解决**：前端用 `TextDecoder('utf-8', {stream: true})` 处理：

```javascript
const decoder = new TextDecoder('utf-8', {stream: true});
const reader = response.body.getReader();
while (true) {
    const {done, value} = await reader.read();
    if (done) break;
    const text = decoder.decode(value, {stream: true});
    appendToChat(text);
}
```

### 坑 5：对话状态丢失

**症状**：用户刷新页面，session 没了。

**解决**：session 必须持久化，刷新时从 localStorage 恢复 session_id。

---

## 七、本篇要点回顾

| 模块 | 关键点 |
|------|--------|
| 短期记忆 | Redis 存 session，自动过期 |
| 长期记忆 | 用户画像 + RAG 召回 |
| 上下文压缩 | 摘要 + 最近轮次 |
| 意图路由 | LLM 分类 + 工具调用 |
| 流式 + 中断 | SSE + 用户可中断 |
| 评测 | 自动化 + LLM-as-judge |

**对话系统的三个核心原则**：

1. **记忆是灵魂**——没有上下文的对话都是单轮问答；
2. **必须有降级路径**——任何环节都可能崩，必须有转人工兜底；
3. **流式 + 中断是用户体验关键**——AI 助手不能让人"等着"。

---

## 八、下一篇预告

下一篇 【架构实战】**LLM文档处理：长文本切片与结构化抽取**，我们深入 LLM 处理长文档的核心难题：

- 文档解析的难点：PDF/扫描件/表格/图片；
- 长文本切片策略：固定长度 vs 语义切分 vs 结构化切片；
- 结构化抽取：表格解析、字段识别、关系抽取；
- 实战：用 LLM 做企业合同信息抽取。

---

> **作者**：架构实战 第二季  
> **版本**：v1.0  
> **最后更新**：2026-10  
> **全文约 4500 字**

本文是「架构实战·第二季」第十篇。前面我们讲了 Agent 和多 Agent 协作，本篇进入 AI Agent 最常见、最刚需的形态——对话系统。

【架构实战】AI对话系统：从客服机器人到智能助手

研发部经理

人工智能

架构设计

后端

本文深入解析AIAgent对话系统架构，涵盖对话状态管理、工具调用、流式响应等核心模块，分享上下文压缩、意图路由、敏感操作确认等实战经验，并总结五大常见坑及解决方案，助你快速构建稳定高效的LLM智能客服系统。

2026上云采购 | AI焕新·智启新局

mmdp

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【架构实战】AI对话系统：从客服机器人到智能助手-腾讯云开发者社区-腾讯云

【架构实战】AI对话系统：从客服机器人到智能助手

【架构实战】AI对话系统：从客服机器人到智能助手

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐