MBE 文档中心

MBE 多语言训练与交互处理方案

概述

本文档说明 MBE 系统在训练和用户交互时如何处理多语言场景。

一、训练时的多语言处理

1.1 当前实现

MBE 系统在训练时已经支持多语言：

多语言文本编码器

训练时使用 paraphrase-multilingual-MiniLM-L12-v2 模型：

# training/titans_expert_trainer.py
from sentence_transformers import SentenceTransformer
text_encoder = SentenceTransformer(
    'paraphrase-multilingual-MiniLM-L12-v2',  # 多语言模型
    device=str(self.device)
)

支持的语言：

  • 中文（简体/繁体）
  • 英语
  • 日语
  • 韩语
  • 法语
  • 德语
  • 西班牙语
  • 俄语
  • 阿拉伯语
  • 等 50+ 种语言

训练数据处理流程

# 1. 加载多语言训练数据
qa_pairs = [
    ExpertQAPair(question="什么是面包？", answer="面包是..."),  # 中文
    ExpertQAPair(question="What is bread?", answer="Bread is..."),  # 英文
    ExpertQAPair(question="パンとは？", answer="パンは..."),  # 日文
]

# 2. 使用多语言编码器转换为嵌入向量
embeddings = text_encoder.encode(texts)  # 自动处理多语言

# 3. 训练模型（模型本身语言无关）
model.train(embeddings, labels)

1.2 多语言训练数据准备

方法 1: 混合多语言数据

# 训练数据可以包含多种语言
training_data = [
    {"question": "什么是面包？", "answer": "面包是...", "language": "zh-CN"},
    {"question": "What is bread?", "answer": "Bread is...", "language": "en"},
    {"question": "パンとは？", "answer": "パンは...", "language": "ja"},
]

方法 2: 翻译增强数据

# 使用翻译服务将单语言数据扩展为多语言
from shared.src.chat.translation import get_translation_service

translator = get_translation_service()

# 原始中文数据
chinese_data = [{"question": "什么是面包？", "answer": "面包是..."}]

# 翻译为其他语言
for lang in ["en", "ja", "ko"]:
    translated_q = await translator.translate(chinese_data[0]["question"], lang)
    translated_a = await translator.translate(chinese_data[0]["answer"], lang)
    training_data.append({
        "question": translated_q["translated_text"],
        "answer": translated_a["translated_text"],
        "language": lang
    })

1.3 训练最佳实践

  1. 数据平衡：确保各语言数据量相对平衡
  2. 质量优先：优先使用高质量的原生语言数据，而非翻译数据
  3. 语言标记：在训练数据中保留语言标记，便于后续分析
  4. 验证集：为每种语言准备独立的验证集

二、用户交互时的多语言处理

2.1 当前实现

翻译服务 (TranslationService)

系统已实现翻译服务，支持：

from shared.src.chat.translation import get_translation_service

translator = get_translation_service()

# 1. 语言检测
detected_lang = await translator.detect_language("Hello, how are you?")
# 返回: "en"

# 2. 文本翻译
result = await translator.translate(
    text="Hello, how are you?",
    target_language="zh-CN",
    source_language="en"
)
# 返回: {"translated_text": "你好，你好吗？", ...}

# 3. 自动翻译（根据用户偏好）
translated = await translator.auto_translate_for_user(
    message={"content": "Hello"},
    user_preferred_language="zh-CN"
)

翻译专家 (TranslationExpert)

系统包含专门的翻译专家，可以处理翻译任务。

2.2 改进方案：集成到引擎流程

方案 A: 前置语言检测和翻译

在引擎处理前检测用户语言，统一转换为系统默认语言：

# src/core/engine.py (改进建议)

async def process(
    self,
    user_input: str,
    device_id: str,
    user_id: Optional[str] = None,
    user_language: Optional[str] = None  # 新增：用户偏好语言
) -> MisesAnalyzeResult:
    """处理用户输入"""
    
    # 1. 检测用户输入语言
    from shared.src.chat.translation import get_translation_service
    translator = get_translation_service()
    
    detected_lang = await translator.detect_language(user_input)
    
    # 2. 如果用户指定了偏好语言，且输入语言不同，则翻译
    if user_language and detected_lang != user_language:
        # 翻译用户输入到偏好语言
        translated = await translator.translate(
            user_input,
            target_language=user_language,
            source_language=detected_lang
        )
        user_input = translated["translated_text"]
        logger.info(f"Translated input from {detected_lang} to {user_language}")
    
    # 3. 继续正常处理流程
    result = await self._process_stage(...)
    
    # 4. 如果用户偏好语言不是系统默认语言，翻译回复
    if user_language and user_language != "zh-CN":
        result.answer = await translator.translate(
            result.answer,
            target_language=user_language,
            source_language="zh-CN"
        )["translated_text"]
    
    return result

方案 B: 多语言原生处理（推荐）

保持用户输入语言，让模型直接处理多语言：

# src/core/engine.py (改进建议)

async def process(
    self,
    user_input: str,
    device_id: str,
    user_id: Optional[str] = None,
    user_language: Optional[str] = None
) -> MisesAnalyzeResult:
    """处理用户输入（多语言原生支持）"""
    
    # 1. 检测语言（用于日志和统计）
    from shared.src.chat.translation import get_translation_service
    translator = get_translation_service()
    detected_lang = await translator.detect_language(user_input)
    
    # 2. 在上下文中添加语言信息
    context = {
        "user_language": detected_lang,
        "user_preferred_language": user_language or detected_lang
    }
    
    # 3. 使用多语言编码器处理（模型本身支持多语言）
    # 由于使用了 paraphrase-multilingual-MiniLM-L12-v2，
    # 模型可以直接理解多语言输入
    
    # 4. 处理流程（语言无关）
    result = await self._process_stage(
        session=session,
        user_input=user_input,  # 保持原始语言
        profile=profile,
        context=context
    )
    
    # 5. 如果用户指定了偏好语言，且回复语言不同，则翻译回复
    if user_language and detected_lang != user_language:
        result.answer = await translator.translate(
            result.answer,
            target_language=user_language,
            source_language=detected_lang
        )["translated_text"]
    
    return result

2.3 用户语言偏好管理

存储用户语言偏好

# 在用户画像中存储语言偏好
class UserProfile:
    preferred_language: str = "zh-CN"  # 默认中文
    detected_languages: List[str] = []  # 历史检测到的语言

API 支持语言参数

# src/mcp/server.py

class AnalyzeRequest(BaseModel):
    user_input: str
    device_id: str
    user_id: Optional[str] = None
    preferred_language: Optional[str] = None  # 新增：用户偏好语言
    auto_detect: bool = True  # 是否自动检测语言

2.4 专家路由的多语言支持

专家路由系统已经使用多语言编码器，可以处理多语言查询：

# src/knowledge/expert_router.py

# 专家路由使用相同的多语言编码器
# 因此可以匹配多语言查询到专家
expert_router.route_and_answer(
    query="What is bread?",  # 英文查询
    context={"language": "en"}
)
# 可以匹配到中文训练的专家（因为嵌入空间是共享的）

三、实施建议

3.1 短期改进（立即实施）

  1. 在引擎入口添加语言检测

    # 在 engine.process() 开始时检测语言
detected_lang = await translator.detect_language(user_input)
logger.info(f"User input language: {detected_lang}")

  2. 添加用户语言偏好参数

    # API 请求中添加 preferred_language 参数
# 用户画像中存储语言偏好

  3. 翻译回复（如果需要）

    # 如果用户偏好语言与回复语言不同，翻译回复

3.2 中期改进（1-2周）

  1. 完善语言检测和翻译集成

    • 在专家路由中考虑语言
    • 在知识库检索中支持多语言
    • 优化翻译质量

  2. 多语言训练数据增强

    • 为现有专家添加多语言训练数据
    • 使用翻译服务扩展训练集

  3. 性能优化

    • 缓存语言检测结果
    • 批量翻译优化

3.3 长期改进（1个月+）

  1. 多语言知识库

    • 支持多语言知识库存储
    • 跨语言知识检索

  2. 语言特定优化

    • 为不同语言优化提示词
    • 语言特定的专家

  3. 用户语言学习

    • 记录用户语言使用模式
    • 自动适应语言偏好

四、代码示例

4.1 训练多语言专家

# scripts/train_multilingual_expert.py

from training.titans_expert_trainer import ExpertTrainer, ExpertTrainingConfig
from shared.src.chat.translation import get_translation_service

async def train_multilingual_expert():
    # 1. 准备多语言训练数据
    qa_pairs = []
    
    # 中文数据
    qa_pairs.extend([
        ExpertQAPair(question="什么是面包？", answer="面包是..."),
        ExpertQAPair(question="如何制作面包？", answer="制作面包需要..."),
    ])
    
    # 英文数据
    qa_pairs.extend([
        ExpertQAPair(question="What is bread?", answer="Bread is..."),
        ExpertQAPair(question="How to make bread?", answer="To make bread..."),
    ])
    
    # 2. 训练（使用多语言编码器）
    config = ExpertTrainingConfig(
        expert_id="bread_master_multilingual",
        expert_name="面包大师（多语言）",
        kb_id="bread_kb",
        epochs=100
    )
    
    trainer = ExpertTrainer(config)
    await trainer.train(qa_pairs)
    
    print("✅ 多语言专家训练完成")

4.2 多语言用户交互

# src/api/multilingual_chat.py

from fastapi import APIRouter
from shared.src.chat.translation import get_translation_service
from src.core.engine import get_engine

router = APIRouter()

@router.post("/chat")
async def multilingual_chat(
    user_input: str,
    user_id: str,
    preferred_language: Optional[str] = None
):
    """多语言聊天接口"""
    
    translator = get_translation_service()
    engine = get_engine()
    
    # 1. 检测用户输入语言
    detected_lang = await translator.detect_language(user_input)
    
    # 2. 获取用户偏好语言（从用户画像或参数）
    if not preferred_language:
        # 从用户画像获取
        profile = await get_user_profile(user_id)
        preferred_language = profile.preferred_language or detected_lang
    
    # 3. 处理（保持原始语言，模型支持多语言）
    result = await engine.process(
        user_input=user_input,
        device_id=user_id,
        user_id=user_id
    )
    
    # 4. 如果用户偏好语言不同，翻译回复
    if preferred_language != detected_lang:
        translated = await translator.translate(
            result.answer,
            target_language=preferred_language,
            source_language=detected_lang
        )
        result.answer = translated["translated_text"]
    
    return {
        "answer": result.answer,
        "detected_language": detected_lang,
        "preferred_language": preferred_language
    }

五、测试建议

5.1 训练测试

# tests/test_multilingual_training.py

def test_multilingual_training():
    """测试多语言训练"""
    # 1. 准备多语言数据
    # 2. 训练模型
    # 3. 测试各语言查询
    # 4. 验证回答质量
    pass

5.2 交互测试

# tests/test_multilingual_interaction.py

async def test_multilingual_chat():
    """测试多语言交互"""
    # 1. 发送中文查询
    # 2. 发送英文查询
    # 3. 发送日文查询
    # 4. 验证语言检测
    # 5. 验证翻译质量
    pass

六、总结

当前状态

✅ 已实现：

  • 多语言文本编码器（训练时）
  • 翻译服务（TranslationService）
  • 翻译专家（TranslationExpert）
  • 语言检测功能

⚠️ 待改进：

  • 引擎流程中未集成语言检测和翻译
  • 用户语言偏好未存储和使用
  • API 未支持语言参数

推荐方案

方案 B（多语言原生处理）：

  1. 保持用户输入原始语言
  2. 利用多语言编码器的能力直接处理
  3. 仅在用户明确要求时翻译回复
  4. 记录语言偏好供后续使用

这样既保持了多语言的灵活性，又避免了不必要的翻译开销。