Multi-Agent Orchestration Patterns in Microsoft Agent Framework – Part 7

from agent_framework.orchestration import SequentialOrchestrator
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
import asyncio

async def create_content_pipeline():
    client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
    
    # Create specialized agents
    researcher = client.create_agent(
        name="Researcher",
        instructions="""
            You are a research specialist. Given a topic:
            1. Identify key subtopics to cover
            2. Find relevant facts and statistics
            3. Note important sources and references
            4. Compile a research brief for the writer
        """
    )
    
    writer = client.create_agent(
        name="Writer",
        instructions="""
            You are a professional content writer. Given research:
            1. Create an engaging article structure
            2. Write clear, compelling prose
            3. Include relevant examples
            4. Maintain consistent tone and style
        """
    )
    
    editor = client.create_agent(
        name="Editor",
        instructions="""
            You are an expert editor. Review the draft for:
            1. Grammar and spelling errors
            2. Clarity and readability
            3. Logical flow and structure
            4. Factual accuracy
            Provide the final polished version.
        """
    )
    
    # Create sequential orchestrator
    orchestrator = SequentialOrchestrator(
        agents=[researcher, writer, editor]
    )
    
    return orchestrator

async def main():
    orchestrator = await create_content_pipeline()
    
    print("Starting content pipeline...")
    print("=" * 50)
    
    # Execute pipeline: researcher -> writer -> editor
    result = await orchestrator.run(
        "Create an article about the benefits of AI agents in enterprise software development"
    )
    
    print(f"\nFinal Article:\n{result}")

if __name__ == "__main__":
    asyncio.run(main())

from agent_framework.orchestration import ConcurrentOrchestrator
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
import asyncio

async def create_analysis_team():
    client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
    
    # Create parallel analysis agents
    market_analyst = client.create_agent(
        name="MarketAnalyst",
        instructions="""
            Analyze market conditions for the given company/topic:
            - Market size and growth trends
            - Competitive landscape
            - Market opportunities and threats
            Format as a structured market analysis section.
        """
    )
    
    tech_analyst = client.create_agent(
        name="TechAnalyst",
        instructions="""
            Analyze technical aspects:
            - Technology stack and architecture
            - Innovation and R&D capabilities  
            - Technical risks and opportunities
            Format as a structured technical analysis section.
        """
    )
    
    financial_analyst = client.create_agent(
        name="FinancialAnalyst",
        instructions="""
            Analyze financial health:
            - Revenue and growth metrics
            - Profitability and margins
            - Financial risks and opportunities
            Format as a structured financial analysis section.
        """
    )
    
    # Aggregator function to combine results
    def aggregate_analysis(results: list) -> str:
        combined = "# Comprehensive Analysis Report\n\n"
        sections = ["## Market Analysis", "## Technical Analysis", "## Financial Analysis"]
        
        for i, (section, result) in enumerate(zip(sections, results)):
            combined += f"{section}\n{result}\n\n"
        
        combined += "## Summary\nAnalysis compiled from market, technical, and financial perspectives."
        return combined
    
    orchestrator = ConcurrentOrchestrator(
        agents=[market_analyst, tech_analyst, financial_analyst],
        aggregator=aggregate_analysis
    )
    
    return orchestrator

async def main():
    orchestrator = await create_analysis_team()
    
    print("Starting parallel analysis (3 agents working simultaneously)...")
    print("=" * 50)
    
    # All agents work in parallel
    result = await orchestrator.run(
        "Analyze Microsoft for potential investment opportunity"
    )
    
    print(f"\nCombined Report:\n{result}")

if __name__ == "__main__":
    asyncio.run(main())

from agent_framework.orchestration import HandoffOrchestrator
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
import asyncio

async def create_support_system():
    client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
    
    # Tiered support agents
    tier1 = client.create_agent(
        name="Tier1Support",
        instructions="""
            You are a Tier 1 support agent. Handle basic inquiries:
            - Password resets
            - Account information
            - FAQ questions
            - Basic troubleshooting
            
            If the issue is technical or complex, say "HANDOFF:Tier2Support"
            If the issue is about billing, say "HANDOFF:BillingAgent"
            Otherwise, resolve the issue directly.
        """
    )
    
    tier2 = client.create_agent(
        name="Tier2Support",
        instructions="""
            You are a Tier 2 technical support agent. Handle:
            - Complex technical issues
            - Bug reports
            - Integration problems
            - Performance issues
            
            If the issue requires engineering, say "HANDOFF:EngineeringEscalation"
            Otherwise, resolve the technical issue.
        """
    )
    
    billing = client.create_agent(
        name="BillingAgent",
        instructions="""
            You are a billing specialist. Handle:
            - Invoice questions
            - Payment issues
            - Refund requests
            - Subscription changes
            
            For refunds over $500, say "HANDOFF:BillingManager"
            Otherwise, resolve the billing issue.
        """
    )
    
    engineering = client.create_agent(
        name="EngineeringEscalation",
        instructions="""
            You are an engineering escalation agent. Handle:
            - Critical bugs
            - System outages
            - Security issues
            Create a detailed ticket for the engineering team.
        """
    )
    
    # Create handoff orchestrator
    orchestrator = HandoffOrchestrator(
        initial_agent=tier1,
        agents={
            "Tier2Support": tier2,
            "BillingAgent": billing,
            "EngineeringEscalation": engineering
        },
        handoff_pattern=r"HANDOFF:(\w+)"  # Regex to detect handoff
    )
    
    return orchestrator

async def main():
    orchestrator = await create_support_system()
    
    test_cases = [
        "I forgot my password",
        "I have a question about my invoice from last month",
        "The API is returning 500 errors consistently"
    ]
    
    for query in test_cases:
        print(f"\nCustomer: {query}")
        print("-" * 40)
        result = await orchestrator.run(query)
        print(f"Resolution: {result}")

if __name__ == "__main__":
    asyncio.run(main())

from agent_framework.orchestration import GroupChatOrchestrator
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
import asyncio

async def create_design_team():
    client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
    
    # Collaborative team members
    product_manager = client.create_agent(
        name="ProductManager",
        instructions="""
            You are a Product Manager. In discussions:
            - Focus on user needs and business value
            - Prioritize features based on impact
            - Consider timelines and resources
            - Make final decisions on scope
            Keep responses concise (2-3 sentences).
        """
    )
    
    engineer = client.create_agent(
        name="Engineer",
        instructions="""
            You are a Senior Engineer. In discussions:
            - Evaluate technical feasibility
            - Suggest implementation approaches
            - Identify potential challenges
            - Estimate complexity
            Keep responses concise (2-3 sentences).
        """
    )
    
    designer = client.create_agent(
        name="Designer",
        instructions="""
            You are a UX Designer. In discussions:
            - Focus on user experience
            - Suggest interface improvements
            - Consider accessibility
            - Propose design patterns
            Keep responses concise (2-3 sentences).
        """
    )
    
    qa_lead = client.create_agent(
        name="QALead",
        instructions="""
            You are a QA Lead. In discussions:
            - Identify testing requirements
            - Note potential edge cases
            - Consider quality metrics
            - Suggest test strategies
            Keep responses concise (2-3 sentences).
        """
    )
    
    # Create group chat with PM as moderator
    orchestrator = GroupChatOrchestrator(
        agents=[product_manager, engineer, designer, qa_lead],
        moderator=product_manager,  # PM controls the conversation
        max_rounds=5,  # Limit discussion rounds
        summary_agent=product_manager  # PM summarizes decisions
    )
    
    return orchestrator

async def main():
    orchestrator = await create_design_team()
    
    print("Starting design discussion...")
    print("=" * 50)
    
    result = await orchestrator.run(
        "Design a new feature: Allow users to schedule AI agent tasks to run at specific times"
    )
    
    print(f"\nFinal Decision:\n{result}")

if __name__ == "__main__":
    asyncio.run(main())

from agent_framework.orchestration import MagenticOrchestrator
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
import asyncio

async def create_project_team():
    client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
    
    # Manager agent that coordinates everything
    manager = client.create_agent(
        name="ProjectManager",
        instructions="""
            You are a Project Manager coordinating a team of specialists.
            
            Your responsibilities:
            1. Break down complex tasks into subtasks
            2. Assign subtasks to appropriate specialists
            3. Maintain a task ledger tracking progress
            4. Synthesize results from specialists
            5. Make decisions when specialists disagree
            
            Available specialists:
            - Researcher: For gathering information
            - Analyst: For data analysis and insights
            - Writer: For content creation
            - Reviewer: For quality review
            
            Format task assignments as: @SpecialistName: [task description]
            Track progress and coordinate until the goal is achieved.
        """
    )
    
    # Specialist agents
    researcher = client.create_agent(
        name="Researcher",
        instructions="""
            You are a Research Specialist. When assigned a task:
            1. Gather relevant information
            2. Identify key facts and data
            3. Compile sources and references
            Return structured research findings.
        """
    )
    
    analyst = client.create_agent(
        name="Analyst",
        instructions="""
            You are a Data Analyst. When assigned a task:
            1. Analyze provided data and research
            2. Identify patterns and insights
            3. Create data-driven recommendations
            Return structured analysis with conclusions.
        """
    )
    
    writer = client.create_agent(
        name="Writer",
        instructions="""
            You are a Content Writer. When assigned a task:
            1. Create clear, engaging content
            2. Structure information logically
            3. Maintain professional tone
            Return polished written content.
        """
    )
    
    reviewer = client.create_agent(
        name="Reviewer",
        instructions="""
            You are a Quality Reviewer. When assigned a task:
            1. Review content for accuracy
            2. Check for completeness
            3. Suggest improvements
            Return review feedback and final approval.
        """
    )
    
    # Create magentic orchestrator
    orchestrator = MagenticOrchestrator(
        manager=manager,
        specialists=[researcher, analyst, writer, reviewer],
        max_iterations=10,  # Max coordination cycles
        allow_human_input=True  # Can pause for human guidance
    )
    
    return orchestrator

async def main():
    orchestrator = await create_project_team()
    
    print("Starting managed project...")
    print("=" * 50)
    
    result = await orchestrator.run(
        """Create a competitive analysis report for AI agent frameworks,
        comparing Microsoft Agent Framework, LangChain, and CrewAI.
        Include market positioning, technical capabilities, and recommendations."""
    )
    
    print(f"\nFinal Deliverable:\n{result}")

if __name__ == "__main__":
    asyncio.run(main())

from agent_framework.a2a import A2AAgent, AgentCard
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
import asyncio

async def discover_external_agent(url: str) -> AgentCard:
    """Discover an external agent's capabilities via A2A protocol."""
    async with aiohttp.ClientSession() as session:
        # Fetch agent card from well-known endpoint
        async with session.get(f"{url}/.well-known/agent.json") as response:
            if response.status == 200:
                data = await response.json()
                return AgentCard(
                    name=data["name"],
                    description=data["description"],
                    url=data["url"],
                    capabilities=data.get("capabilities", []),
                    authentication=data.get("authentication", {})
                )
    raise Exception(f"Could not discover agent at {url}")

async def create_cross_system_workflow():
    client = AzureOpenAIResponsesClient(credential=AzureCliCredential())
    
    # Local agent
    local_processor = client.create_agent(
        name="LocalProcessor",
        instructions="Process and prepare data for external analysis."
    )
    
    # Discover and connect to external A2A agent
    external_card = await discover_external_agent("https://external-service.com")
    
    external_agent = A2AAgent(
        name=external_card.name,
        url=external_card.url,
        agent_card=external_card,
        auth_token=os.getenv("EXTERNAL_API_TOKEN")
    )
    
    # Another local agent for final processing
    result_formatter = client.create_agent(
        name="ResultFormatter",
        instructions="Format and present the final results clearly."
    )
    
    # Use in sequential workflow
    from agent_framework.orchestration import SequentialOrchestrator
    
    orchestrator = SequentialOrchestrator(
        agents=[local_processor, external_agent, result_formatter]
    )
    
    return orchestrator

async def main():
    try:
        orchestrator = await create_cross_system_workflow()
        
        print("Running cross-system workflow...")
        result = await orchestrator.run("Analyze customer sentiment from Q4 feedback")
        
        print(f"Result: {result}")
    except Exception as e:
        print(f"A2A connection failed: {e}")
        print("Falling back to local-only processing...")

if __name__ == "__main__":
    asyncio.run(main())