从零构建高可用Agent智能问答系统：架构设计与工程实践

1次阅读

没有评论

共计 2790 个字符，预计需要花费 7 分钟才能阅读完成。

传统问答系统在处理多轮对话时常常面临两个核心问题：

上下文丢失：大多数基于规则的问答系统无法有效跟踪对话历史，导致每次用户提问都被当作独立事件处理。例如用户先问 ” 北京的天气 ”，接着问 ” 那上海呢 ”，系统无法理解 ” 那 ” 指的是前文的天气查询。
意图混淆：当用户输入存在歧义时（如 ” 帮我订明天去北京的票 ”，未说明是机票还是火车票），传统系统要么要求用户明确说明，要么随机选择一种处理方式，体验较差。

纯规则引擎：开发维护成本高，但可控性强。适合业务逻辑固定的场景（如银行转账流程）。
纯端到端模型：需大量标注数据，存在 ” 黑箱 ” 风险。适合通用场景但业务逻辑难以干预。

我们采用 BERT+ 规则引擎 的混合方案：

NLU 层用 BERT 处理开放域意图识别
业务逻辑强的场景（如订单查询）走规则引擎
两者结果通过置信度加权融合

graph TD
    A[用户输入] --> B(NLU 模块)
    B --> C{是否业务流程?}
    C -->| 是 | D[规则引擎]
    C -->| 否 | E[BERT 模型]
    D & E --> F(DM 模块)
    F --> G[NLG 模块]
    G --> H[响应输出]

关键设计点：

采用 Hash 结构存储会话，Key 格式：session:{user_id}:{timestamp}
每个会话包含：
context: 最近 3 轮对话的 BERT 编码向量（用于上下文理解）
slots: 当前对话已填充的槽位（如{"city":"北京","date":"2023-08-15"}）
expire: 30 分钟 TTL（避免僵尸会话）

from flask import Flask, request
from utils.auth import jwt_required

app = Flask(__name__)

@app.route('/api/v1/chat', methods=['POST'])
@jwt_required
def chat():
    """
    核心对话接口
    params: {
        "query": "用户输入",
        "session_id": "可选会话 ID"
    }
    """
    data = request.get_json()
    query = data['query']
    session_id = data.get('session_id')

    # 处理逻辑...
    return {
        "response": "回答内容",
        "new_session_id": "新会话 ID"
    }

from transformers import BertTokenizer, BertModel
import torch

# 加载预训练模型
tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
model = BertModel.from_pretrained('bert-base-chinese')

# 文本预处理
def preprocess(text):
    inputs = tokenizer(
        text, 
        max_length=64, 
        padding='max_length',
        truncation=True,
        return_tensors="pt"
    )
    return inputs

# 获取意图向量
def get_intent(text):
    inputs = preprocess(text)
    with torch.no_grad():
        outputs = model(**inputs)
    # 取 [CLS] 位置的向量作为句向量
    return outputs.last_hidden_state[:,0,:].numpy()

class DialogManager:
    def __init__(self):
        self.states = {
            "greeting": self.handle_greeting,
            "query_weather": self.handle_weather,
            "confirm": self.handle_confirm
        }

    def process(self, intent, entities, session):
        handler = self.states.get(session.current_state, self.default_handler)
        return handler(intent, entities, session)

    def handle_weather(self, intent, entities, session):
        if not session.slots.get("city"):
            return "请问您想查询哪个城市的天气?"
        # 其他处理逻辑...

import redis
from concurrent.futures import ThreadPoolExecutor

# 最佳实践：每个工作进程独立连接池
pool = redis.ConnectionPool(
    max_connections=100,
    host='redis-host',
    port=6379,
    decode_responses=True
)
redis_client = redis.Redis(connection_pool=pool)

# 配合线程池使用
executor = ThreadPoolExecutor(max_workers=50)

采用 AC 自动机实现毫秒级匹配：

from ahocorasick import Automaton

ac = Automaton()
# 加载敏感词库
for word in sensitive_words:
    ac.add_word(word, word)
ac.make_automaton()

def filter_text(text):
    found = set()
    for end_index, original in ac.iter(text):
        found.add(original)
    return bool(found)  # 返回是否命中敏感词

在 4 核 8G 云服务器上测试：