Orchestrators

from dynamiq import Workflow
from dynamiq.connections import OpenAI as OpenAIConnection, ScaleSerp as ScaleSerpConnection
from dynamiq.flows import Flow
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI
from dynamiq.nodes.tools.scale_serp import ScaleSerpTool
from dynamiq.nodes.types import Behavior, InferenceMode

search_tool = ScaleSerpTool(connection=ScaleSerpConnection())  # reads SERP_API_KEY

llm = OpenAI(connection=OpenAIConnection(), model="gpt-4o", temperature=0.1)

research_agent = Agent(
    name="Research Analyst",
    role="Find recent market news and provide referenced highlights.",
    llm=llm,
    tools=[search_tool],
    inference_mode=InferenceMode.XML,
    max_loops=6,
    behaviour_on_max_loops=Behavior.RETURN,
)

writer_agent = Agent(
    name="Brief Writer",
    role="Turn research highlights into a concise executive brief.",
    llm=llm,
    inference_mode=InferenceMode.XML,
    max_loops=4,
    behaviour_on_max_loops=Behavior.RETURN,
)

manager_agent = Agent(
    name="Manager",
    role=(
        "Delegate research and writing to sub-agents.\n"
        "Always call tools with {'input': '<task>'} payloads and assemble the final brief."
    ),
    llm=llm,
    tools=[research_agent, writer_agent],
    inference_mode=InferenceMode.XML,
    parallel_tool_calls_enabled=True,
    max_loops=8,
    behaviour_on_max_loops=Behavior.RETURN,
)

workflow = Workflow(flow=Flow(nodes=[manager_agent]))
result = workflow.run(
    input_data={"input": "Summarize the latest developments in battery technology for investors."},
)
print(result.output[manager_agent.id]["output"]["content"])

from typing import Any

from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes import InputTransformer
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.agents.orchestrators.graph import END, START, GraphOrchestrator
from dynamiq.nodes.agents.orchestrators.graph_manager import GraphAgentManager
from dynamiq.nodes.llms import OpenAI

llm = OpenAI(connection=OpenAIConnection(), model="gpt-4o", temperature=0.1)

email_writer = Agent(
    name="email-writer-agent",
    llm=llm,
    role="Write personalized emails taking into account feedback.",
    input_transformer=InputTransformer(selector={"input": "$.context.agent_input"}),
)

def gather_feedback(context: dict[str, Any], **kwargs):
    """Gather feedback about the email draft."""
    draft = context.get("history", [{}])[-1].get("content", "No draft")
    feedback = input(f"Email draft:\n{draft}\nPress Enter to send, or type feedback to refine: \n")

    if feedback.strip() == "":
        return {"result": "Email was sent!", "agent_input": None}
    return {
        "result": "Email was canceled!",
        "agent_input": f"Draft of canceled email:\n{draft}\nUser feedback:\n{feedback}",
    }

def router(context: dict[str, Any], **kwargs):
    """Determine the next state based on the provided feedback."""
    if context.get("agent_input"):
        return "generate_sketch"
    return END

orchestrator = GraphOrchestrator(
    name="Graph orchestrator",
    manager=GraphAgentManager(llm=llm),
)

orchestrator.add_state_by_tasks("generate_sketch", [email_writer])
orchestrator.add_state_by_tasks("gather_feedback", [gather_feedback])

orchestrator.add_edge(START, "generate_sketch")
orchestrator.add_edge("generate_sketch", "gather_feedback")
orchestrator.add_conditional_edge("gather_feedback", ["generate_sketch", END], router)

orchestrator.run(input_data={"input": "Write and post email: invite the team to the offsite."})
print(orchestrator._chat_history[-1]["content"])