多语言 LLM 部署指南

import langdetect
from langdetect import detect, DetectorFactory

# 设置种子保证检测结果稳定
DetectorFactory.seed = 42

def detect_language(text: str) -> str:
    """检测文本语言，返回 ISO 639-1 代码"""
    # 预处理：去除代码块、URL 等干扰内容
    import re
    clean = re.sub(r'```.*?```', '', text, flags=re.DOTALL)
    clean = re.sub(r'https?://\S+', '', clean)
    
    if len(clean.strip()) < 10:
        return "en"  # 太短，默认英文
    
    try:
        lang = detect(clean)
        # 映射到支持的语言集合
        supported = {"zh-cn", "zh-tw", "en", "ja", "ko", "fr", "de", "es"}
        if lang in supported:
            return lang
        elif lang.startswith("zh"):
            return "zh-cn"
        else:
            return "en"  # 不支持的语言回退到英文
    except:
        return "en"

def route_by_language(lang: str, config: dict) -> dict:
    """根据语言路由到不同的模型/配置"""
    routing = {
        "zh-cn": {"model": "qwen2.5-72b", "embedding": "bge-m3", "system_lang": "zh"},
        "zh-tw": {"model": "qwen2.5-72b", "embedding": "bge-m3", "system_lang": "zh-tw"},
        "en": {"model": "llama-3.1-70b", "embedding": "bge-m3", "system_lang": "en"},
        "ja": {"model": "qwen2.5-72b", "embedding": "bge-m3", "system_lang": "ja"},
        "default": {"model": "llama-3.1-70b", "embedding": "bge-m3", "system_lang": "en"}
    }
    return routing.get(lang, routing["default"])

from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue

class MultilingualRetriever:
    def __init__(self):
        self.qdrant = QdrantClient(host='localhost', port=6333)
        self.embedding_url = "http://embedding-server:8080/embed"
    
    async def embed(self, text: str) -> list[float]:
        """用 bge-m3 生成多语言向量"""
        import httpx
        async with httpx.AsyncClient() as client:
            resp = await client.post(self.embedding_url, json={"inputs": text})
            return resp.json()[0]
    
    async def search(self, query: str, user_lang: str, top_k: int = 5):
        """跨语言检索：查询是中文，可以检索到英文文档"""
        query_vector = await self.embed(query)
        
        # 方案1：纯向量检索（跨语言天然支持）
        results = self.qdrant.search(
            collection_name="multilingual_docs",
            query_vector=query_vector,
            limit=top_k * 2,  # 多取一些用于过滤
        )
        
        # 方案2：优先返回用户语言，但不排除其他语言
        scored_results = []
        for hit in results:
            doc_lang = hit.payload.get("lang", "en")
            # 同语言加分，但不排除其他语言
            lang_boost = 1.2 if doc_lang == user_lang else 1.0
            scored_results.append({
                "content": hit.payload["content"],
                "lang": doc_lang,
                "score": hit.score * lang_boost,
                "original_score": hit.score
            })
        
        scored_results.sort(key=lambda x: x["score"], reverse=True)
        return scored_results[:top_k]
    
    async def search_with_translation(self, query: str, user_lang: str):
        """进阶方案：先检索，再翻译结果"""
        results = await self.search(query, user_lang)
        
        # 对非用户语言的文档做即时翻译
        import asyncio
        tasks = []
        for r in results:
            if r["lang"] != user_lang:
                tasks.append(self._translate(r["content"], r["lang"], user_lang))
            else:
                tasks.append(asyncio.coroutine(lambda r=r: r["content"])())
        
        translated = await asyncio.gather(*tasks)
        for r, t in zip(results, translated):
            r["translated_content"] = t
        
        return results

CULTURAL_ADAPTATIONS = {
    "zh-cn": {
        "greeting": "您好，有什么可以帮您？",
        "formality": "formal",  # 正式用语
        "date_format": "YYYY年MM月DD日",
        "number_format": "万进制",  # 万、亿
        "taboo_topics": ["政治敏感话题"],
        "response_style": "detailed",  # 中文用户偏好详细回答
    },
    "en": {
        "greeting": "Hi! How can I help you?",
        "formality": "casual",
        "date_format": "MM/DD/YYYY",
        "number_format": "thousand",  # K, M, B
        "taboo_topics": [],
        "response_style": "concise",  # 英文用户偏好简洁
    },
    "ja": {
        "greeting": "こんにちは、お手伝いできますか？",
        "formality": "formal",
        "date_format": "YYYY年MM月DD日",
        "number_format": "万进制",
        "taboo_topics": [],
        "response_style": "polite_detailed",  # 日文偏好礼貌且详细
    }
}

def adapt_response(text: str, target_lang: str) -> str:
    """根据目标语言文化调整响应风格"""
    config = CULTURAL_ADAPTATIONS.get(target_lang, CULTURAL_ADAPTATIONS["en"])
    
    if config["response_style"] == "concise":
        # 英文：精简，去除冗余修饰
        pass  # 通过 system prompt 控制
    elif config["response_style"] == "detailed":
        # 中文：添加结构化说明
        pass
    
    return text

def handle_mixed_language(text: str) -> dict:
    """处理混合语言输入"""
    # 检测主要语言
    primary_lang = detect_language(text)
    
    # 提取嵌入的外文片段
    import re
    en_fragments = re.findall(r'[a-zA-Z][a-zA-Z\s]+', text)
    
    strategy = "keep_mixed"  # 保留混合，不做翻译
    
    return {
        "primary_lang": primary_lang,
        "has_mixed": len(en_fragments) > 0,
        "strategy": strategy,
        "query_text": text,  # 原样传给多语言模型
        "embedding_text": text  # bge-m3 天然支持混合语言
    }