构建多语言支持的 Agent 系统：从概念到实现

副标题：探索多智能体协作、语言无关设计与实际应用场景

---

第一部分：引言与基础

1. 引人注目的标题

构建多语言支持的 Agent 系统：从概念到实现

副标题：探索多智能体协作、语言无关设计与实际应用场景

2. 摘要/引言

问题陈述

在当今全球化和数字化的时代，我们面临着两个主要挑战：

1. 语言障碍：随着全球化进程的加速，不同语言之间的交流障碍成为了信息共享和协作的主要瓶颈。
2. 复杂任务处理：现代应用场景往往涉及复杂的多步骤任务，单一的AI模型难以高效且可靠地完成所有工作。

传统的单Agent系统虽然可以处理特定语言下的特定任务，但在跨语言协作和复杂任务分解方面存在明显不足。此外，大多数现有系统要么专注于单语言环境，要么在多语言支持上做得不够完善。

核心方案

本文提出的解决方案是构建一个多语言支持的多Agent系统，该系统具有以下特点：

1. 多Agent协作架构：通过将复杂任务分解为多个子任务，由不同专长的Agent协作完成。
2. 语言无关设计：系统核心逻辑与语言处理分离，支持灵活添加新语言。
3. 模块化结构：各组件高度解耦，便于维护和扩展。

我们将使用Python作为主要开发语言，结合LangChain框架和现代大语言模型（如GPT-4、Claude等）来实现这一系统。

主要成果/价值

读完本文后，你将能够：

1. 理解多Agent系统的核心概念和设计原则
2. 掌握构建语言无关的AI系统的方法
3. 学会使用LangChain等工具实现Agent协作
4. 了解多语言处理的最佳实践和常见挑战
5. 获得一个可直接运行的多语言多Agent系统原型

文章导览

本文将按照以下结构展开：

• 第一部分：介绍背景、目标读者和文章结构
• 第二部分：深入探讨多Agent系统的理论基础、架构设计和核心概念
• 第三部分：详细讲解系统的实现步骤、代码解析和关键技术
• 第四部分：展示系统运行结果、性能优化和未来发展方向
• 第五部分：总结全文并提供参考资料和附录

3. 目标读者与前置知识

目标读者

本文适合以下读者：

• 有一定Python编程基础的开发者
• 对大语言模型和AI应用感兴趣的技术人员
• 需要构建跨语言AI系统的产品经理和架构师
• 希望了解多Agent协作系统的研究人员

前置知识

阅读本文前，建议你具备以下基础知识：

• Python 3.8+编程经验
• 基本的RESTful API概念
• 对大语言模型（如GPT）有基本了解
• 环境变量和配置管理的基本知识
• 版本控制（Git）的基本使用

虽然我们会详细解释相关概念，但具备上述知识将帮助你更好地理解和实践本文内容。

4. 文章目录

1. 第一部分：引言与基础

   • 引人注目的标题
   • 摘要/引言
   • 目标读者与前置知识
   • 文章目录

2. 第二部分：核心概念与理论基础

   • 问题背景与动机
   • 多Agent系统核心概念
   • 多语言处理基础
   • 系统架构设计原则
   • 相关技术栈介绍

3. 第三部分：环境准备与实现

   • 开发环境配置
   • 项目结构设计
   • 核心模块实现
   • Agent协作机制
   • 多语言支持集成

4. 第四部分：系统验证与扩展

   • 系统测试与验证
   • 性能优化策略
   • 常见问题与解决方案
   • 扩展功能与应用场景

5. 第五部分：总结与展望

   • 全文总结
   • 行业发展与未来趋势
   • 参考资料
   • 附录

---

第二部分：核心概念与理论基础

5. 问题背景与动机

全球化背景下的语言挑战

在当今互联的世界中，语言障碍仍然是信息交流和协作的主要障碍之一。根据联合国教科文组织的数据，世界上有超过7000种语言，其中只有少数几种被广泛使用于互联网和商业环境中。

这种语言多样性带来了以下挑战：

1. 信息获取不平等：大量高质量内容仅以少数语言提供，限制了非母语使用者的信息获取。
2. 跨文化协作困难：全球团队协作中，语言差异常常导致沟通效率低下和误解。
3. 本地化成本高昂：企业需要投入大量资源将产品和服务本地化到不同语言市场。

AI系统的演进：从单模型到多Agent

与此同时，人工智能系统正在经历从单一模型向多Agent协作的演进：

1. 第一代AI系统：基于规则的专家系统，处理特定领域的结构化任务。
2. 第二代AI系统：单一机器学习模型，如早期的文本分类器和图像识别系统。
3. 第三代AI系统：大型预训练模型，如GPT-4、Claude等，具备广泛的知识和多种能力。
4. 第四代AI系统：多Agent协作系统，通过多个专业化Agent的协作处理复杂任务。

这种演进背后的主要驱动力是：

• 复杂任务难以通过单一模型高效完成
• 专业化Agent在特定领域表现更优
• 模块化设计便于系统维护和扩展
• 并行处理能力提升整体效率

现有解决方案的局限性

尽管市场上已有一些多语言AI工具和Agent框架，但它们往往存在以下局限性：

1. 语言支持有限：许多系统仅支持主流语言，对小语种支持不足。
2. Agent间协作机制简单：缺乏复杂的任务分解和协调机制。
3. 扩展性差：添加新语言或新功能需要大量修改核心代码。
4. 性能瓶颈：在处理多语言复杂任务时响应时间过长。
5. 开发门槛高：需要深入了解多个领域的专业知识才能构建类似系统。

我们的解决方案动机

基于上述观察，我们设计了一个多语言支持的Agent系统，旨在：

1. 打破语言壁垒：提供灵活的多语言支持框架，便于扩展新语言。
2. 实现高效协作：设计强大的Agent协作机制，处理复杂任务。
3. 降低开发门槛：提供模块化、可配置的架构，便于快速定制和扩展。
4. 保证系统性能：优化关键路径，确保即使在复杂场景下也能保持良好性能。

接下来，我们将深入探讨构建这样一个系统所需的核心概念和理论基础。

6. 核心概念与理论基础

在开始实现之前，让我们先建立一些核心概念的共同理解。这些概念将贯穿全文，是我们设计和实现的基础。

Agent的定义与特征

Agent（智能体）是人工智能领域的一个核心概念，指的是能够感知环境、做出决策并采取行动的实体。在我们的上下文中，Agent是具有特定能力的软件模块。

一个典型的Agent具有以下特征：

1. 自主性：Agent能够在没有人类直接干预的情况下运行。
2. 反应性：Agent能够感知环境变化并做出相应反应。
3. 主动性：Agent不仅对环境做出反应，还能主动追求目标。
4. 社会性：Agent能够与其他Agent或人类进行交互和协作。

在我们的系统中，每个Agent都有其特定的专长和职责，例如翻译Agent、分析Agent、规划Agent等。

多Agent系统（MAS）

多Agent系统（Multi-Agent System, MAS）是由多个交互的Agent组成的系统。这些Agent可能是自主的，也可能是半自主的，它们通过协作、协调和协商来解决单个Agent难以解决的问题。

多Agent系统的优势包括：

1. 任务分解：复杂任务可以分解为多个子任务，由不同Agent并行处理。
2. 专业化：每个Agent可以专注于特定领域，提供更专业的服务。
3. 容错性：单个Agent的故障不会导致整个系统崩溃。
4. 可扩展性：可以方便地添加新Agent来增强系统能力。

多Agent系统的设计需要考虑以下关键问题：

• Agent间的通信机制
• 任务分配和协调策略
• 冲突解决机制
• 知识共享和管理

语言无关设计

语言无关设计（Language-Agnostic Design）是一种软件设计方法，旨在使系统的核心逻辑与特定自然语言处理分离，从而支持多种语言而无需修改核心代码。

实现语言无关设计的关键策略包括：

1. 抽象层设计：通过抽象层将语言处理与业务逻辑分离。
2. 资源外部化：将语言特定的资源（如提示词、错误消息）存储在外部文件中。
3. 插件架构：语言处理模块作为插件，可以根据需要加载或卸载。
4. 统一接口：为所有语言处理模块定义统一的接口标准。

在我们的系统中，语言无关设计不仅支持多语言处理，还使系统能够适应语言模型的更新和变化。

提示工程（Prompt Engineering）

提示工程（Prompt Engineering）是设计和优化输入提示以引导语言模型产生期望输出的过程。在多语言Agent系统中，提示工程尤为重要，因为：

1. 不同语言需要不同的提示策略：直接翻译提示可能无法在所有语言中产生相同效果。
2. Agent角色设定：需要精心设计提示来设定Agent的角色、能力和限制。
3. 任务指导：清晰的任务描述对于Agent正确理解和执行任务至关重要。
4. 输出格式控制：通过提示可以控制Agent的输出格式，便于后续处理。

有效的提示工程需要考虑以下要素：

• 角色设定（Role Setting）
• 任务描述（Task Description）
• 上下文信息（Context）
• 输出格式要求（Output Format）
• 示例（Few-Shot Examples）

向量数据库与语义搜索

向量数据库（Vector Database）是专门用于存储和查询高维向量数据的数据库系统。在多语言Agent系统中，向量数据库主要用于：

1. 语义搜索：将查询和文档转换为向量，通过相似度计算实现语义搜索。
2. 记忆管理：存储Agent的对话历史和重要信息，支持长期记忆。
3. 知识检索：实现检索增强生成（RAG），为Agent提供外部知识。

常用的向量数据库包括：

• Chroma（轻量级，易于集成）
• Pinecone（云服务，高性能）
• Weaviate（开源，功能丰富）
• Milvus（开源，企业级）

向量数据库的核心概念包括：

• 嵌入（Embeddings）：将文本转换为向量表示
• 相似度搜索（Similarity Search）：计算向量间的相似度
• 索引（Indexing）：优化向量数据的存储和检索
• 元数据过滤（Metadata Filtering）：结合元数据进行精确过滤

工作流与编排

工作流编排（Workflow Orchestration）是指定义、管理和协调复杂任务流程的过程。在多Agent系统中，工作流编排确保各个Agent按照正确的顺序和条件执行任务。

工作流编排的关键要素包括：

1. 任务定义：明确每个任务的输入、输出和执行条件。
2. 控制流：定义任务之间的依赖关系和执行顺序。
3. 数据流：管理任务之间的数据传递和转换。
4. 错误处理：定义任务失败时的处理策略。
5. 监控与日志：跟踪工作流执行状态和性能指标。

在我们的系统中，我们将使用LangChain的表达式语言（LCEL）和类似工具来实现工作流编排。

核心概念关系图

为了更好地理解这些核心概念之间的关系，让我们来看一个实体关系图：

这个ER图展示了我们系统中主要概念之间的关系：

• 多Agent系统包含多个Agent和工作流定义
• Agent执行任务，使用提示，并访问记忆
• 任务是工作流的一部分
• 语言模块为多Agent系统提供多语言支持
• 向量数据库存储嵌入，被多Agent系统使用

接下来，让我们深入探讨多语言处理的基础知识。

7. 多语言处理基础

多语言处理是构建我们系统的关键技术之一。让我们了解一些核心概念和挑战。

多语言自然语言处理（NLP）的挑战

多语言NLP面临着比单语言NLP更多的挑战：

1. 语言特性差异：

   • 不同语言有不同的语法结构
   • 书写系统差异（如从左到右 vs 从右到左）
   • 形态复杂性差异（如中文没有时态变化，而俄语有复杂的格变化）

2. 资源不平衡：

   • 高资源语言（如英语、中文）有大量标注数据
   • 低资源语言缺乏足够的训练数据和工具

3. 文化和语境差异：

   • 相同概念在不同文化中可能有不同含义
   • 幽默、隐喻等语言现象的跨语言理解困难

4. 翻译质量保证：

   • 保持语义一致性的同时适应目标语言表达习惯
   • 专业术语和特定领域概念的准确翻译

多语言表示学习

多语言表示学习（Multilingual Representation Learning）旨在学习能够跨语言共享的语义表示。这是实现语言无关系统的基础。

主要方法包括：

1. 多语言词嵌入：

   • MUSE（Multilingual Unsupervised and Supervised Embeddings）
   • VecMap（Vector Mapping）

2. 多语言预训练语言模型：

   • mBERT（Multilingual BERT）
   • XLM-R（Cross-lingual Language Model - RoBERTa）
   • mT5（Multilingual T5）
   • PaLM、GPT-4等现代大模型的多语言能力

这些模型通过在多种语言的语料上预训练，学习到了跨语言的通用表示，使得同一个模型可以处理多种语言的任务。

机器翻译技术

机器翻译是多语言系统的核心组件之一。现代机器翻译技术主要包括：

1. 神经机器翻译（NMT）：

   • 使用神经网络直接将源语言句子映射到目标语言
   • 端到端训练，无需人工设计特征

2. 大语言模型翻译：

   • 利用GPT-4、Claude等大模型的翻译能力
   • 可以处理复杂语境和专业术语
   • 支持少样本学习，适应特定领域需求

3. 翻译质量评估：

   • BLEU（Bilingual Evaluation Understudy）
   • chrF（Character n-gram F-score）
   • COMET（Cross-lingual Optimized Metric for Evaluation of Translation）

在我们的系统中，我们将结合使用专用翻译API和大语言模型的翻译能力，以获得最佳的多语言支持效果。

语言检测与识别

在处理多语言输入时，首先需要确定输入文本的语言。这就是语言检测（Language Detection）的任务。

语言检测的方法包括：

1. 基于字符n-gram的方法：

   • 统计不同语言中字符n-gram的分布
   • 通过比较输入文本与各语言模型的相似度确定语言

2. 基于机器学习的方法：

   • 使用分类器（如朴素贝叶斯、SVM）进行语言分类
   • 现代方法使用预训练语言模型进行语言检测

3. 常用工具：

   • langdetect（Python库）
   • fastText（Facebook开源的高效文本分类工具）
   • 大语言模型（如GPT-4也可用于语言检测）

在我们的系统中，语言检测是处理用户输入的第一步，它决定了后续使用哪些语言资源和处理策略。

8. 系统架构设计原则

在设计多语言支持的Agent系统时，我们遵循以下架构设计原则：

模块化设计

模块化是我们架构的核心原则。我们将系统分解为多个独立、可替换的模块：

1. 优势：

   • 便于独立开发和测试
   • 支持灵活的组件替换
   • 简化故障排查和维护
   • 允许并行开发

2. 实现方式：

   • 明确定义模块间的接口
   • 减少模块间的直接依赖
   • 使用依赖注入管理模块关系
   • 提供模块的热插拔能力

关注点分离

关注点分离（Separation of Concerns）是将不同功能区域的代码分离，使每个部分专注于自己的职责。

在我们的系统中，我们将分离以下关注点：

1. 语言处理与业务逻辑分离：

   • 语言特定代码放在独立模块
   • 核心业务逻辑不依赖特定语言

2. Agent能力与协调逻辑分离：

   • 各Agent专注于自己的任务
   • 单独的协调器负责Agent间的协作

3. 存储与计算分离：

   • 数据存储与计算逻辑解耦
   • 支持不同存储后端的替换

可扩展性设计

系统设计考虑了多个维度的可扩展性：

1. 语言扩展性：

   • 支持轻松添加新语言
   • 语言资源采用外部配置

2. Agent扩展性：

   • 支持动态注册新Agent
   • Agent能力声明机制

3. 性能扩展性：

   • 支持水平扩展
   • 关键路径支持缓存

容错与恢复

健壮的系统必须能够处理故障并优雅恢复：

1. 故障隔离：

   • 单个Agent的故障不应影响整个系统
   • 使用超时和断路器模式

2. 重试机制：

   • 对于暂时性错误提供自动重试
   • 实现指数退避策略

3. 状态管理：

   • 支持任务状态持久化
   • 提供断点续传能力

可观测性

系统应提供充分的可观测性，便于监控和调试：

1. 日志记录：

   • 结构化日志记录
   • 多级别日志控制

2. 指标收集：

   • 关键性能指标监控
   • 业务指标跟踪

3. 分布式追踪：

   • 请求链路追踪
   • 跨Agent调用跟踪

9. 相关技术栈介绍

在我们的实现中，我们将使用以下技术栈：

编程语言与环境

• Python 3.9+：主要开发语言，丰富的AI生态系统
• Poetry/Pipenv：依赖管理和虚拟环境
• Docker：容器化部署，确保环境一致性

Agent框架

• LangChain：构建LLM应用的框架，提供Agent抽象
• AutoGen：微软的多Agent对话框架
• Semantic Kernel：微软的轻量级SDK，集成AI和常规编程

大语言模型

• OpenAI GPT-4：强大的多语言理解和生成能力
• Claude 3：Anthropic的模型，长文档处理能力强
• Llama 2：Meta的开源模型，可本地部署

向量数据库

• Chroma：轻量级向量数据库，易于集成和开发
• FAISS：Facebook的相似性搜索库，高性能
• Pinecone：云向量数据库服务，可扩展

多语言处理

• fastText：高效的语言检测
• Deep Translator：多语言翻译库，支持多个翻译服务
• ** spaCy**：工业级NLP库，支持多语言

后端与API

• FastAPI：现代、快速的Web框架，用于构建API
• Pydantic：数据验证和设置管理
• WebSocket：实时通信，支持Agent间的流式交互

前端（可选）

• React/Vue：构建用户界面
• Tailwind CSS：实用优先的CSS框架
• shadcn/ui：基于Radix UI和Tailwind CSS的组件库

开发工具

• Git：版本控制
• Pre-commit：代码质量检查
• Black/Flake8：Python代码格式化和检查
• MkDocs：项目文档生成

现在我们已经建立了理论基础，接下来让我们开始系统的实现部分。

---

第三部分：环境准备与实现

10. 环境准备

在开始实现之前，我们需要设置开发环境。以下是详细的步骤。

系统要求

• 操作系统：Windows 10/11、macOS 10.15+ 或 Linux（推荐 Ubuntu 20.04+）
• Python版本：3.9 或更高
• 内存：至少 8GB RAM（推荐 16GB+）
• 存储空间：至少 10GB 可用空间
• 网络连接：用于下载模型和访问API服务

安装Python和依赖管理

首先，确保你安装了正确版本的Python：

# 检查Python版本
python --version
# 或
python3 --version

如果需要安装或更新Python，请访问Python官方网站下载适合你系统的版本。

接下来，我们将使用Poetry作为依赖管理工具。如果你还没有安装Poetry，可以使用以下命令安装：

# Linux/macOS
curl -sSL https://install.python-poetry.org | python3 -

# Windows (PowerShell)
(Invoke-WebRequest -Uri https://install.python-poetry.org -UseBasicParsing).Content | py -

安装完成后，验证安装：

poetry --version

创建项目结构

现在，让我们创建项目结构：

# 创建项目目录
mkdir multilingual-agent-system
cd multilingual-agent-system

# 使用Poetry初始化项目
poetry init

按照提示填写项目信息，或者直接按回车使用默认值。完成后，Poetry会创建一个pyproject.toml文件。

接下来，让我们创建项目的基本目录结构：

multilingual-agent-system/
├── pyproject.toml
├── README.md
├── .env.example
├── .gitignore
├── src/
│   └── multiagent_system/
│       ├── __init__.py
│       ├── agents/
│       │   ├── __init__.py
│       │   ├── base.py
│       │   ├── translator.py
│       │   ├── coordinator.py
│       │   └── specialist.py
│       ├── core/
│       │   ├── __init__.py
│       │   ├── config.py
│       │   ├── memory.py
│       │   └── state.py
│       ├── languages/
│       │   ├── __init__.py
│       │   ├── detector.py
│       │   ├── translator.py
│       │   └── resources/
│       │       ├── en.json
│       │       ├── zh.json
│       │       └── es.json
│       ├── workflows/
│       │   ├── __init__.py
│       │   ├── base.py
│       │   └── default.py
│       └── api/
│           ├── __init__.py
│           ├── main.py
│           └── routes.py
├── tests/
│   ├── __init__.py
│   ├── test_agents.py
│   ├── test_languages.py
│   └── test_workflows.py
└── scripts/
    └── setup.py

你可以使用以下命令创建这些目录：

mkdir -p src/multiagent_system/{agents,core,languages/resources,workflows,api}
mkdir -p tests
mkdir -p scripts

安装依赖

现在，让我们编辑pyproject.toml文件，添加我们需要的依赖：

[tool.poetry]
name = "multilingual-agent-system"
version = "0.1.0"
description = "A multilingual multi-agent system framework"
authors = ["Your Name <your.email@example.com>"]
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.9"
langchain = "^0.1.0"
langchain-openai = "^0.0.2"
langchain-community = "^0.0.10"
fastapi = "^0.109.0"
uvicorn = "^0.27.0"
pydantic = "^2.5.0"
pydantic-settings = "^2.1.0"
python-dotenv = "^1.0.0"
chromadb = "^0.4.22"
sentence-transformers = "^2.2.2"
fasttext = "^0.9.2"
deep-translator = "^1.11.4"
python-multipart = "^0.0.6"
websockets = "^12.0"

[tool.poetry.group.dev.dependencies]
pytest = "^7.4.0"
pytest-asyncio = "^0.23.0"
black = "^24.1.0"
flake8 = "^7.0.0"
mypy = "^1.8.0"
pre-commit = "^3.6.0"

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

现在，安装依赖：

poetry install

这可能需要一些时间，因为我们安装了多个依赖包。

设置环境变量

复制.env.example为.env并填写必要的配置：

cp .env.example .env

编辑.env文件：

# 应用配置
APP_NAME=Multilingual Agent System
APP_ENV=development
APP_DEBUG=true

# API配置
API_HOST=0.0.0.0
API_PORT=8000

# OpenAI配置
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_MODEL=gpt-4

# 向量数据库配置
CHROMA_DB_PATH=./data/chroma

# 语言配置
DEFAULT_LANGUAGE=en
SUPPORTED_LANGUAGES=en,zh,es

# 内存配置
CONVERSATION_MEMORY_WINDOW=10

确保将your_openai_api_key_here替换为你的实际OpenAI API密钥。

创建基础配置文件

现在，让我们创建一些基础配置文件。首先是.gitignore：

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg

# Virtual Environment
venv/
ENV/
env/

# IDE
.vscode/
.idea/
*.swp
*.swo
*~

# Environment variables
.env

# Data
data/

# Logs
*.log

# Testing
.pytest_cache/
.coverage
htmlcov/

# Jupyter Notebook
.ipynb_checkpoints

# OS files
.DS_Store
Thumbs.db

接下来，创建一个简单的README.md：

# Multilingual Agent System

一个支持多语言的多Agent协作系统框架。

## 功能特性

- 多Agent协作框架
- 多语言支持（可扩展）
- 灵活的工作流编排
- 可插拔的Agent设计
- RESTful API接口

## 快速开始

### 安装依赖

```bash
poetry install

配置环境

复制.env.example为.env并填写必要的配置：

cp .env.example .env

运行服务

poetry run uvicorn src.multiagent_system.api.main:app --reload

项目结构

multilingual-agent-system/
├── src/
│   └── multiagent_system/
│       ├── agents/          # Agent实现
│       ├── core/            # 核心功能
│       ├── languages/       # 多语言支持
│       ├── workflows/       # 工作流定义
│       └── api/             # API接口
├── tests/                   # 测试代码
└── scripts/                 # 辅助脚本

许可证

MIT

环境准备工作已经完成。接下来，我们将开始实现系统的核心组件。

### 11. 核心配置与基础模块

让我们从实现核心配置和基础模块开始。

#### 配置管理

首先，创建`src/multiagent_system/core/config.py`，用于管理系统配置：

```python
"""
系统配置管理模块
"""
from typing import List, Optional
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    """应用配置类"""
    
    # 应用配置
    app_name: str = "Multilingual Agent System"
    app_env: str = "development"
    app_debug: bool = True
    
    # API配置
    api_host: str = "0.0.0.0"
    api_port: int = 8000
    
    # OpenAI配置
    openai_api_key: str
    openai_model: str = "gpt-4"
    openai_temperature: float = 0.7
    openai_max_tokens: int = 2000
    
    # 向量数据库配置
    chroma_db_path: str = "./data/chroma"
    embedding_model: str = "all-MiniLM-L6-v2"
    
    # 语言配置
    default_language: str = "en"
    supported_languages: str = "en,zh,es"
    
    # 内存配置
    conversation_memory_window: int = 10
    
    # Agent配置
    agent_timeout: int = 300  # 5分钟
    max_retry_attempts: int = 3
    
    class Config:
        env_file = ".env"
        case_sensitive = False
    
    @property
    def supported_languages_list(self) -> List[str]:
        """获取支持的语言列表"""
        return [lang.strip() for lang in self.supported_languages.split(",")]


@lru_cache()
def get_settings() -> Settings:
    """获取配置单例"""
    return Settings()

这个配置类使用Pydantic Settings来管理环境变量，并提供了一些方便的属性方法。

基础Agent类

接下来，创建src/multiagent_system/agents/base.py，定义所有Agent的基类：

"""
Agent基类模块
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional
from datetime import datetime
import uuid
from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
from ..core.config import get_settings
from ..core.state import AgentState


class BaseAgent(ABC):
    """Agent基类"""
    
    def __init__(
        self,
        name: str,
        role: str,
        description: str,
        language: Optional[str] = None,
    ):
        """
        初始化Agent
        
        Args:
            name: Agent名称
            role: Agent角色
            description: Agent描述
            language: 使用的语言，默认为系统默认语言
        """
        self.settings = get_settings()
        self.agent_id = str(uuid.uuid4())
        self.name = name
        self.role = role
        self.description = description
        self.language = language or self.settings.default_language
        self.created_at = datetime.now()
        self.state = AgentState.IDLE
        self.memory: List[BaseMessage] = []
    
    @abstractmethod
    async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        执行Agent任务
        
        Args:
            input_data: 输入数据字典
            
        Returns:
            执行结果字典
        """
        pass
    
    @abstractmethod
    def get_system_prompt(self) -> str:
        """
        获取系统提示词
        
        Returns:
            系统提示词字符串
        """
        pass
    
    def add_to_memory(self, message: BaseMessage) -> None:
        """
        添加消息到记忆
        
        Args:
            message: 要添加的消息
        """
        self.memory.append(message)
        # 限制记忆长度
        if len(self.memory) > self.settings.conversation_memory_window * 2:
            self.memory = self.memory[-self.settings.conversation_memory_window * 2:]
    
    def get_memory_context(self) -> List[BaseMessage]:
        """
        获取记忆上下文
        
        Returns:
            记忆中的消息列表
        """
        return self.memory.copy()
    
    def clear_memory(self) -> None:
        """清空记忆"""
        self.memory = []
    
    def update_state(self, new_state: AgentState) -> None:
        """
        更新Agent状态
        
        Args:
            new_state: 新状态
        """
        self.state = new_state
    
    def to_dict(self) -> Dict[str, Any]:
        """
        转换为字典表示
        
        Returns:
            Agent信息字典
        """
        return {
            "agent_id": self.agent_id,
            "name": self.name,
            "role": self.role,
            "description": self.description,
            "language": self.language,
            "created_at": self.created_at.isoformat(),
            "state": self.state.value,
        }
    
    def __repr__(self) -> str:
        return f"<BaseAgent {self.name} ({self.agent_id})>"

这个基类定义了所有Agent都应该具有的基本属性和方法。接下来，让我们定义Agent状态枚举。

Agent状态管理

创建src/multiagent_system/core/state.py：

"""
状态管理模块
"""
from enum import Enum
from typing import Dict, Any, Optional
from datetime import datetime
import uuid


class AgentState(Enum):
    """Agent状态枚举"""
    IDLE = "idle"           # 空闲
    INITIALIZING = "initializing"  # 初始化中
    THINKING = "thinking"   # 思考中
    ACTING = "acting"       # 执行中
    WAITING = "waiting"     # 等待中
    OBSERVING = "observing" # 观察中
    COMPLETED = "completed" # 已完成
    ERROR = "error"         # 错误


class TaskState(Enum):
    """任务状态枚举"""
    PENDING = "pending"       # 待处理
    ASSIGNED = "assigned"     # 已分配
    IN_PROGRESS = "in_progress"  # 进行中
    COMPLETED = "completed"   # 已完成
    FAILED = "failed"         # 失败
    CANCELLED = "cancelled"   # 已取消


class Task:
    """任务类"""
    
    def __init__(
        self,
        description: str,
        assigned_agent: Optional[str] = None,
        priority: int = 5,
        metadata: Optional[Dict[str, Any]] = None,
    ):
        """
        初始化任务
        
        Args:
            description: 任务描述
            assigned_agent: 分配的Agent ID
            priority: 优先级，1-10，数字越大优先级越高
            metadata: 附加元数据
        """
        self.task_id = str(uuid.uuid4())
        self.description = description
        self.assigned_agent = assigned_agent
        self.priority = priority
        self.metadata = metadata or {}
        self.state = TaskState.PENDING
        self.created_at = datetime.now()
        self.started_at: Optional[datetime] = None
        self.completed_at: Optional[datetime] = None
        self.result: Optional[Dict[str, Any]] = None
        self.error: Optional[str] = None
    
    def start(self) -> None:
        """开始任务"""
        self.state = TaskState.IN_PROGRESS
        self.started_at = datetime.now()
    
    def complete(self, result: Dict[str, Any]) -> None:
        """
        完成任务
        
        Args:
            result: 任务结果
        """
        self.state = TaskState.COMPLETED
        self.completed_at = datetime.now()
        self.result = result
    
    def fail(self, error: str) -> None:
        """
        任务失败
        
        Args:
            error: 错误信息
        """
        self.state = TaskState.FAILED
        self.completed_at = datetime.now()
        self.error = error
    
    def to_dict(self) -> Dict[str, Any]:
        """
        转换为字典表示
        
        Returns:
            任务信息字典
        """
        return {
            "task_id": self.task_id,
            "description": self.description,
            "assigned_agent": self.assigned_agent,
            "priority": self.priority,
            "metadata": self.metadata,
            "state": self.state.value,
            "created_at": self.created_at.isoformat(),
            "started_at": self.started_at.isoformat() if self.started_at else None,
            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
            "result": self.result,
            "error": self.error,
        }

这个模块定义了Agent和任务的状态，以及任务类的基本实现。

12. 多语言支持模块

接下来，让我们实现多语言支持模块。

语言检测器

首先，创建src/multiagent_system/languages/detector.py：

"""
语言检测模块
"""
from typing import Dict, List, Optional, Tuple
import os
import fasttext
from ..core.config import get_settings


class LanguageDetector:
    """语言检测器"""
    
    def __init__(self, model_path: Optional[str] = None):
        """
        初始化语言检测器
        
        Args:
            model_path: fastText模型路径，如果未提供则尝试下载
        """
        self.settings = get_settings()
        self.model = None
        self._load_model(model_path)
    
    def _load_model(self, model_path: Optional[str] = None) -> None:
        """
        加载fastText语言检测模型
        
        Args:
            model_path: 模型路径
        """
        if model_path and os.path.exists(model_path):
            self.model = fasttext.load_model(model_path)
        else:
            # 尝试查找或下载模型
            # 注意：在生产环境中，你可能希望提前下载好模型
            try:
                # 这里简化处理，实际使用时应该确保模型存在
                # 可以从 https://fasttext.cc/docs/en/language-identification.html 下载
                raise FileNotFoundError(
                    "Language detection model not found. "
                    "Please download lid.176.bin from "
                    "https://fasttext.cc/docs/en/language-identification.html "
                    "and provide the path."
                )
            except Exception as e:
                # 如果没有fastText模型，我们可以使用一个简单的备用方案
                # 或者依赖大语言模型进行语言检测
                print(f"Warning: {e}. Using fallback language detection.")
                self.model = None
    
    def detect(
        self, 
        text: str, 
        confidence_threshold: float = 0.7
    ) -> Tuple[str, float]:
        """
        检测文本语言
        
        Args:
            text: 待检测的文本
            confidence_threshold: 置信度阈值
            
        Returns:
            (语言代码, 置信度)元组
        """
        # 预处理文本
        cleaned_text = text.strip().replace("\n", " ")
        
        if not cleaned_text:
            return self.settings.default_language, 1.0
        
        if self.model:
            # 使用fastText模型进行检测
            predictions = self.model.predict(cleaned_text, k=3)
            # 解析结果，fastText返回格式类似 ('__label__en',)
            lang_code = predictions[0][0].replace("__label__", "")
            confidence = float(predictions[1][0])
            
            # 检查是否在支持的语言列表中
            if lang_code in self.settings.supported_languages_list:
                if confidence >= confidence_threshold:
                    return lang_code, confidence
            else:
                # 如果检测到的语言不支持，检查其他预测结果
                for i, lang in enumerate(predictions[0][1:], start=1):
                    lang_code = lang.replace("__label__", "")
                    confidence = float(predictions[1][i])
                    if lang_code in self.settings.supported_languages_list and confidence >= confidence_threshold:
                        return lang_code, confidence
        
        # 如果fastText不可用或结果不理想，使用简单启发式方法
        return self._heuristic_detect(cleaned_text)
    
    def _heuristic_detect(self, text: str) -> Tuple[str, float]:
        """
        启发式语言检测（备用方案）
        
        Args:
            text: 待检测的文本
            
        Returns:
            (语言代码, 置信度)元组
        """
        # 计算文本中各语言特征字符的比例
        # 这是一个简化的实现，实际应用中可以使用更复杂的方法
        
        # 检查中文字符
        chinese_chars = sum(1 for char in text if '\u4e00' <= char <= '\u9fff')
        if chinese_chars / max(len(text), 1) > 0.3:
            return "zh", 0.8
        
        # 检查西班牙语特征
        spanish_chars = text.count('ñ') + text.count('¿') + text.count('¡')
        if spanish_chars > 0:
            return "es", 0.7
        
        # 默认返回英语
        return self.settings.default_language, 0.5
    
    def detect_batch(
        self, 
        texts: List[str], 
        confidence_threshold: float = 0.7
    ) -> List[Tuple[str, float]]:
        """
        批量检测文本语言
        
        Args:
            texts: 待检测的文本列表
            confidence_threshold: 置信度阈值
            
        Returns:
            (语言代码, 置信度)元组列表
        """
        return [self.detect(text, confidence_threshold) for text in texts]

这个模块提供了语言检测功能，使用fastText作为主要方法，并提供了启发式备用方案。

多语言翻译器

接下来，创建src/multiagent_system/languages/translator.py：

"""
多语言翻译模块
"""
from typing import Dict, Optional
from deep_translator import GoogleTranslator, ChatGptTranslator
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from ..core.config import get_settings


class MultiLanguageTranslator:
    """多语言翻译器"""
    
    def __init__(self):
        """初始化翻译器"""
        self.settings = get_settings()
        self._init_translators()
        self._llm = None
    
    def _init_translators(self) -> None:
        """初始化翻译后端"""
        # 初始化Google翻译器作为备用
        try:
            self.google_translator = GoogleTranslator()
        except Exception as e:
            print(f"Warning: Could not initialize