AgentScope示例应用：多智能体对话

【免费下载链接】agentscope 项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope

引言：为什么需要多智能体对话？

在人工智能快速发展的今天，单一智能体已经难以满足复杂任务的需求。想象一下这样的场景：一个项目团队需要讨论技术方案，不同专业背景的成员各抒己见，最终达成共识。这正是多智能体对话要解决的问题——让多个AI智能体像真实团队成员一样进行有意义的交流。

AgentScope作为面向开发者的多智能体编程框架，提供了强大的多智能体对话能力。本文将深入探讨如何使用AgentScope构建高效的多智能体对话系统。

核心概念解析

消息中心（MsgHub）

MsgHub是AgentScope中管理多智能体对话的核心组件，它提供了以下关键功能：

多智能体格式化器（MultiAgent Formatter）

与单智能体对话不同，多智能体对话需要使用专门的格式化器来处理多个参与者的消息：

格式化器类型	适用场景	特点
DashScopeChatFormatter	单智能体对话	使用role字段区分用户和智能体
DashScopeMultiAgentFormatter	多智能体对话	使用name字段区分不同智能体

实战：构建多角色对话系统

基础配置

首先，我们需要设置环境并导入必要的模块：

import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.pipeline import MsgHub, sequential_pipeline

创建个性化智能体

每个智能体都应该有独特的个性和背景故事：

def create_participant_agent(
    name: str,
    age: int,
    career: str,
    character: str,
) -> ReActAgent:
    """创建具有特定属性的参与者智能体"""
    return ReActAgent(
        name=name,
        sys_prompt=(
            f"You're a {age}-year-old {career} named {name} and you're "
            f"a {character} person. Speak naturally and engage in "
            f"meaningful conversations with other participants."
        ),
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=True,
        ),
        formatter=DashScopeMultiAgentFormatter(),
    )

构建对话工作流

使用MsgHub管理多智能体对话：

async def multi_agent_conversation() -> None:
    """运行多智能体对话工作流"""
    
    # 创建具有不同特征的多个参与者智能体
    alice = create_participant_agent("Alice", 30, "teacher", "friendly")
    bob = create_participant_agent("Bob", 14, "student", "rebellious")
    charlie = create_participant_agent("Charlie", 28, "doctor", "thoughtful")
    diana = create_participant_agent("Diana", 35, "engineer", "practical")

    # 在消息中心中创建对话
    async with MsgHub(
        participants=[alice, bob, charlie],
        announcement=Msg(
            "system",
            "Welcome to the discussion! Please introduce yourselves "
            "and share your thoughts on AI technology.",
            "system",
        ),
    ) as hub:
        
        # 顺序执行对话
        await sequential_pipeline([alice, bob, charlie])
        
        # 动态添加新参与者
        print("##### Diana joins the conversation #####")
        hub.add(diana)
        await hub.broadcast(
            Msg("system", "Diana has joined our discussion.", "system")
        )
        await diana()
        
        # 动态移除参与者
        print("##### Bob leaves the conversation #####")
        hub.delete(bob)
        await hub.broadcast(
            Msg("bob", "I need to go now. It was great talking with everyone!", "assistant")
        )
        
        # 继续剩余参与者的对话
        await alice()
        await charlie()
        await diana()

# 运行对话
asyncio.run(multi_agent_conversation())

高级特性与应用场景

1. 实时参与者管理

AgentScope支持动态的参与者管理，这在以下场景中特别有用：

2. 消息广播机制

MsgHub的广播机制确保所有参与者都能收到相关消息：

async def advanced_broadcast_example():
    """高级广播功能示例"""
    async with MsgHub(participants=[alice, bob, charlie]) as hub:
        # 选择性广播
        await hub.broadcast(
            Msg("moderator", "Important announcement for everyone!", "system")
        )
        
        # 定向消息（通过设置特定接收者）
        # 在实际应用中可以通过自定义逻辑实现

3. 内存管理

每个智能体都有独立的内存系统，记录对话历史：

async def check_memory_example():
    """检查智能体内存示例"""
    print("Alice的记忆内容:")
    for msg in await alice.memory.get_memory():
        print(f"{msg.name}: {msg.get_text_content()}")
    
    print("\nBob的记忆内容:")
    for msg in await bob.memory.get_memory():
        print(f"{msg.name}: {msg.get_text_content()}")

性能优化建议

1. 模型配置优化

# 使用共享模型实例提高性能
shared_model = DashScopeChatModel(
    model_name="qwen-max",
    api_key=os.environ["DASHSCOPE_API_KEY"],
    stream=True,
)

shared_formatter = DashScopeMultiAgentFormatter()

# 为所有智能体使用相同的模型和格式化器实例
alice = ReActAgent(
    name="Alice",
    sys_prompt="...",
    model=shared_model,
    formatter=shared_formatter,
    memory=InMemoryMemory(),
)

2. 异步执行优化

async def optimized_conversation():
    """优化后的对话执行"""
    agents = [alice, bob, charlie, diana]
    
    # 使用fanout管道进行并行执行
    from agentscope.pipeline import fanout_pipeline
    
    results = await fanout_pipeline(
        agents,
        Msg("system", "Please share your opinion simultaneously.", "system"),
        enable_gather=True
    )
    
    # 处理并行执行结果
    for agent, result in zip(agents, results):
        print(f"{agent.name}: {result.get_text_content()}")

实际应用案例

案例1：技术方案讨论会

async def tech_discussion():
    """技术方案讨论会示例"""
    # 创建不同技术背景的专家
    backend_expert = create_participant_agent(
        "BackendExpert", 32, "backend developer", "technical"
    )
    frontend_expert = create_participant_agent(
        "FrontendExpert", 28, "frontend developer", "creative"
    )
    product_manager = create_participant_agent(
        "ProductManager", 35, "product manager", "strategic"
    )
    
    async with MsgHub(
        participants=[backend_expert, frontend_expert, product_manager],
        announcement=Msg(
            "system",
            "Discuss the technical architecture for our new microservices platform. "
            "Consider scalability, maintainability, and development efficiency.",
            "system",
        ),
    ):
        await sequential_pipeline([
            product_manager,  # 首先由产品经理提出需求
            backend_expert,   # 后端专家给出技术方案
            frontend_expert,  # 前端专家提出界面设计考虑
            backend_expert,   # 后端专家回应前端需求
            product_manager   # 产品经理总结讨论
        ])

案例2：多语言翻译协作

async def multilingual_translation():
    """多语言翻译协作示例"""
    # 创建不同语言的翻译专家
    english_expert = create_participant_agent(
        "EnglishExpert", 30, "translator", "precise"
    )
    chinese_expert = create_participant_agent(
        "ChineseExpert", 32, "translator", "cultural"
    )
    japanese_expert = create_participant_agent(
        "JapaneseExpert", 28, "translator", "detailed"
    )
    
    async with MsgHub(participants=[
        english_expert, chinese_expert, japanese_expert
    ]):
        # 协作翻译流程
        await english_expert()
        await chinese_expert()
        await japanese_expert()
        await english_expert()  # 最终校对

最佳实践与注意事项

1. 智能体角色设计

角色属性	设计建议	示例
姓名	使用有意义的名称	"TechLead", "DesignExpert"
年龄	符合角色背景	25-45岁之间的专业人士
职业	明确专业领域	"软件工程师", "产品设计师"
性格	体现对话风格	"analytical", "creative", "practical"

2. 对话流程控制

# 良好的流程控制示例
async def well_structured_conversation():
    """结构良好的对话示例"""
    async with MsgHub(participants=agents) as hub:
        # 阶段1: 自我介绍
        await hub.broadcast(Msg("system", "Phase 1: Self-introduction", "system"))
        await sequential_pipeline(agents)
        
        # 阶段2: 主题讨论
        await hub.broadcast(Msg("system", "Phase 2: Topic Discussion", "system"))
        for _ in range(2):  # 进行两轮讨论
            await sequential_pipeline(agents)
        
        # 阶段3: 总结
        await hub.broadcast(Msg("system", "Phase 3: Conclusion", "system"))
        await sequential_pipeline(agents)

3. 错误处理与重试机制

async def robust_conversation():
    """健壮的对话处理示例"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            async with MsgHub(participants=agents) as hub:
                await sequential_pipeline(agents)
            break  # 成功完成，退出重试循环
        except Exception as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt == max_retries - 1:
                raise  # 最后一次尝试仍然失败，抛出异常
            await asyncio.sleep(2)  # 等待后重试

总结与展望

AgentScope的多智能体对话功能为构建复杂的AI协作系统提供了强大基础。通过MsgHub的消息管理、多智能体格式化器的智能提示构建，以及灵活的管道系统，开发者可以轻松创建各种多智能体应用场景。

未来，随着AgentScope的持续发展，我们可以期待更多高级功能，如：

• 更精细的对话状态管理
• 增强的上下文理解能力
• 实时协作和协调机制
• 更强大的异常处理和恢复能力

多智能体对话不仅是技术挑战，更是开启AI协作新纪元的钥匙。通过AgentScope，每个开发者都有机会构建出真正智能、协作的AI系统，推动人工智能向更高层次发展。

开始你的多智能体对话之旅吧，探索AI协作的无限可能！

【免费下载链接】agentscope 项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope