智能客服系统开发方案：RAG+多智能体技术实现

一、系统架构设计

二、核心模块实现

1. 企业微信接入层

# wechat_connector.py
import json
import requests
from flask import Flask, request

app = Flask(__name__)

class WeChatEnterprise:
    def __init__(self, corp_id, secret, agent_id):
        self.corp_id = corp_id
        self.secret = secret
        self.agent_id = agent_id
        self.token = self._get_access_token()
        
    def _get_access_token(self):
        url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={
              self.corp_id}&corpsecret={
              self.secret}"
        response = requests.get(url).json()
        return response.get('access_token', '')
    
    def handle_message(self, data):
        msg_type = data.get('MsgType')
        user_id = data.get('FromUserName')
        content = data.get('Content', '').strip()
        
        # 消息路由到智能体系统
        from agent_orchestrator import route_message
        response = route_message(user_id, content, msg_type)
        
        # 发送回复
        self.send_text_message(user_id, response)
    
    def send_text_message(self, user_id, content):
        url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={
              self.token}"
        payload = {
            
            "touser": user_id,
            "msgtype": "text",
            "agentid": self.agent_id,
            "text": {
            "content": content}
        }
        requests.post(url, json=payload)

@app.route('/wechat', methods=['POST'])
def wechat_handler():
    data = request.json
    wechat = WeChatEnterprise(corp_id="YOUR_CORP_ID", 
                             secret="YOUR_SECRET",
                             agent_id="YOUR_AGENT_ID")
    wechat.handle_message(data)
    return "Success"

if __name__ == '__main__':
    app.run(port=5000)

2. 多智能体协调中心

# agent_orchestrator.py
from agents.user_info_agent import UserInfoAgent
from agents.product_agent import ProductAgent
from agents.company_agent import CompanyAgent
from conversation_manager import ConversationManager

class AgentOrchestrator:
    def __init__(self):
        self.conv_manager = ConversationManager()
        self.agents = {
            
            'user_info': UserInfoAgent(),
            'product': ProductAgent(),
            'company': CompanyAgent()
        }
    
    def route_message(self, user_id, message):
        # 获取对话上下文
        context = self.conv_manager.get_context(user_id)
        
        # 智能体路由决策
        if self._needs_user_info(context):
            agent = self.agents['user_info']
        elif self._needs_product_recommendation(context):
            agent = self.agents['product']
        elif self._needs_company_intro(context):
            agent = self.agents['company']
        else:
            # 默认使用产品智能体
            agent = self.agents['product']
        
        # 处理消息并更新上下文
        response = agent.process(message, context)
        self.conv_manager.update_context(user_id, message, response)
        return response
    
    def analyze_intention(self, user_id):
        # 获取完整对话记录
        conversation = self.conv_manager.get_full_conversation(user_id)
        
        # 调用意向分析智能体
        from agents.intention_agent import IntentionAgent
        intention = IntentionAgent().analyze(conversation)
        
        # 返回意向等级 (1-4)
        return intention
    
    def _needs_user_info(self, context):
        # 检查是否已收集关键用户信息
        return not all(key in context for key in ['name', 'contact', 'interests'])
    
    def _needs_product_recommendation(self, context):
        # 检查是否已推荐产品
        return 'recommended_products' not in context
    
    def _needs_company_intro(self, context):
        # 根据关键词触发公司介绍
        keywords = ['公司', '介绍', '关于', '背景']
        return any(kw in context['last_message'] for kw in keywords)

# 单例模式实例化
orchestrator = AgentOrchestrator()

def route_message(user_id, message):
    return orchestrator.route_message(user_id, message)

def analyze_intention(user_id):
    return orchestrator.analyze_intention(user_id)

3. RAG知识库集成

# rag_engine.py
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader

class RAGKnowledgeBase:
    def __init__(self, data_path='data/knowledge/'):
        self.embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-base-zh")
        self.knowledge_base = self._initialize_knowledge_base(data_path)
    
    def _initialize_knowledge_base(self, data_path):
        try:
            # 尝试加载现有索引
            return FAISS.load_local("data/vector_store", self.embeddings)
        except:
            # 首次创建索引
            loader = DirectoryLoader(data_path, glob="**/*.txt")
            documents = loader.load()
            
            text_splitter = RecursiveCharacterTextSplitter(
                chunk_size=500,
                chunk_overlap=50
            )
            docs = text_splitter.split_documents(documents)
            
            knowledge_base = FAISS.from_documents(docs, self.embeddings)
            knowledge_base.save_local("data/vector_store")
            return knowledge_base
    
    def retrieve_information(self, query, top_k=3):
        results = self.knowledge_base.similarity_search(query, k=top_k)
        return [doc.page_content for doc in results]
    
    def get_company_introduction(self):
        return self.retrieve_information("公司简介", top_k=1)[0]
    
    def get_product_catalog(self, category=None):
        query = "产品目录" + (f" {
              category}" if category else "")
        return self.retrieve_information(query, top_k=5)

# 知识库单例
knowledge_base = RAGKnowledgeBase()

4. 智能体实现示例

用户信息收集智能体

# agents/user_info_agent.py
import re

class UserInfoAgent:
    def __init__(self):
        self.info_fields = {
            
            'name': {
            'question': '请问您怎么称呼？', 'pattern': r'[我叫|是](w{2,4})'},
            'contact': {
            'question': '方便留个电话或微信吗？', 'pattern': r'1[3-9]d{9}|w+@w+.w+'},
            'interests': {
            'question': '您对哪类产品感兴趣呢？', 'pattern': None}
        }
    
    def process(self, message, context):
        # 检查是否包含信息
        extracted = self._extract_info(message, context)
        
        # 更新上下文
        context.update(extracted)
        
        # 确定下一个问题
        next_question = self._get_next_question(context)
        
        if next_question:
            return next_question
        else:
            # 信息收集完成
            return ("感谢提供信息！接下来为您推荐相关产品..."
                    if context.get('interests') else 
                    "需要我为您介绍产品吗？")

    def _extract_info(self, message, context):
        extracted = {
            }
        for field, config in self.info_fields.items():
            if field not in context:
                if config['pattern']:
                    match = re.search(config['pattern'], message)
                    if match:
                        extracted[field] = match.group(1)
                elif field == 'interests':
                    # 使用NER识别产品类别
                    from utils.nlp_utils import extract_product_categories
                    categories = extract_product_categories(message)
                    if categories:
                        extracted[field] = categories
        return extracted

    def _get_next_question(self, context):
        for field in self.info_fields:
            if field not in context:
                return self.info_fields[field]['question']
        return None

产品推荐智能体

# agents/product_agent.py
from rag_engine import knowledge_base
import random

class ProductAgent:
    def __init__(self):
        self.product_db = {
            
            '电子产品': ['智能手机X1', '平板电脑P2', '智能手表W3'],
            '家居': ['智能灯具L1', '空气净化器A2', '扫地机器人R3'],
            '办公': ['投影仪PJ1', '会议系统M2', '打印机PT3']
        }
    
    def process(self, message, context):
        # 确定用户兴趣
        interests = context.get('interests', [])
        
        if not interests:
            # 未指定兴趣时推荐热门产品
            return self._recommend_popular()
        
        # 基于兴趣推荐
        recommended = []
        for interest in interests:
            if interest in self.product_db:
                recommended.extend(random.sample(
                    self.product_db[interest], 
                    min(2, len(self.product_db[interest]))
                )
        
        if recommended:
            # 保存推荐记录
            context['recommended_products'] = recommended
            
            # 从RAG获取详细信息
            details = []
            for product in recommended:
                info = knowledge_base.retrieve_information(product, top_k=1)
                details.append(f"{
              product}：{
              info[0][:50]}...")
            
            return ("为您推荐以下产品：
" + 
                    "
".join(details) +
                    "
需要更详细的产品手册吗？")
        else:
            return "暂时没有找到匹配的产品，需要人工帮助吗？"
    
    def _recommend_popular(self):
        # 推荐热门产品
        popular = []
        for cat in self.product_db.values():
            popular.extend(cat[:1])
        return "热门推荐：
" + "
".join(popular)

意向分析智能体

# agents/intention_agent.py
import requests
import json

class IntentionAgent:
    DEEPSEEK_API_URL = "https://api.deepseek.com/v1/classify"
    API_KEY = "YOUR_DEEPSEEK_API_KEY"
    
    def analyze(self, conversation):
        # 构造分类提示
        prompt = self._construct_prompt(conversation)
        
        # 调用DeepSeek API
        headers = {
            
            "Authorization": f"Bearer {
              self.API_KEY}",
            "Content-Type": "application/json"
        }
        payload = {
            
            "model": "deepseek-classifier-v1.0",
            "inputs": [{
            "text": prompt}],
            "parameters": {
            
                "categories": ["1", "2", "3", "4"],
                "task_description": "根据对话内容判断客户购买意向等级"
            }
        }
        
        response = requests.post(
            self.DEEPSEEK_API_URL,
            headers=headers,
            data=json.dumps(payload)
        
        if response.status_code == 200:
            result = response.json()
            # 返回置信度最高的类别
            return int(result['predictions'][0]['predicted_label'])
        else:
            # 默认返回中等意向
            return 3
    
    def _construct_prompt(self, conversation):
        # 构建分类提示语
        prompt = (
            "请根据以下客服对话内容，判断客户的购买意向等级：
"
            "等级说明：
"
            "1 - 高意向：明确表示购买意愿
"
            "2 - 中高意向：询问细节、价格等关键信息
"
            "3 - 中等意向：一般性咨询
"
            "4 - 低意向：仅了解信息

"
            "对话内容：
"
        )
        
        # 添加对话记录
        for i, entry in enumerate(conversation):
            role = "客户" if i % 2 == 0 else "客服"
            prompt += f"{
              role}: {
              entry['message']}
"
        
        prompt += "
请直接输出数字等级："
        return prompt

三、对话管理系统

# conversation_manager.py
import datetime
from collections import deque

class ConversationManager:
    def __init__(self, max_context_length=10):
        # 使用字典存储对话状态
        self.conversations = {
            }
        self.max_context = max_context_length
    
    def get_context(self, user_id):
        # 返回最近对话上下文
        if user_id in self.conversations:
            return self.conversations[user_id]['context']
        return self._init_new_conversation(user_id)
    
    def get_full_conversation(self, user_id):
        # 获取完整对话记录
        if user_id in self.conversations:
            return self.conversations[user_id]['history']
        return []
    
    def update_context(self, user_id, user_message, bot_response):
        if user_id not in self.conversations:
            self._init_new_conversation(user_id)
        
        # 更新对话历史
        self.conversations[user_id]['history'].extend([
            {
            'role': 'user', 'message': user_message, 'time': datetime.datetime.now()},
            {
            'role': 'bot', 'message': bot_response, 'time': datetime.datetime.now()}
        ])
        
        # 更新上下文（最近N条）
        context = self.conversations[user_id]['context']
        context['last_message'] = user_message
        context['last_response'] = bot_response
        
        # 维护上下文队列
        context_queue = context.get('recent', deque(maxlen=self.max_context))
        context_queue.append((user_message, bot_response))
        context['recent'] = context_queue
    
    def end_conversation(self, user_id):
        # 结束对话并返回意向分析结果
        intention = analyze_intention(user_id)  # 调用分析模块
        if user_id in self.conversations:
            # 保存对话记录到数据库
            self._save_to_db(user_id, intention)
            del self.conversations[user_id]
        return intention
    
    def _init_new_conversation(self, user_id):
        self.conversations[user_id] = {
            
            'start_time': datetime.datetime.now(),
            'history': [],
            'context': {
            
                'recent': deque(maxlen=self.max_context),
                'state': 'initial'
            }
        }
        return self.conversations[user_id]['context']
    
    def _save_to_db(self, user_id, intention):
        # 实际项目中应保存到数据库
        print(f"保存对话记录：用户{
              user_id}，意向等级{
              intention}")

四、系统部署与优化

部署架构

性能优化策略

对话上下文缓存

# 使用Redis缓存对话状态
import redis

class CachedConversationManager(ConversationManager):
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.StrictRedis(host=redis_host, port=redis_port, db=0)
    
    def get_context(self, user_id):
        # 从Redis获取上下文
        cached = self.redis.get(f"conv:{
              user_id}")
        if cached:
            return json.loads(cached)
        return self._init_new_conversation(user_id)
    
    def update_context(self, user_id, user_message, bot_response):
        super().update_context(user_id, user_message, bot_response)
        # 更新Redis缓存
        context = self.conversations[user_id]['context']
        self.redis.setex(
            f"conv:{
              user_id}", 
            300,  # 5分钟TTL
            json.dumps(context)
        )

智能体并行执行

# 使用协程并行处理
import asyncio

class ParallelAgentOrchestrator(AgentOrchestrator):
    async def process_parallel(self, user_id, message):
        context = self.conv_manager.get_context(user_id)
        
        # 创建并行任务
        tasks = [
            agent.process_async(message, context)
            for agent in self.agents.values()
        ]
        
        # 并行执行
        results = await asyncio.gather(*tasks)
        
        # 结果融合
        return self._combine_responses(results)
    
    def _combine_responses(self, results):
        # 基于优先级合并策略
        if any('产品推荐' in res for res in results):
            return next(res for res in results if '产品推荐' in res)
        # 其他合并逻辑...

五、安全与监控

安全措施

企业微信消息签名验证

# 添加签名验证
import hashlib
import hmac

def verify_signature(token, timestamp, nonce, signature):
    tmp_list = sorted([token, timestamp, nonce])
    tmp_str = ''.join(tmp_list).encode('utf-8')
    calc_sign = hmac.new(token.encode('utf-8'), tmp_str, hashlib.sha256).hexdigest()
    return calc_sign == signature

@app.route('/wechat', methods=['POST'])
def wechat_handler():
    signature = request.args.get('signature', '')
    timestamp = request.args.get('timestamp', '')
    nonce = request.args.get('nonce', '')
    
    if not verify_signature("YOUR_TOKEN", timestamp, nonce, signature):
        return "Invalid signature", 403
    
    # 处理消息...

监控系统

# monitoring.py
from prometheus_client import start_http_server, Counter, Histogram

# 定义监控指标
REQUEST_COUNT = Counter(
    'agent_requests_total',
    'Total agent requests',
    ['agent_type', 'status']
)

RESPONSE_TIME = Histogram(
    'agent_response_seconds',
    'Agent response time',
    ['agent_type']
)

# 在智能体中添加监控
def monitor_agent(agent_type):
    def decorator(func):
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                REQUEST_COUNT.labels(agent_type, 'success').inc()
                return result
            except Exception:
                REQUEST_COUNT.labels(agent_type, 'error').inc()
                raise
            finally:
                duration = time.time() - start
                RESPONSE_TIME.labels(agent_type).observe(duration)
        return wrapper
    return decorator

# 应用装饰器
class ProductAgent:
    @monitor_agent('product')
    def process(self, message, context):
        # 原有逻辑...

六、测试与评估

测试用例设计

# tests/test_intention_analysis.py
import unittest
from agents.intention_agent import IntentionAgent

class TestIntentionAnalysis(unittest.TestCase):
    def setUp(self):
        self.agent = IntentionAgent()
    
    def test_high_intention(self):
        conversation = [
            {
            "role": "user", "message": "这款手机有现货吗？"},
            {
            "role": "bot", "message": "目前有库存，需要预留吗？"},
            {
            "role": "user", "message": "今天能发货吗？价格多少？"}
        ]
        self.assertEqual(self.agent.analyze(conversation), 1)
    
    def test_low_intention(self):
        conversation = [
            {
            "role": "user", "message": "你们公司做什么产品的？"},
            {
            "role": "bot", "message": "我们主营智能硬件设备"},
            {
            "role": "user", "message": "好的谢谢，有需要再联系"}
        ]
        self.assertEqual(self.agent.analyze(conversation), 4)
    
    # 更多测试用例...

if __name__ == '__main__':
    unittest.main()

评估指标

对话完成率：95%+
信息收集完整率：90%+
推荐产品点击率：40%+
意向分类准确率：85%+
平均响应时间：<1.5秒