我要投稿

RAG 实现多语言客户端技术方案要点分析

发布日期：2025-08-26 08:04:08 浏览次数： 1518

作者：AI悠悠

微信搜一搜，关注“AI悠悠”

场景

RAG项目对应的目标客户群体是多个不同国家、不同语言，比如英语、中文、德语等。开发RAG项目，该如何应对这种场景呢？

一、知识库搭建

1、知识库存储相关数据，是否要多语言存储？

若能够提供多语言版本的知识源数据文档，建议各种语言都Embeddings向量化存储一份；可分在不同的表里，也可以在同一个表里。

若无法提供多语言版本的知识源数据文档，或有些能有些不能，则只存储其中一份较为完善的文档。

二、提示词模版

1、提示词模版是否需要每种语言各设置一份？

强烈建议，每种语言各设置一份；并且多次检索的每次调用大模型的提示词语言保持语言一致，否则会出现语言随机出现的情况。

2、如何判断客户输入的是什么语言？

可使用lingua包，对用户输入的文本进行语言识别。

pip install lingua-language-detector

from lingua import Language, LanguageDetectorBuilder
def detect_language(text: str) -> str:    """检测文本语言，返回 zh/en"""    if not text:        return "en"    detector = LanguageDetectorBuilder.from_languages(Language.CHINESE, Language.ENGLISH, Language.SPANISH).build()    lang = detector.detect_language_of(text)    if lang == Language.CHINESE:        return "zh"    else:        return "en"

3、多轮对话引擎如何设置提示词模版？

以LlamaIndex框架的多轮对话引擎来实现，这里我们采用CondenseQuestionChatEngine对话引擎+CitationQueryEngine 查询引擎。

其中查询引擎CitationQueryEngine 需要设置2个提示词模版，citation_qa_template和citation_refine_template。如何根据多语言设置呢？LlamaIndex默认已经帮我们实现一套英文版本的提示词，根据源码可以找到提示词模版，然后拿出来翻译成各种语言版本即可。

对话引擎CondenseQuestionChatEngine还有一个提示词模版condense_question_prompt，设置方法跟上面的一样。

最终在使用该引擎时候指定对自定义的提示词模版引擎。

    async def achat_stream(self, query: str, session_id: str,lang:str='en'):        """        RAG-Base方法        :param query:客户的提示        :param session_id:对话的SessionId        :param lang:客户的语言        :return:        """        logger.warning(f"query={query},lang={lang},session_id={session_id}")        # 1- 初始化知识库索引        muxue_vector=MuXueVectorStore()        semantic_index = await muxue_vector.init_kbm_vector_index()        # 2- 获取长期记忆和短期记忆        memory = await self.get_session_memory(session_id)        # logger.info(f"memory={memory}")
        # 获取提示词模版        tcp = MuXueCitationQueryPrompt(lang)        citation_qa_tmpl = tcp.get_citation_qa_prompt()        citation_refine_tmpl = tcp.get_citation_refine_prompt()        query_engine = CitationQueryEngine.from_args(            semantic_index,            similarity_top_k=3,            citation_chunk_size=512,            streaming=True,            citation_qa_template=citation_qa_tmpl,            citation_refine_template=citation_refine_tmpl,            verbose=True,            filters=MetadataFilters(filters=[MetadataFilter(key="lang", value=lang)]),            node_postprocessors=[FallbackNodePostprocessor(threshold=0.6,verbose=True,lang=lang)]  #  过滤低分        )        cq_prompt_custom=MuXueCondenseQuestionPrompt(lang).get_prompt()        chat_engine = CondenseQuestionChatEngine.from_defaults(            query_engine=query_engine,            condense_question_prompt=cq_prompt_custom,            memory=memory,            verbose=True,        )        resp = await chat_engine.astream_chat(query)        async for delta in resp.async_response_gen():            # print(f"delta:{delta}")            yield delta        # 结束后再读取引用（此时后台写历史已完成）        sources = [            {                "id": getattr(n.node, "node_id", None),                "text": n.node.get_content()[:200],                 "metadata": getattr(n.node, "metadata", {}),            }            for n in (resp.sources[0].raw_output.source_nodes if resp.sources else [])        ]        yield {"type": "sources", "content": sources}