Agent Orchestrator in Machine Learning

from typing import TypedDict, Literal, Annotated
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 1. Define shared state
class OrchestratorState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str
    results: dict

# 2. Define specialized agents (each is a graph node)
def research_agent(state: OrchestratorState) -> OrchestratorState:
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    response = llm.invoke([
        SystemMessage(content="You are a research specialist."),
        *state["messages"]
    ])
    return {"messages": [response],
            "results": {**state.get("results", {}), "research": response.content}}

def analysis_agent(state: OrchestratorState) -> OrchestratorState:
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    research = state.get("results", {}).get("research", "")
    response = llm.invoke([
        SystemMessage(content=f"Analyze this research:\n{research}"),
        *state["messages"]
    ])
    return {"messages": [response],
            "results": {**state.get("results", {}), "analysis": response.content}}

# 3. Supervisor decides next agent based on state
def supervisor(state: OrchestratorState) -> OrchestratorState:
    results = state.get("results", {})
    if not results.get("research"): return {"next_agent": "research"}
    elif not results.get("analysis"): return {"next_agent": "analysis"}
    else: return {"next_agent": "end"}

def route_next(state) -> Literal["research", "analysis", "__end__"]:
    return "__end__" if state.get("next_agent") == "end" else state["next_agent"]

# 4. Build and compile the graph
graph = StateGraph(OrchestratorState)
graph.add_node("supervisor", supervisor)
graph.add_node("research", research_agent)
graph.add_node("analysis", analysis_agent)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_next)
graph.add_edge("research", "supervisor")
graph.add_edge("analysis", "supervisor")

orchestrator = graph.compile()
result = orchestrator.invoke({
    "messages": [HumanMessage(content="Analyze UPI's impact on financial inclusion in India")],
    "next_agent": "", "results": {}
})

from crewai import Agent, Task, Crew, Process
from crewai.tools import tool

# 1. Define agents with roles and backstories
researcher = Agent(
    role="Senior Research Analyst",
    goal="Find comprehensive, accurate data about the given topic",
    backstory="You are a seasoned analyst at a top Indian consulting firm. "
              "You specialize in technology and financial inclusion research.",
    tools=[search_web],  # custom @tool-decorated functions
    llm="gpt-4o",
    verbose=True
)

analyst = Agent(
    role="Data Analyst", 
    goal="Analyze research findings and extract actionable insights",
    backstory="You are a quantitative analyst who works with RBI and NPCI data.",
    tools=[calculate_metrics],
    llm="gpt-4o",
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Produce a clear, well-structured report from research and analysis",
    backstory="You write for CTOs at Indian fintech companies.",
    llm="gpt-4o",
    verbose=True
)

# 2. Define tasks assigned to agents
research_task = Task(
    description="Research UPI's impact on financial inclusion in India.",
    expected_output="A detailed research brief with statistics and sources",
    agent=researcher
)
analysis_task = Task(
    description="Analyze the research findings. Identify trends and correlations.",
    expected_output="An analytical summary with key metrics",
    agent=analyst
)
report_task = Task(
    description="Write a report combining the research and analysis.",
    expected_output="A polished 2000-word report",
    agent=writer
)

# 3. Create crew with hierarchical process (manager supervises)
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.hierarchical,  # manager agent auto-created
    manager_llm="gpt-4o",
    verbose=True
)

result = crew.kickoff()
print(result)

from agents import Agent, Runner, handoff

# 1. Define specialized agents
billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing inquiries. Process refunds up to INR 5,000.",
    tools=[lookup_order, process_refund],
)
shipping_agent = Agent(
    name="Shipping Specialist", 
    instructions="Handle shipping inquiries. Track packages, update addresses.",
    tools=[track_shipment, update_address],
)
escalation_agent = Agent(
    name="Senior Support Lead",
    instructions="Handle escalated cases. Refunds up to INR 25,000.",
    tools=[process_large_refund, create_engineering_ticket],
)

# 2. Triage agent routes to specialists via handoffs
triage_agent = Agent(
    name="Customer Support Orchestrator",
    instructions="Route customers to the right specialist. Never handle tasks yourself.",
    handoffs=[
        handoff(billing_agent, description="Transfer to billing"),
        handoff(shipping_agent, description="Transfer to shipping"),
        handoff(escalation_agent, description="Escalate to senior"),
    ],
)

# 3. Run with loop protection
import asyncio
async def handle_query(msg: str):
    result = await Runner.run(triage_agent, input=msg, max_turns=10)
    return result.final_output

response = asyncio.run(handle_query("I was charged twice for order #IND-2026-4521"))

import asyncio
from dataclasses import dataclass
from openai import AsyncOpenAI
import time

client = AsyncOpenAI()

@dataclass
class AgentConfig:
    name: str
    system_prompt: str
    model: str = "gpt-4o"
    timeout_seconds: float = 30.0

@dataclass
class AgentResult:
    agent_name: str
    output: str
    latency_ms: float
    success: bool
    error: str | None = None

async def invoke_agent(agent: AgentConfig, prompt: str) -> AgentResult:
    """Invoke a single agent with timeout and error handling."""
    start = time.monotonic()
    try:
        response = await asyncio.wait_for(
            client.chat.completions.create(
                model=agent.model,
                messages=[
                    {"role": "system", "content": agent.system_prompt},
                    {"role": "user", "content": prompt}
                ]
            ),
            timeout=agent.timeout_seconds
        )
        return AgentResult(agent.name, response.choices[0].message.content,
                          (time.monotonic() - start) * 1000, True)
    except Exception as e:
        return AgentResult(agent.name, "",
                          (time.monotonic() - start) * 1000, False, str(e))

async def parallel_fan_out(agents, prompt, max_retries=2):
    """Fan out to multiple agents in parallel, retry failures."""
    tasks = [invoke_agent(a, prompt) for a in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception) or not r.success:
            for _ in range(max_retries):
                retry = await invoke_agent(agents[i], prompt)
                if retry.success:
                    results[i] = retry
                    break
    return [r for r in results if isinstance(r, AgentResult) and r.success]

async def synthesize(results, prompt, threshold=0.7, total_agents=3):
    """Synthesize results with consensus check."""
    if len(results) / total_agents < threshold:
        raise RuntimeError(f"Below consensus: {len(results)}/{total_agents}")
    outputs = "\n".join([f"[{r.agent_name}]: {r.output}" for r in results])
    resp = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "system", "content": "Synthesize these agent outputs."},
                  {"role": "user", "content": f"Question: {prompt}\n{outputs}"}]
    )
    return resp.choices[0].message.content

# Usage
agents = [
    AgentConfig("Risk Analyst", "Analyze financial risk. Focus on downside."),
    AgentConfig("Growth Analyst", "Analyze growth potential."),
    AgentConfig("Tech Analyst", "Analyze technical feasibility."),
]
final = asyncio.run(
    (lambda: parallel_fan_out(agents, "Expand fintech to Tier-2 Indian cities?"))()
)

# LangGraph orchestrator configuration (YAML)
orchestrator:
  name: support-orchestrator
  recursion_limit: 25
  
agents:
  - name: triage
    model: gpt-4o-mini
    temperature: 0.0
    max_tokens: 500
    role: router
    
  - name: billing
    model: gpt-4o
    temperature: 0.0
    max_tokens: 2000
    tools: [lookup_order, process_refund]
    
  - name: shipping
    model: gpt-4o
    temperature: 0.0
    max_tokens: 2000
    tools: [track_shipment, update_address]
    
  - name: escalation
    model: gpt-4o
    temperature: 0.0
    max_tokens: 3000
    tools: [create_ticket, process_large_refund]
    requires_human_approval: true

state:
  persistence: redis
  ttl_seconds: 3600
  
monitoring:
  trace_backend: langsmith
  cost_alert_threshold_usd: 0.50
  latency_p99_alert_ms: 30000
  
error_handling:
  max_retries: 2
  retry_backoff_ms: [1000, 3000]
  fallback_agent: escalation
  timeout_seconds: 45