如何在 Python 中让 AI 智能体彼此对话

pip install -U swarms

from swarms import FileChannel

channel = FileChannel("runs/hello.md")
channel.clear()  # start from an empty file every time you run this

channel.post("Human", "What should we call the new CLI?")
channel.post("Namer", "Three options: swarmctl, hive, colony.")

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

from swarms import Agent, FileChannel

channel = FileChannel("runs/vector_db_report.md")
channel.clear()  # this script starts a fresh conversation

researcher = Agent(
    agent_name="Researcher",
    system_prompt="You gather facts and figures. Answer in a short bullet list.",
    model_name="gemini/gemini-2.5-pro",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)
critic = Agent(
    agent_name="Critic",
    system_prompt=(
        "You read the conversation so far and list every claim that is "
        "unsupported or likely wrong, with one line on why."
    ),
    model_name="deepseek/deepseek-v4-pro",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)
writer = Agent(
    agent_name="Writer",
    system_prompt=(
        "You write a short report from the research and the critique in the "
        "conversation. Leave out any claim the critique flagged."
    ),
    model_name="anthropic/claude-sonnet-5-5",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)

task = "Compare pgvector and Qdrant for a 10M-vector semantic search workload."
for agent in (researcher, critic, writer):
    agent.run(task)

for message in channel.messages():
    print(f"--- {message['agent_name']} ---")
    print(message["content"][:300])

from swarms import Agent, FileChannel

channel = FileChannel("runs/vector_db_report.md")
print(f"{len(channel.messages())} messages already in the channel")

editor = Agent(
    agent_name="Editor",
    system_prompt="You cut reports to half their length and keep every number.",
    model_name="gpt-5.4",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)

print(editor.run("Cut the report above to five sentences."))

from swarms import Agent, SequentialWorkflow, SqliteChannel


def make_agent(name: str, model: str, prompt: str) -> Agent:
    return Agent(
        agent_name=name,
        system_prompt=prompt,
        model_name=model,
        max_loops=1,
        print_on=False,
    )


channel = SqliteChannel("runs/channels.db", name="pricing-page")

workflow = SequentialWorkflow(
    agents=[
        make_agent("Analyst", "gpt-5.4-mini", "List what buyers compare when they read an API pricing page."),
        make_agent("Copywriter", "anthropic/claude-sonnet-5-5", "Write pricing page copy from the analysis in the conversation."),
        make_agent("Reviewer", "gemini/gemini-2.5-pro", "Point out anything in the copy that is vague or could mislead a buyer."),
    ],
    max_loops=1,
    channel=channel,
)
workflow.run("Draft the pricing page copy for a usage-based API.")

for row in channel.fetch_messages():
    print(row["id"], row["created_at"], row["agent_name"], row["content"][:50])

print(SqliteChannel.list_conversations("runs/channels.db"))
channel.close()

import threading

from swarms import Agent, ConcurrentWorkflow, QueueChannel


def make_agent(name: str, model: str, prompt: str) -> Agent:
    return Agent(
        agent_name=name,
        system_prompt=prompt,
        model_name=model,
        max_loops=1,
        print_on=False,
    )


channel = QueueChannel()
workflow = ConcurrentWorkflow(
    agents=[
        make_agent("Bull", "gpt-5.4-mini", "Make the strongest case for the trade."),
        make_agent("Bear", "groq/openai/gpt-oss-120b", "Make the strongest case against the trade."),
        make_agent("Risk", "anthropic/claude-haiku-5-5", "List the risks a desk manager would ask about."),
    ],
    channel=channel,
)


def produce():
    try:
        workflow.run("Should we buy two-year US Treasuries this week?")
    finally:
        channel.close()  # ends the consumer loop, even if the run fails


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

for message in channel:  # blocks until the next post
    print(f"[{message['agent_name']}] {message['content'][:80]}")

print(f"{len(channel.messages())} messages in the history")

from swarms import Agent, HttpChannel

channel = HttpChannel(
    "http://localhost:8077/channels/ev-report",
    headers={"Authorization": "Bearer dev-token"},
)

agent = Agent(
    agent_name="Researcher",
    model_name="mistral/mistral-large-latest",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)
print(agent.run("Summarize the EV battery market in three bullets."))
print([m["agent_name"] for m in channel.messages()])
channel.close()

import json
from pathlib import Path

from swarms import Agent, CommsChannel


class JsonlChannel(CommsChannel):
    """One JSON object per line: easy to grep, diff and load into pandas."""

    def __init__(self, path: str):
        self.path = Path(path)

    def post(self, agent_name: str, content: str) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        row = {"agent_name": agent_name, "content": str(content)}
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(row) + "\n")

    def messages(self):
        if not self.path.exists():
            return []
        with self.path.open(encoding="utf-8") as f:
            return [json.loads(line) for line in f if line.strip()]

    def clear(self) -> None:
        self.path.unlink(missing_ok=True)


channel = JsonlChannel("runs/triage.jsonl")
channel.clear()

triager = Agent(
    agent_name="Triager",
    system_prompt="Classify the bug report by severity and component.",
    model_name="gpt-5.4-mini",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)
fixer = Agent(
    agent_name="Fixer",
    system_prompt="Propose a fix for the bug, using the triage in the conversation.",
    model_name="anthropic/claude-opus-5-5",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)

report = "Uploading a 2 GB file returns HTTP 500 after 30 seconds."
triager.run(report)
fixer.run(report)

print(channel.path.read_text())

pip install -U swarms