Multi-Agent Systems

from qwen_agent.agents import Assistant, ReActChat, Router

# Define specialized agents
vl_agent = Assistant(
    llm={'model': 'qwen-vl-max'},
    name='Vision Assistant',
    description='Can understand image content'
)

tool_agent = ReActChat(
    llm={'model': 'qwen-max'},
    name='Tool Assistant',
    description='Can use drawing tools and run code',
    function_list=['image_gen', 'code_interpreter']
)

# Create router
bot = Router(
    llm={'model': 'qwen-max'},
    agents=[vl_agent, tool_agent]
)

messages = [{'role': 'user', 'content': 'Draw a chart showing sales data'}]
for response in bot.run(messages):
    print(response)

1. User Query
      ↓
2. Router analyzes query
      ↓
3. Selects appropriate agent
   OR handles directly
      ↓
4. Selected agent processes
      ↓
5. Response to user

from qwen_agent.agents import Router

router = Router(
    llm={'model': 'qwen-max'},
    agents=[agent1, agent2, agent3],
    name='MainRouter',
    description='Routes tasks to specialized agents',
    files=['knowledge.pdf'],  # Optional RAG
    function_list=['search']  # Optional tools for router itself
)

from qwen_agent.agents import Assistant, ReActChat, Router
from qwen_agent.gui import WebUI

def init_agent_service():
    llm_cfg = {'model': 'qwen-max'}
    llm_cfg_vl = {'model': 'qwen-vl-max'}
    tools = ['image_gen', 'code_interpreter']

    # Vision agent
    bot_vl = Assistant(
        llm=llm_cfg_vl,
        name='Multimodal Assistant',
        description='Can understand image content'
    )

    # Tool agent
    bot_tool = ReActChat(
        llm=llm_cfg,
        name='Tool Assistant',
        description='Can use drawing tools and run code to solve problems',
        function_list=tools
    )

    # Router agent (also serves as text agent)
    bot = Router(
        llm=llm_cfg,
        agents=[bot_vl, bot_tool]
    )
    return bot

def app_gui():
    bot = init_agent_service()
    chatbot_config = {'verbose': True}
    WebUI(bot, chatbot_config=chatbot_config).run()

if __name__ == '__main__':
    app_gui()

from qwen_agent.agents import GroupChat

cfgs = {
    'background': 'A collaborative problem-solving team',
    'agents': [
        {
            'name': 'Researcher',
            'description': 'Expert in research and analysis',
            'instructions': 'You are a thorough researcher who provides detailed analysis.',
            'selected_tools': []
        },
        {
            'name': 'Coder',
            'description': 'Expert programmer',
            'instructions': 'You write clean, efficient code to solve problems.',
            'selected_tools': ['code_interpreter']
        },
        {
            'name': 'Writer',
            'description': 'Expert technical writer',
            'instructions': 'You document solutions clearly and concisely.',
            'selected_tools': []
        }
    ]
}

bot = GroupChat(
    agents=cfgs,
    llm={'model': 'qwen-max'},
    agent_selection_method='auto'
)

messages = [{'role': 'user', 'content': 'Build a data analysis solution'}]
for response in bot.run(messages, max_round=5):
    print(response)

messages = [
    {'role': 'user', 'content': '@Researcher please analyze this problem first'}
]

for response in bot.run(messages):
    print(response)

cfgs = {
    'background': 'A team working on a project',
    'agents': [
        {
            'name': 'Alice',
            'description': 'Project manager and stakeholder',
            'is_human': True  # Mark as human
        },
        {
            'name': 'AI Assistant',
            'description': 'Helpful AI assistant',
            'instructions': 'You help Alice with various tasks.',
            'selected_tools': ['code_interpreter']
        }
    ]
}

bot = GroupChat(agents=cfgs, llm={'model': 'qwen-max'})

from qwen_agent.agents import GroupChat
from qwen_agent.gui import gradio

CFGS = {
    'background': 'A mutual aid group',
    'agents': [
        {
            'name': 'Tang',
            'description': 'A hardworking person (real user)',
            'is_human': True
        },
        {
            'name': 'Zhen Huan',
            'description': 'A palace concubine',
            'instructions': 'You are Zhen Huan. You speak in classical Chinese. After each message, call image_gen to show your mood.',
            'knowledge_files': [],
            'selected_tools': ['image_gen']
        },
        {
            'name': 'ikun',
            'description': 'Familiar with celebrity Cai Xukun',
            'instructions': 'You are a fan of Cai Xukun. You speak briefly and like using emoticons. You recently love watching "Legend of Zhen Huan".',
            'knowledge_files': [],
            'selected_tools': []
        },
        {
            'name': 'Datou',
            'description': 'An athlete who dislikes celebrity culture',
            'instructions': 'You are an athlete. You love sports. You dislike celebrity worship. You like recommending fitness to others.',
            'knowledge_files': [],
            'selected_tools': []
        }
    ]
}

bot = GroupChat(
    agents=CFGS,
    llm={'model': 'qwen-max'},
    agent_selection_method='auto'
)

from qwen_agent import Agent, MultiAgentHub
from qwen_agent.llm.schema import Message
from typing import List, Iterator

class CustomOrchestrator(Agent, MultiAgentHub):
    """Custom multi-agent orchestration."""
    
    def __init__(self, agents, llm, **kwargs):
        super().__init__(llm=llm, **kwargs)
        self._agents = agents
    
    def _run(self, messages: List[Message], **kwargs) -> Iterator[List[Message]]:
        # Custom orchestration logic
        
        # Example: Sequential processing
        current_messages = messages
        for agent in self.agents:
            print(f"Running {agent.name}...")
            response = agent.run_nonstream(current_messages)
            current_messages.extend(response)
        
        yield current_messages

# Usage
agents = [
    Assistant(llm={'model': 'qwen-max'}, name='Agent1'),
    Assistant(llm={'model': 'qwen-max'}, name='Agent2')
]

orchestrator = CustomOrchestrator(
    agents=agents,
    llm={'model': 'qwen-max'}
)

# Get all agents
agents = orchestrator.agents  # List[Agent]

# Get agent names
names = orchestrator.agent_names  # List[str]

# Get non-user agents
ai_agents = orchestrator.nonuser_agents  # List[Agent]

from qwen_agent.agents import Assistant, Router, GroupChat

# Bottom layer: Specialized workers
research_team = GroupChat(
    agents=[
        {'name': 'Researcher1', 'description': '...', 'instructions': '...'},
        {'name': 'Researcher2', 'description': '...', 'instructions': '...'},
    ],
    llm={'model': 'qwen-max'}
)

dev_team = GroupChat(
    agents=[
        {'name': 'Developer1', 'description': '...', 'instructions': '...'},
        {'name': 'Developer2', 'description': '...', 'instructions': '...'},
    ],
    llm={'model': 'qwen-max'}
)

# Top layer: Router coordinates teams
project_manager = Router(
    llm={'model': 'qwen-max'},
    agents=[research_team, dev_team],
    name='ProjectManager'
)

messages = [{'role': 'user', 'content': 'Build a new feature'}]
for response in project_manager.run(messages):
    print(response)

import asyncio
from qwen_agent.agents import Assistant

async def parallel_agents(messages):
    agents = [
        Assistant(llm={'model': 'qwen-max'}, name=f'Agent{i}')
        for i in range(3)
    ]
    
    # Run agents in parallel
    tasks = [
        asyncio.create_task(run_agent(agent, messages))
        for agent in agents
    ]
    
    results = await asyncio.gather(*tasks)
    return results

async def run_agent(agent, messages):
    response = agent.run_nonstream(messages)
    return response

# Usage
messages = [{'role': 'user', 'content': 'Analyze from different perspectives'}]
results = asyncio.run(parallel_agents(messages))

from qwen_agent.agents import Assistant

class Pipeline:
    def __init__(self):
        self.agents = [
            Assistant(
                llm={'model': 'qwen-max'},
                name='Analyzer',
                system_message='Analyze the input and extract key points'
            ),
            Assistant(
                llm={'model': 'qwen-max'},
                name='Enhancer',
                system_message='Enhance the analysis with additional context'
            ),
            Assistant(
                llm={'model': 'qwen-max'},
                name='Formatter',
                system_message='Format the final output professionally'
            )
        ]
    
    def run(self, messages):
        current = messages
        for agent in self.agents:
            print(f"Running {agent.name}...")
            response = agent.run_nonstream(current)
            current.extend(response)
        return current

pipeline = Pipeline()
result = pipeline.run([{'role': 'user', 'content': 'Analyze this data...'}])

bot = GroupChat(
    agents=cfgs,
    llm={'model': 'qwen-max'}
)

chatbot_config = {
    'verbose': True  # Show detailed agent selection and responses
}

WebUI(bot, chatbot_config=chatbot_config).run()

from qwen_agent.log import logger
import logging

logger.setLevel(logging.DEBUG)

# Shows agent selection and communication
for response in bot.run(messages):
    logger.info(f"Response from {response[-1].name}")
    print(response)

# Multi-agent systems use more tokens
# Each agent sees conversation history
# Optimize by:

# 1. Limit conversation rounds
for response in bot.run(messages, max_round=3):
    pass

# 2. Use concise agent instructions
agent = Assistant(
    llm={'model': 'qwen-max'},
    system_message='Brief, focused instructions'  # Not lengthy prompts
)

# 3. Selectively pass context
# Only include relevant history for each agent

# Sequential agents increase latency
# Optimize by:

# 1. Use parallel processing where possible
# 2. Cache agent responses
# 3. Use faster models for routing
router = Router(
    llm={'model': 'qwen-turbo'},  # Faster model for routing
    agents=[agent1, agent2]
)