多语言AI Agent的构建：跨语言理解与任务执行

引言

在人工智能技术快速发展的今天，AI Agent（智能代理）正逐渐成为连接人类与数字世界的重要桥梁。然而，随着全球化进程的加速，传统的单语言AI Agent已经无法满足日益增长的跨语言交互需求。想象一下，一个能够理解并执行来自全球各地用户请求的智能助手，无论用户使用的是中文、英文、阿拉伯语还是斯瓦希里语，都能提供一致且高质量的服务。这正是多语言AI Agent的魅力所在。

在这篇技术博客中，我们将深入探讨多语言AI Agent的构建原理、技术实现和实际应用。作为一名在AI领域拥有超过15年经验的从业者，我将带你穿越技术的迷雾，用清晰的语言和生动的例子，揭示多语言AI Agent背后的奥秘。

1. 核心概念

1.1 什么是多语言AI Agent？

多语言AI Agent是一种能够理解、处理和执行多种语言指令的智能系统。与传统的单语言AI Agent相比，它具有跨语言理解能力、多语言任务执行能力和语言间知识迁移能力。

在深入探讨之前，让我们先明确几个核心概念：

• AI Agent (智能代理)：一种能够感知环境、做出决策并执行行动的智能系统。它通常具有自主性、反应性、主动性和社交能力。
• 多语言理解 (Multilingual Understanding)：系统能够理解多种自然语言输入的能力。
• 跨语言迁移 (Cross-lingual Transfer)：将在一种语言上学到的知识和能力应用到其他语言上的过程。
• 多语言任务执行 (Multilingual Task Execution)：系统能够根据不同语言的指令执行相应任务的能力。

1.2 问题背景

随着全球化的深入发展，企业和个人的跨语言交互需求日益增长。根据联合国教科文组织的数据，世界上有超过7000种语言，而互联网上的内容虽然以英语为主，但其他语言的内容正在快速增长。然而，大多数现有的AI系统仍然主要支持少数几种主流语言，这导致了严重的"语言鸿沟"。

在商业领域，一家跨国公司可能需要为不同国家的客户提供本地化的客服支持；在教育领域，一个在线学习平台可能需要为使用不同语言的学生提供个性化的学习体验；在医疗领域，一个诊断系统可能需要理解来自不同国家患者的症状描述。这些场景都迫切需要高效的多语言AI Agent。

1.3 问题描述

构建多语言AI Agent面临着几个核心挑战：

1. 语言多样性：不同语言在语法、语义、文化背景等方面存在巨大差异。
2. 资源不平衡：主流语言（如英语、中文）拥有大量的标注数据和语料资源，而许多小众语言则资源匮乏。
3. 语义一致性：确保同一概念在不同语言中的理解和表达保持一致。
4. 文化适应性：理解和尊重不同语言背后的文化差异，避免文化误解。
5. 实时性能：在保持多语言能力的同时，确保系统的响应速度和性能。

1.4 问题解决

解决这些问题需要多学科的交叉融合，包括自然语言处理、机器学习、知识表示、认知科学等。近年来，随着深度学习技术的发展，特别是Transformer架构和预训练语言模型的出现，多语言AI Agent的构建取得了重大突破。

在接下来的章节中，我们将详细探讨如何从技术上构建一个高效的多语言AI Agent，包括核心算法原理、数学模型、代码实现等。

1.5 边界与外延

在深入探讨多语言AI Agent之前，我们需要明确其边界和外延：

核心边界：

• 多语言理解与生成
• 跨语言知识迁移
• 多语言任务规划与执行

相关外延：

• 文化智能 (Cultural Intelligence)
• 方言处理
• 代码切换 (Code-switching) 处理
• 低资源语言支持

2. 概念结构与核心要素组成

2.1 多语言AI Agent的核心架构

一个典型的多语言AI Agent通常由以下几个核心组件组成：

1. 多语言感知模块：负责接收和理解不同语言的输入。
2. 跨语言知识引擎：存储和管理多语言知识，支持跨语言知识推理。
3. 多语言任务规划器：根据理解的意图，规划多语言环境下的任务执行策略。
4. 多语言执行模块：执行具体的任务，并生成多语言的输出。
5. 反馈学习模块：根据用户反馈和执行结果，持续改进系统性能。

让我们用Mermaid图来展示这个架构：

2.2 核心要素详解

2.2.1 多语言感知模块

多语言感知模块是AI Agent的"眼睛和耳朵"，负责将不同语言的输入转换为系统可以理解的语义表示。这个模块通常包含以下子组件：

• 语言识别器：自动识别输入文本或语音的语言类型。
• 多语言分词器：针对不同语言特点进行文本分词处理。
• 多语言语义编码器：将不同语言的文本转换为共享的语义向量空间。

2.2.2 跨语言知识引擎

跨语言知识引擎是AI Agent的"大脑"，负责存储和管理多语言知识，支持跨语言的知识检索和推理。核心功能包括：

• 多语言知识图谱：以图结构存储多语言实体、概念和关系。
• 跨语言语义对齐：建立不同语言中相同或相似概念的映射关系。
• 多语言知识推理：基于多语言知识进行逻辑推理和常识推断。

2.2.3 多语言任务规划器

多语言任务规划器是AI Agent的"指挥员"，负责根据用户意图和当前环境，规划合适的任务执行策略。关键功能包括：

• 意图理解与分类：理解用户在不同语言下表达的真实意图。
• 任务分解与调度：将复杂任务分解为多个子任务，并合理调度执行。
• 语言自适应策略：根据用户语言和上下文，选择最合适的响应语言和方式。

2.2.4 多语言执行模块

多语言执行模块是AI Agent的"手脚"，负责具体执行规划好的任务，并生成多语言的响应。主要功能包括：

• 工具调用与API集成：调用外部工具和API完成具体任务。
• 多语言响应生成：根据用户语言和上下文，生成自然流畅的多语言响应。
• 多媒体输出适配：根据需要生成文本、语音、图像等多种形式的输出。

2.2.5 反馈学习模块

反馈学习模块是AI Agent的"进化引擎"，负责根据用户反馈和执行结果，持续改进系统性能。核心机制包括：

• 用户反馈收集与分析：收集和分析用户的显式和隐式反馈。
• 在线学习与模型更新：基于反馈数据进行在线学习，更新系统模型。
• 性能监控与评估：持续监控系统在不同语言和任务上的表现。

3. 概念之间的关系

3.1 核心属性维度对比

为了更清晰地理解多语言AI Agent与相关概念的区别，我们从几个核心属性维度进行对比：

概念类型	语言支持	知识迁移能力	任务执行范围	文化适应性	技术复杂度
单语言AI Agent	单一语言	无	特定语言任务	低	低
多语言翻译系统	多种语言	有限	仅翻译	中	中
多语言理解系统	多种语言	中	理解任务	中	中高
多语言AI Agent	多种语言	高	广泛任务	高	高

3.2 概念联系的ER实体关系图

让我们用ER图来展示多语言AI Agent各核心概念之间的关系：

3.3 交互关系图

现在，让我们用交互关系图来展示多语言AI Agent各组件之间的动态交互：

4. 核心算法原理 & 具体操作步骤

4.1 多语言语义表示算法

多语言语义表示是构建多语言AI Agent的基础。其核心目标是将不同语言的文本映射到一个共享的语义向量空间中，使得语义相似的文本在这个空间中距离相近，无论它们使用的是哪种语言。

4.1.1 基于多语言预训练模型的方法

近年来，基于Transformer架构的多语言预训练模型取得了显著成功。代表性的模型包括mBERT、XLM-R、LaBSE等。这些模型通过在大量多语言语料上进行预训练，学习到了跨语言的语义表示能力。

让我们以mBERT为例，了解其工作原理。mBERT是在BERT模型的基础上，使用104种语言的维基百科数据进行预训练得到的。它共享了一套词表和模型参数，通过掩码语言建模（Masked Language Modeling, MLM）任务进行预训练。

具体操作步骤：

1. 多语言词表构建：使用Byte-Pair Encoding (BPE)算法从多语言语料中构建共享词表。
2. 掩码语言建模：随机遮蔽输入序列中的部分token，训练模型预测被遮蔽的token。
3. 跨语言微调：在标注的跨语言任务数据上进行微调，进一步提升模型的跨语言能力。

让我们用Python代码来演示如何使用多语言预训练模型进行语义表示：

from transformers import AutoTokenizer, AutoModel
import torch

class MultilingualSemanticEncoder:
    def __init__(self, model_name="xlm-roberta-base"):
        """
        初始化多语言语义编码器
        
        参数:
            model_name (str): 预训练模型名称，默认为"xlm-roberta-base"
        """
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model.to(self.device)
        print(f"模型已加载到设备: {self.device}")
    
    def encode(self, texts, max_length=128):
        """
        编码文本为语义向量
        
        参数:
            texts (list): 输入文本列表
            max_length (int): 最大序列长度
            
        返回:
            torch.Tensor: 语义向量张量，形状为 (batch_size, hidden_size)
        """
        # 分词和编码
        inputs = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=max_length,
            return_tensors="pt"
        )
        
        # 将输入移到指定设备
        inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        # 前向传播，不计算梯度
        with torch.no_grad():
            outputs = self.model(**inputs)
        
        # 使用 <[BOS_never_used_51bce0c785ca2f68081bfa7d91973934]> token 作为句子表示
        sentence_embeddings = outputs.last_hidden_state[:, 0, :]
        
        # L2 归一化
        sentence_embeddings = torch.nn.functional.normalize(sentence_embeddings, p=2, dim=1)
        
        return sentence_embeddings
    
    def compute_similarity(self, text1, text2):
        """
        计算两段文本的语义相似度
        
        参数:
            text1 (str): 第一段文本
            text2 (str): 第二段文本
            
        返回:
            float: 相似度分数，范围 [0, 1]
        """
        embeddings = self.encode([text1, text2])
        similarity = torch.cosine_similarity(embeddings[0:1], embeddings[1:2]).item()
        return similarity

# 使用示例
if __name__ == "__main__":
    # 初始化编码器
    encoder = MultilingualSemanticEncoder()
    
    # 测试不同语言的语义相似度
    test_pairs = [
        ("你好，世界", "Hello, world"),
        ("我爱机器学习", "I love machine learning"),
        ("今天天气真好", "What a beautiful day today"),
        ("人工智能正在改变世界", "AI is transforming the world"),
        ("猫坐在垫子上", "The cat sat on the mat"),
        ("猫坐在垫子上", "狗在公园里奔跑")  # 负例
    ]
    
    for text1, text2 in test_pairs:
        similarity = encoder.compute_similarity(text1, text2)
        print(f"'{text1}' 与 '{text2}' 的语义相似度: {similarity:.4f}")

4.1.2 跨语言语义对齐算法

除了使用预训练模型外，我们还可以通过跨语言语义对齐算法，将不同语言的词向量或句子向量映射到共享空间中。一个经典的方法是基于对抗训练的跨语言词向量对齐。

算法步骤：

1. 源语言和目标语言词向量训练：分别在源语言和目标语言语料上训练单语言词向量。
2. 初始种子词典构建：使用少量已知的双语词典作为初始种子。
3. 线性变换学习：基于种子词典，学习一个线性变换矩阵，将源语言词向量映射到目标语言空间。
4. 对抗训练优化：使用生成对抗网络（GAN）进一步优化映射，使得判别器无法区分映射后的源语言向量和目标语言向量。
5. 迭代精炼：利用对齐后的词向量扩展种子词典，重复上述步骤，直到收敛。

让我们用Python实现一个简化版的跨语言词向量对齐算法：

import numpy as np
from sklearn.decomposition import PCA
from scipy.linalg import orthogonal_procrustes

class CrossLingualWordAligner:
    def __init__(self, source_embeddings, target_embeddings, source_word2id, target_word2id):
        """
        初始化跨语言词向量对齐器
        
        参数:
            source_embeddings (np.ndarray): 源语言词向量矩阵
            target_embeddings (np.ndarray): 目标语言词向量矩阵
            source_word2id (dict): 源语言词到ID的映射
            target_word2id (dict): 目标语言词到ID的映射
        """
        self.source_emb = source_embeddings
        self.target_emb = target_embeddings
        self.source_word2id = source_word2id
        self.target_word2id = target_word2id
        self.W = None  # 变换矩阵
        
    def preprocess_embeddings(self, embeddings, normalize=True, pca_remove=True, pca_components=2):
        """
        预处理词向量
        
        参数:
            embeddings (np.ndarray): 词向量矩阵
            normalize (bool): 是否归一化
            pca_remove (bool): 是否移除主成分
            pca_components (int): 要移除的主成分数量
            
        返回:
            np.ndarray: 预处理后的词向量
        """
        processed = embeddings.copy()
        
        # 归一化
        if normalize:
            processed = processed / np.linalg.norm(processed, axis=1, keepdims=True)
        
        # 移除主成分（一种常见的后处理技巧）
        if pca_remove and pca_components > 0:
            pca = PCA(n_components=pca_components)
            pca.fit(processed)
            for i in range(pca_components):
                principal_component = pca.components_[i:i+1]
                processed -= np.dot(processed, principal_component.T) * principal_component
        
        # 再次归一化
        if normalize:
            processed = processed / np.linalg.norm(processed, axis=1, keepdims=True)
        
        return processed
    
    def learn_mapping(self, seed_dict, source_preprocess=True, target_preprocess=True):
        """
        基于种子词典学习线性变换矩阵
        
        参数:
            seed_dict (dict): 源语言词到目标语言词的映射词典
            source_preprocess (bool): 是否预处理源语言词向量
            target_preprocess (bool): 是否预处理目标语言词向量
            
        返回:
            np.ndarray: 学习到的变换矩阵 W
        """
        # 收集种子词对的词向量
        source_vecs = []
        target_vecs = []
        
        for src_word, tgt_word in seed_dict.items():
            if src_word in self.source_word2id and tgt_word in self.target_word2id:
                src_id = self.source_word2id[src_word]
                tgt_id = self.target_word2id[tgt_word]
                source_vecs.append(self.source_emb[src_id])
                target_vecs.append(self.target_emb[tgt_id])
        
        if not source_vecs:
            raise ValueError("没有有效的种子词对！")
        
        # 转换为numpy数组
        source_vecs = np.array(source_vecs)
        target_vecs = np.array(target_vecs)
        
        # 预处理词向量
        if source_preprocess:
            source_vecs = self.preprocess_embeddings(source_vecs)
        if target_preprocess:
            target_vecs = self.preprocess_embeddings(target_vecs)
        
        # 使用正交 Procrustes 方法学习变换矩阵
        # 目标是找到正交矩阵 W，使得 ||X * W - Y||^2 最小
        self.W, _ = orthogonal_procrustes(source_vecs, target_vecs)
        
        return self.W
    
    def transform_source_embeddings(self, preprocess=True):
        """
        变换源语言词向量到目标语言空间
        
        参数:
            preprocess (bool): 是否在变换前后进行预处理
            
        返回:
            np.ndarray: 变换后的源语言词向量
        """
        if self.W is None:
            raise ValueError("请先学习变换矩阵！")
        
        source_emb = self.source_emb.copy()
        
        # 预处理
        if preprocess:
            source_emb = self.preprocess_embeddings(source_emb)
        
        # 应用变换
        transformed = np.dot(source_emb, self.W)
        
        # 再次预处理
        if preprocess:
            transformed = self.preprocess_embeddings(transformed)
        
        return transformed
    
    def get_translation_candidates(self, source_word, k=5, preprocess=True):
        """
        获取源语言词的目标语言翻译候选
        
        参数:
            source_word (str): 源语言词
            k (int): 返回的候选数量
            preprocess (bool): 是否使用预处理
            
        返回:
            list: 翻译候选列表，按相似度降序排列
        """
        if source_word not in self.source_word2id:
            return []
        
        # 变换源语言词向量
        transformed_source = self.transform_source_embeddings(preprocess=preprocess)
        target_emb = self.target_emb.copy()
        
        # 预处理目标语言词向量
        if preprocess:
            target_emb = self.preprocess_embeddings(target_emb)
        
        # 获取源词向量
        src_id = self.source_word2id[source_word]
        src_vec = transformed_source[src_id:src_id+1]
        
        # 计算与所有目标语言词的余弦相似度
        similarities = np.dot(target_emb, src_vec.T).flatten()
        
        # 获取最相似的k个词
        top_k_indices = np.argsort(similarities)[::-1][:k]
        
        # 构建候选列表
        candidates = []
        for idx in top_k_indices:
            word = list(self.target_word2id.keys())[list(self.target_word2id.values()).index(idx)]
            candidates.append((word, similarities[idx]))
        
        return candidates

# 使用示例
if __name__ == "__main__":
    # 注意：这里我们使用模拟数据，实际应用中应该加载真实的词向量
    np.random.seed(42)
    vocab_size = 1000
    embedding_dim = 300
    
    # 模拟源语言和目标语言词向量
    source_embeddings = np.random.randn(vocab_size, embedding_dim)
    target_embeddings = np.random.randn(vocab_size, embedding_dim)
    
    # 模拟词到ID的映射
    source_words = [f"src_word_{i}" for i in range(vocab_size)]
    target_words = [f"tgt_word_{i}" for i in range(vocab_size)]
    
    source_word2id = {word: i for i, word in enumerate(source_words)}
    target_word2id = {word: i for i, word in enumerate(target_words)}
    
    # 初始化对齐器
    aligner = CrossLingualWordAligner(
        source_embeddings, 
        target_embeddings, 
        source_word2id, 
        target_word2id
    )
    
    # 模拟种子词典
    seed_dict = {
        f"src_word_{i}": f"tgt_word_{i}" 
        for i in range(min(100, vocab_size))
    }
    
    # 学习变换矩阵
    W = aligner.learn_mapping(seed_dict)
    print(f"学习到的变换矩阵形状: {W.shape}")
    
    # 测试翻译候选
    test_word = "src_word_50"
    candidates = aligner.get_translation_candidates(test_word, k=5)
    print(f"\n'{test_word}' 的翻译候选:")
    for word, score in candidates:
        print(f"  {word}: {score:.4f}")

4.2 多语言任务规划算法

多语言任务规划是指根据用户的多语言输入，理解其意图，并规划出合适的任务执行序列。这通常涉及意图识别、实体提取、任务分解和调度等步骤。

4.2.1 多语言意图识别算法

意图识别是任务规划的第一步，其目标是识别用户输入的真实意图。对于多语言场景，我们需要模型能够识别不同语言下表达的相同意图。

基于预训练模型的多语言意图识别方法：

1. 多语言预训练模型微调：使用标注的多语言意图识别数据微调预训练模型。
2. 跨语言数据增强：通过回译等方法生成更多的训练数据。
3. 零样本/少样本学习：利用大规模预训练模型的零样本/少样本学习能力，处理低资源语言。

让我们用Python代码实现一个简单的多语言意图识别器：

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import numpy as np

class MultilingualIntentRecognizer:
    def __init__(self, model_name, intent_labels, id2label=None, label2id=None):
        """
        初始化多语言意图识别器
        
        参数:
            model_name (str): 预训练模型名称或路径
            intent_labels (list): 意图标签列表
            id2label (dict): ID到标签的映射（可选）
            label2id (dict): 标签到ID的映射（可选）
        """
        self.intent_labels = intent_labels
        
        # 构建ID和标签的映射
        if id2label is None:
            self.id2label = {i: label for i, label in enumerate(intent_labels)}
        else:
            self.id2label = id2label
            
        if label2id is None:
            self.label2id = {label: i for i, label in enumerate(intent_labels)}
        else:
            self.label2id = label2id
        
        # 加载模型和分词器
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            model_name,
            num_labels=len(intent_labels),
            id2label=self.id2label,
            label2id=self.label2id
        )
        
        # 设置设备
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model.to(self.device)
        print(f"模型已加载到设备: {self.device}")
    
    def predict_intent(self, texts, top_k=1):
        """
        预测文本的意图
        
        参数:
            texts (str or list): 输入文本或文本列表
            top_k (int): 返回的top-k意图
            
        返回:
            list: 预测结果列表，每个元素为意图和置信度的列表
        """
        # 处理单个文本的情况
        if isinstance(texts, str):
            texts = [texts]
        
        # 分词和编码
        inputs = self.tokenizer(
            texts,
            padding=True,
            truncation=True,
            max_length=128,
            return_tensors="pt"
        )
        
        # 将输入移到指定设备
        inputs = {k: v.to(self.device) for k, v in inputs.items()}
        
        # 前向传播，不计算梯度
        self.model.eval()
        with torch.no_grad():
            outputs = self.model(**inputs)
            logits = outputs.logits
            probabilities = torch.softmax(logits, dim=-1)
        
        # 获取top-k预测结果
        top_probs, top_indices = torch.topk(probabilities, top_k, dim=-1)
        
        # 构建结果列表
        results = []
        for i in range(len(texts)):
            intent_results = []
            for j in range(top_k):
                intent = self.id2label[top_indices[i, j].item()]
                confidence = top_probs[i, j].item()
                intent_results.append((intent, confidence))
            results.append(intent_results)
        
        return results
    
    def fine_tune(self, train_texts, train_labels, val_texts=None, val_labels=None, 
                  epochs=3, batch_size=16, learning_rate=2e-5):
        """
        微调模型
        
        参数:
            train_texts (list): 训练文本列表
            train_labels (list): 训练标签列表
            val_texts (list): 验证文本列表（可选）
            val_labels (list): 验证标签列表（可选）
            epochs (int): 训练轮数
            batch_size (int): 批次大小
            learning_rate (float): 学习率
        """
        from torch.utils.data import Dataset, DataLoader
        from transformers import AdamW
        
        # 自定义数据集类
        class IntentDataset(Dataset):
            def __init__(self, texts, labels, tokenizer, max_length=128):
                self.texts = texts
                self.labels = labels
                self.tokenizer = tokenizer
                self.max_length = max_length
            
            def __len__(self):
                return len(self.texts)
            
            def __getitem__(self, idx):
                text = self.texts[idx]
                label = self.labels[idx]
                
                encoding = self.tokenizer(
                    text,
                    truncation=True,
                    max_length=self.max_length,
                    padding="max_length",
                    return_tensors="pt"
                )
                
                return {
                    "input_ids": encoding["input_ids"].flatten(),
                    "attention_mask": encoding["attention_mask"].flatten(),
                    "labels": torch.tensor(label, dtype=torch.long)
                }
        
        # 将标签转换为ID
        train_label_ids = [self.label2id[label] for label in train_labels]
        
        # 创建训练数据集和数据加载器
        train_dataset = IntentDataset(train_texts, train_label_ids, self.tokenizer)
        train_dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
        
        # 如果有验证数据，创建验证数据集和数据加载器
        val_dataloader = None
        if val_texts and val_labels:
            val_label_ids = [self.label2id[label] for label in val_labels]
            val_dataset = IntentDataset(val_texts, val_label_ids, self.tokenizer)
            val_dataloader = DataLoader(val_dataset, batch_size=batch_size)
        
        # 设置优化器
        optimizer = AdamW(self.model.parameters(), lr=learning_rate)
        
        # 训练模型
        self.model.train()
        for epoch in range(epochs):
            print(f"Epoch {epoch + 1}/{epochs}")
            total_loss = 0
            
            for batch in train_dataloader:
                # 将批次数据移到指定设备
                batch = {k: v.to(self.device) for k, v in batch.items()}
                
                # 前向传播
                outputs = self.model(**batch)
                loss = outputs.loss
                total_loss += loss.item()
                
                # 反向传播和优化
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()
            
            avg_train_loss = total_loss / len(train_dataloader)
            print(f"Average training loss: {avg_train_loss:.4f}")
            
            # 如果有验证数据，在验证集上评估
            if val_dataloader:
                self.model.eval()
                val_loss = 0
                correct = 0
                total = 0
                
                with torch.no_grad():
                    for batch in val_dataloader:
                        batch = {k: v.to(self.device) for k, v in batch.items()}
                        outputs = self.model(**batch)
                        loss = outputs.loss
                        val_loss += loss.item()
                        
                        logits = outputs.logits
                        _, predicted = torch.max(logits, dim=1)
                        total += batch["labels"].size(0)
                        correct += (predicted == batch["labels"]).sum().item()
                
                avg_val_loss = val_loss / len(val_dataloader)
                val_accuracy = correct / total
                print(f"Validation loss: {avg_val_loss:.4f}, accuracy: {val_accuracy:.4f}")
                
                self.model.train()
        
        print("微调完成！")

# 使用示例
if __name__ == "__main__":
    # 定义意图标签
    intent_labels = [
        "greeting",
        "ask_weather",
        "book_flight",
        "book_hotel",
        "order_food",
        "unknown"
    ]
    
    # 初始化意图识别器（使用一个小型的多语言预训练模型）
    recognizer = MultilingualIntentRecognizer(
        model_name="bert-base-multilingual-cased",
        intent_labels=intent_labels
    )
    
    # 模拟训练数据（实际应用中应该使用真实的标注数据）
    train_texts = [
        "你好", "Hello", "Hola",  # greeting
        "今天天气怎么样？", "What's the weather like today?", "¿Cómo está el clima hoy?",  # ask_weather
        "我想订一张去北京的机票", "I want to book a flight to Beijing", "Quiero reservar un vuelo a Pekín",  # book_flight
        "帮我订一个酒店房间", "Help me book a hotel room", "Ayúdame a reservar una habitación de hotel",  # book_hotel
        "我想点一份披萨", "I want to order a pizza", "Quiero pedir una pizza",  # order_food
        "你能帮我修一下电视吗？", "Can you help me fix the TV?", "¿Puedes ayudarme a arreglar la televisión?"  # unknown
    ]
    
    train_labels = [
        "greeting", "greeting", "greeting",
        "ask_weather", "ask_weather", "ask_weather",
        "book_flight", "book_flight", "book_flight",
        "book_hotel", "book_hotel", "book_hotel",
        "order_food", "order_food", "order_food",
        "unknown", "unknown", "unknown"
    ]
    
    # 这里我们不会实际运行微调，因为它需要较长时间
    # 但在实际应用中，你应该取消下面这行的注释
    # recognizer.fine_tune(train_texts, train_labels, epochs=3)
    
    # 测试预测
    test_texts = [
        "你好，我想了解一下明天的天气",
        "Hello, can you help me book a flight to Paris?",
        "Hola, quiero pedir una comida para llevar",
        "Bonjour, je voudrais réserver une chambre d'hôtel"
    ]
    
    results = recognizer.predict_intent(test_texts, top_k=3)
    
    for text, intent_results in zip(test_texts, results):
        print(f"\n文本: {text}")
        for intent, confidence in intent_results:
            print(f"  意图: {intent}, 置信度: {confidence:.4f}")

4.2.2 基于层次任务网络的多语言任务规划

在理解了用户意图后，我们需要规划具体的任务执行步骤。层次任务网络（Hierarchical Task Network, HTN）是一种常用的任务规划方法，它将复杂任务分解为更简单的子任务，直到达到可执行的原子任务。

对于多语言场景，我们需要考虑以下几点：

1. 多语言参数提取：从用户输入中提取任务所需的参数，支持多语言实体识别。
2. 语言自适应任务选择：根据用户语言选择合适的任务执行方式和工具。
3. 多语言响应模板：为不同语言准备相应的响应模板。

让我们用Python代码实现一个简单的基于HTN的多语言任务规划器：

from typing import List, Dict, Any, Optional, Tuple
import re

class MultilingualTask:
    def __init__(self, name: str, language: str = "en"):
        """
        初始化多语言任务
        
        参数:
            name (str): 任务名称
            language (str): 任务语言
        """
        self.name = name
        self.language = language
        self.parameters = {}
    
    def set_parameter(self, key: str, value: Any):
        """设置任务参数"""
        self.parameters[key] = value
    
    def get_parameter(self, key: str) -> Any:
        """获取任务参数"""
        return self.parameters.get(key)
    
    def __repr__(self) -> str:
        return f"MultilingualTask(name={self.name}, language={self.language}, parameters={self.parameters})"


class MultilingualHTNPlanner:
    def __init__(self, language_resources: Dict[str, Dict[str, Any]]):
        """
        初始化多语言HTN规划器
        
        参数:
            language_resources (dict): 语言资源字典，包含不同语言的响应模板、实体模式等
        """
        self.language_resources = language_resources
        self.tasks = {}
        self.methods = {}
        self.operators = {}
    
    def register_task(self, task_name: str, task_type: str = "compound"):
        """
        注册任务
        
        参数:
            task_name (str): 任务名称
            task_type (str): 任务类型，"compound"（复合任务）或 "primitive"（原子任务）
        """
        self.tasks[task_name] = {
            "type": task_type,
            "methods": [] if task_type == "compound" else None
        }
    
    def register_method(self, task_name: str, method_name: str, 
                       preconditions: List[Dict[str, Any]], 
                       subtasks: List[Tuple[str, Dict[str, Any]]]):
        """
        注册复合任务的分解方法
        
        参数:
            task_name (str): 复合任务名称
            method_name (str): 方法名称
            preconditions (list): 前提条件列表
            subtasks (list): 子任务列表，每个元素为(子任务名, 参数绑定)
        """
        if task_name not in self.tasks or self.tasks[task_name]["type"] != "compound":
            raise ValueError(f"Task {task_name} is not a registered compound task")
        
        self.tasks[task_name]["methods"].append({
            "name": method_name,
            "preconditions": preconditions,
            "subtasks": subtasks
        })
    
    def register_operator(self, operator_name: str, 
                         preconditions: List[Dict[str, Any]],
                         effects: List[Dict[str, Any]],
                         response_templates: Dict[str, str]):
        """
        注册原子任务的操作符
        
        参数:
            operator_name (str): 操作符名称
            preconditions (list): 前提条件列表
            effects (list): 效果列表
            response_templates (dict): 不同语言的响应模板
        """
        self.operators[operator_name] = {
            "preconditions": preconditions,
            "effects": effects,
            "response_templates": response_templates
        }
    
    def extract_parameters(self, text: str, language: str, 
                          parameter_patterns: Dict[str, str]) -> Dict[str, str]:
        """
        从文本中提取参数
        
        参数:
            text (str): 输入文本
            language (str): 语言
            parameter_patterns (dict): 参数模式字典
        
        返回:
            dict: 提取的参数字典
        """
        parameters = {}
        
        # 获取指定语言的参数模式，如果不存在则使用英语
        lang_patterns = parameter_patterns.get(language, parameter_patterns.get("en", {}))
        
        for param_name, pattern in lang_patterns.items():
            match = re.search(pattern, text)
            if match:
                parameters[param_name] = match.group(1)
        
        return parameters
    
    def check_preconditions(self, preconditions: List[Dict[str, Any]], 
                           state: Dict[str, Any], task: MultilingualTask) -> bool:
        """
        检查前提条件是否满足
        
        参数:
            preconditions (list): 前提条件列表
            state (dict): 当前状态
            task (MultilingualTask): 当前任务
        
        返回:
            bool: 前提条件是否满足
        """
        for condition in preconditions:
            # 简化的条件检查，实际应用中可能需要更复杂的逻辑
            cond_type = condition.get("type")
            
            if cond_type == "parameter_exists":
                param_name = condition.get("parameter")
                if task.get_parameter(param_name) is None:
                    return False
            
            elif cond_type == "state_check":
                key = condition.get("key")
                expected_value = condition.get("value")
                if state.get(key) != expected_value:
                    return False
        
        return True
    
    def apply_effects(self, effects: List[Dict[str, Any]], state: Dict[str, Any]) -> Dict[str, Any]:
        """
        应用效果到状态
        
        参数:
            effects (list): 效果列表
            state (dict): 当前状态
        
        返回:
            dict: 更新后的状态
        """
        new_state = state.copy()
        
        for effect in effects:
            effect_type = effect.get("type")
            
            if effect_type == "set_state":
                key = effect.get("key")
                value = effect.get("value")
                new_state[key] = value
        
        return new_state
    
    def plan(self, task: MultilingualTask, state: Dict[str, Any]) -> Tuple[List[MultilingualTask], Dict[str, Any]]:
        """
        规划任务执行序列
        
        参数:
            task (MultilingualTask): 初始任务
            state (dict): 初始状态
        
        返回:
            tuple: (任务执行序列, 最终状态)
        """
        plan = []
        current_state = state.copy()
        task_stack = [task]
        
        while task_stack:
            current_task = task_stack.pop()
            
            # 如果是原子任务，直接添加到计划中
            if current_task.name in self.operators:
                operator = self.operators[current_task.name]
                
                # 检查前提条件
                if not self.check_preconditions(operator["preconditions"], current_state, current_task):
                    raise ValueError(f"Preconditions not met for operator {current_task.name}")
                
                # 添加到计划
                plan.append(current_task)
                
                # 应用效果
                current_state = self.apply_effects(operator["effects"], current_state)
            
            # 如果是复合任务，尝试分解
            elif current_task.name in self.tasks and self.tasks[current_task.name]["type"] == "compound":
                task_info = self.tasks[current_task.name]
                method_found = False
                
                for method in task_info["methods"]:
                    # 检查方法的前提条件
                    if self.check_preconditions(method["preconditions"], current_state, current_task):
                        # 方法适用，将子任务压入栈中（注意顺序，因为栈是LIFO）
                        for subtask_name, param_bindings in reversed(method["subtasks"]):
                            # 创建子任务
                            subtask = MultilingualTask(subtask_name, current_task.language)
                            
                            # 绑定参数
                            for param_name, binding in param_bindings.items():
                                # 支持从当前任务参数中获取值
                                if binding.startswith("$"):
                                    source_param = binding[1:]
                                    value = current_task.get_parameter(source_param)
                                    subtask.set_parameter(param_name, value)
                                else:
                                    subtask.set_parameter(param_name, binding)
                            
                            task_stack.append(subtask)
                        
                        method_found = True
                        break
                
                if not method_found:
                    raise ValueError(f"No applicable method found for task {current_task.name}")
            
            else:
                raise ValueError(f"Unknown task: {current_task.name}")
        
        return plan, current_state
    
    def generate_response(self, plan: