Agent

from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI

llm = OpenAI(
    connection=OpenAIConnection(),  # reads OPENAI_API_KEY from the environment
    model="gpt-4o",
    temperature=0.3,
)

agent = Agent(
    name="assistant",
    llm=llm,
    role="Helpful assistant that answers questions concisely.",
    max_loops=10,
)

result = agent.run(input_data={"input": "What is the capital of France?"})
print(result.output.get("content"))

from dynamiq.connections import E2B as E2BConnection, OpenAI as OpenAIConnection
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI
from dynamiq.nodes.tools.e2b_sandbox import E2BInterpreterTool

e2b_tool = E2BInterpreterTool(connection=E2BConnection())  # reads E2B_API_KEY

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

agent = Agent(
    name="data-agent",
    llm=llm,
    tools=[e2b_tool],
    role="Senior Data Scientist",
    max_loops=10,
)

result = agent.run(
    input_data={"input": "Add the first 10 numbers and tell if the result is prime."}
)
print(result.output.get("content"))

from dynamiq import Workflow
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.flows import Flow
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI

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

agent = Agent(
    name="assistant",
    id="agent",
    llm=llm,
    role="Helpful assistant that answers concisely.",
)

wf = Workflow(flow=Flow(nodes=[agent]))
result = wf.run(input_data={"input": "Summarize the difference between TCP and UDP in two sentences."})
print(result.output[agent.id]["output"]["content"])

REFLECTIVE_ROLE = """
You are a meticulous technical writer.
Before giving your final answer:
1. Draft a response.
2. Critique the draft: check factual accuracy, missing caveats, and clarity.
3. Revise the draft to address every critique point.
Only return the revised version as your final answer.
"""

agent = Agent(
    name="reviewer",
    id="reviewer",
    llm=llm,
    role=REFLECTIVE_ROLE,
    max_loops=4,
)

result = agent.run(
    input_data={"input": "How are sin(x) and cos(x) connected in electrodynamics?"}
)
print(result.output["content"])

from dynamiq.nodes.types import Behavior, InferenceMode

agent = Agent(
    name="researcher",
    llm=llm,
    tools=[e2b_tool],
    inference_mode=InferenceMode.XML,
    behaviour_on_max_loops=Behavior.RETURN,
    max_loops=8,
)

from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes.agents import Agent
from dynamiq.nodes.llms import OpenAI

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

agent = Agent(
    name="vision-assistant",
    llm=llm,
    role="Describes the contents of images concisely.",
)

result = agent.run(
    input_data={
        "input": "What is shown in this image?",
        "images": ["https://example.com/photo.jpg"],
    }
)
print(result.output.get("content"))

result = agent.run(
    input_data={
        "input": "Search for the latest earnings report.",
        "tool_params": {
            "global": {"user_region": "EU"},
            "by_name": {"search-tool": {"num_results": 10}},
            "by_id": {"tool-node-id-123": {"api_key": "value-with-highest-priority"}},
        },
    }
)

from pydantic import BaseModel

class Verdict(BaseModel):
    summary: str
    confidence: float

agent = Agent(name="judge", llm=llm, response_format=Verdict)

result = agent.run(input_data={"input": "Is this invoice complete? Items: ..."})
print(result.output["content"])  # {'summary': '...', 'confidence': 0.92}

from dynamiq.nodes.agents.utils import SummarizationConfig

agent = Agent(
    name="long-task-agent",
    llm=llm,
    tools=[e2b_tool],
    max_loops=40,
    summarization_config=SummarizationConfig(
        enabled=True,
        context_usage_ratio=0.8,      # summarize when 80% of the context window is used
        max_preserved_tokens=10000,   # keep this many tokens of recent messages verbatim
    ),
)

from dynamiq.storages.file.base import FileStoreConfig
from dynamiq.storages.file.in_memory import InMemoryFileStore

agent = Agent(
    name="file-agent",
    llm=llm,
    file_store=FileStoreConfig(
        enabled=True,
        backend=InMemoryFileStore(),
        agent_file_write_enabled=True,  # allow writes, not just reads
        todo_enabled=False,             # optionally add todo-management tools
    ),
)

from dynamiq.connections import E2B as E2BConnection
from dynamiq.sandboxes import SandboxConfig
from dynamiq.sandboxes.e2b import E2BSandbox

sandbox = E2BSandbox(connection=E2BConnection(), timeout=3600, base_path="/home/user")

agent = Agent(
    name="sandbox-agent",
    llm=llm,
    sandbox=SandboxConfig(enabled=True, backend=sandbox),
    max_loops=10,
)

from dynamiq import Workflow
from dynamiq.callbacks.streaming import StreamingIteratorCallbackHandler
from dynamiq.flows import Flow
from dynamiq.runnables import RunnableConfig
from dynamiq.types.streaming import StreamingConfig, StreamingMode

agent = Agent(
    name="research-assistant",
    llm=llm,
    role="Research assistant that explains its reasoning step by step.",
    streaming=StreamingConfig(enabled=True, mode=StreamingMode.ALL),
    max_loops=5,
)

handler = StreamingIteratorCallbackHandler()
wf = Workflow(flow=Flow(nodes=[agent]))
wf.run(
    input_data={"input": "Summarize the pros and cons of vector databases."},
    config=RunnableConfig(callbacks=[handler]),
)

for chunk in handler:
    print(chunk)

result = agent.run(input_data={"input": "Write a CSV with the first 5 primes and return it."})
answer = result.output["content"]
files = result.output.get("files", [])