QueueChannel Quickstart

import threading

from swarms import Agent, QueueChannel

channel = QueueChannel()

agent = Agent(
    agent_name="Researcher",
    system_prompt="Gather the key facts as a short bullet list.",
    model_name="gpt-5.4",
    max_loops=1,
    channel=channel,
)


def produce():
    try:
        agent.run("List five facts about solid-state batteries.")
    finally:
        channel.close()


threading.Thread(target=produce, daemon=True).start()

for message in channel:
    print(f"[{message['agent_name']}]\n{message['content']}\n")

import threading

from swarms import Agent, QueueChannel, SequentialWorkflow

channel = QueueChannel()

workflow = SequentialWorkflow(
    agents=[
        Agent(
            agent_name="Researcher",
            system_prompt="Gather the key facts as a short bullet list.",
            model_name="gpt-5.4",
            max_loops=1,
        ),
        Agent(
            agent_name="Analyst",
            system_prompt="Weigh the facts in the conversation and state the main risk.",
            model_name="gpt-5.4",
            max_loops=1,
        ),
        Agent(
            agent_name="Writer",
            system_prompt="Turn the conversation into a three-paragraph brief.",
            model_name="gpt-5.4",
            max_loops=1,
        ),
    ],
    max_loops=1,
    channel=channel,
)


def produce():
    try:
        workflow.run("Report on the EV battery market.")
    finally:
        channel.close()


threading.Thread(target=produce, daemon=True).start()

for message in channel:
    print(f"\n[{message['agent_name']}]\n{message['content']}")

import threading

from swarms import Agent, GroupChat, QueueChannel


def speaker(name, stance):
    return Agent(
        agent_name=name,
        system_prompt=f"You argue for {stance}. Keep each message short.",
        model_name="gpt-5.4",
        max_loops=1,
    )


channel = QueueChannel()

chat = GroupChat(
    agents=[
        speaker("Optimist", "the benefits"),
        speaker("Pessimist", "the risks"),
        speaker("Realist", "a balanced view"),
    ],
    max_loops=6,
    channel=channel,
)

result = {}


def produce():
    try:
        result["output"] = chat.run("Should hospitals adopt AI for triage?")
    finally:
        channel.close()


threading.Thread(target=produce, daemon=True).start()

for message in channel:
    print(f"{message['agent_name']}: {message['content']}\n")

print("Final output:", result["output"])

import threading

from swarms import Agent, ConcurrentWorkflow, QueueChannel

channel = QueueChannel()

workflow = ConcurrentWorkflow(
    agents=[
        Agent(agent_name=f"{angle}-Analyst", model_name="gpt-5.4", max_loops=1)
        for angle in ["Technology", "Regulation", "Supply-Chain"]
    ],
    channel=channel,
)

done = threading.Event()


def produce():
    try:
        workflow.run("What will shape the EV battery market over the next five years?")
    finally:
        done.set()
        channel.close()


threading.Thread(target=produce, daemon=True).start()

while True:
    message = channel.wait(timeout=5)
    if message is not None:
        print(f"[{message['agent_name']}] {message['content'][:100]}")
    elif done.is_set():
        break
    else:
        print("... still working")

import json

with open("runs/transcript.json", "w") as f:
    json.dump(channel.messages(), f, indent=2)