如何让多台机器上的 AI 智能体共享同一段对话

"""A channel server for Swarms HttpChannel: bearer-token auth, SQLite storage, standard library only."""

import hmac
import json
import os
import re
import sqlite3
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

TOKEN = os.environ["CHANNEL_TOKEN"]
DB_PATH = os.environ.get("CHANNEL_DB", "channels.db")
HOST = os.environ.get("CHANNEL_HOST", "127.0.0.1")
PORT = int(os.environ.get("CHANNEL_PORT", "8765"))
MAX_BODY = 1_000_000  # bytes per message
ROUTE = re.compile(r"^/channels/([A-Za-z0-9_.-]{1,100})$")

db = sqlite3.connect(DB_PATH, check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute(
    "CREATE TABLE IF NOT EXISTS messages ("
    " id INTEGER PRIMARY KEY AUTOINCREMENT,"
    " channel TEXT NOT NULL,"
    " agent_name TEXT NOT NULL,"
    " content TEXT NOT NULL,"
    " created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)"
)
db.execute("CREATE INDEX IF NOT EXISTS by_channel ON messages (channel, id)")
db.commit()
lock = threading.Lock()  # one writer at a time on the shared connection


class Handler(BaseHTTPRequestHandler):
    def _reply(self, status, body=None):
        data = json.dumps(body).encode() if body is not None else b""
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def _channel(self):
        """Check the token and the path. Returns the channel name, or None after replying."""
        sent = self.headers.get("Authorization", "")
        if not hmac.compare_digest(sent.encode(), f"Bearer {TOKEN}".encode()):
            self._reply(401, {"error": "missing or wrong bearer token"})
            return None
        match = ROUTE.match(self.path)
        if match is None:
            self._reply(404, {"error": "use /channels/<name>"})
            return None
        return match.group(1)

    def do_GET(self):
        name = self._channel()
        if name is None:
            return
        with lock:
            rows = db.execute(
                "SELECT agent_name, content FROM messages WHERE channel = ? ORDER BY id",
                (name,),
            ).fetchall()
        self._reply(200, [{"agent_name": a, "content": c} for a, c in rows])

    def do_POST(self):
        name = self._channel()
        if name is None:
            return
        length = int(self.headers.get("Content-Length") or 0)
        if length > MAX_BODY:
            self._reply(413, {"error": "message too large"})
            return
        try:
            row = json.loads(self.rfile.read(length))
            agent_name, content = row["agent_name"], row["content"]
            if not (isinstance(agent_name, str) and isinstance(content, str)):
                raise TypeError
        except (ValueError, KeyError, TypeError):
            self._reply(400, {"error": 'send {"agent_name": str, "content": str}'})
            return
        with lock:
            db.execute(
                "INSERT INTO messages (channel, agent_name, content) VALUES (?, ?, ?)",
                (name, agent_name, content),
            )
            db.commit()
        self._reply(201, {"agent_name": agent_name, "content": content})

    def do_DELETE(self):
        name = self._channel()
        if name is None:
            return
        with lock:
            db.execute("DELETE FROM messages WHERE channel = ?", (name,))
            db.commit()
        self._reply(204)


if __name__ == "__main__":
    print(f"Serving channels on http://{HOST}:{PORT}/channels/<name>")
    ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()

export CHANNEL_TOKEN="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"
python channel_server.py

URL=http://127.0.0.1:8765/channels/battery-brief

curl -i $URL                       # HTTP/1.0 401 Unauthorized
curl -X POST $URL -H "Authorization: Bearer $CHANNEL_TOKEN" \
     -d '{"agent_name": "Ops", "content": "hello"}'
# {"agent_name": "Ops", "content": "hello"}
curl $URL -H "Authorization: Bearer $CHANNEL_TOKEN"
# [{"agent_name": "Ops", "content": "hello"}]
curl -i -X DELETE $URL -H "Authorization: Bearer $CHANNEL_TOKEN"   # HTTP/1.0 204 No Content

"""Run one agent of the team against a shared HttpChannel.

Usage: python agent_node.py researcher|analyst|writer
"""

import os
import sys
import time

from swarms import Agent, HttpChannel

TASK = "Write a one-page brief on home battery storage for a utility's product team."

ROLES = {
    "researcher": {
        "agent_name": "Researcher",
        "model_name": "gpt-5.4-mini",
        "system_prompt": "You gather facts, figures and open questions. Answer in a short bullet list.",
        "waits_for": None,
    },
    "analyst": {
        "agent_name": "Analyst",
        "model_name": "deepseek/deepseek-v4-pro",
        "system_prompt": "You read the research in the conversation and list the risks, costs and trade-offs.",
        "waits_for": "Researcher",
    },
    "writer": {
        "agent_name": "Writer",
        "model_name": "anthropic/claude-sonnet-5-5",
        "system_prompt": "You turn the research and analysis in the conversation into a one-page brief.",
        "waits_for": "Analyst",
    },
}


def wait_for(channel, agent_name, poll_seconds=2.0, timeout_seconds=900):
    """Block until `agent_name` has posted to the channel."""
    deadline = time.monotonic() + timeout_seconds
    while not any(m["agent_name"] == agent_name for m in channel.messages()):
        if time.monotonic() > deadline:
            raise TimeoutError(f"{agent_name} did not post within {timeout_seconds}s")
        time.sleep(poll_seconds)


role = ROLES[sys.argv[1]]
channel = HttpChannel(
    os.environ["CHANNEL_URL"],
    headers={"Authorization": f"Bearer {os.environ['CHANNEL_TOKEN']}"},
    timeout=30.0,
)

if role["waits_for"]:
    print(f"Waiting for {role['waits_for']} ...", flush=True)
    wait_for(channel, role["waits_for"])

agent = Agent(
    agent_name=role["agent_name"],
    system_prompt=role["system_prompt"],
    model_name=role["model_name"],
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)
print(agent.run(TASK))
channel.close()

export CHANNEL_URL=http://127.0.0.1:8765/channels/battery-brief
export CHANNEL_TOKEN=...   # the same token the server uses

python agent_node.py writer       # machine C: prints "Waiting for Analyst ..."
python agent_node.py analyst      # machine B: prints "Waiting for Researcher ..."
python agent_node.py researcher   # machine A: runs at once

import os

from swarms import HttpChannel

channel = HttpChannel(
    os.environ["CHANNEL_URL"],
    headers={"Authorization": f"Bearer {os.environ['CHANNEL_TOKEN']}"},
)
for message in channel.messages():
    print(f"\n## {message['agent_name']}\n{message['content']}")
channel.close()

## Human
Write a one-page brief on home battery storage for a utility's product team.

## Researcher
- ...

## Analyst
- ...

## Writer
...

import os

from swarms import Agent, HttpChannel, SequentialWorkflow

channel = HttpChannel(
    os.environ["CHANNEL_URL"],
    headers={"Authorization": f"Bearer {os.environ['CHANNEL_TOKEN']}"},
)

researcher = Agent(
    agent_name="Researcher",
    system_prompt="You gather facts, figures and open questions. Answer in a short bullet list.",
    model_name="groq/openai/gpt-oss-120b",
    max_loops=1,
    print_on=False,
)
analyst = Agent(
    agent_name="Analyst",
    system_prompt="You read the research in the conversation and list the risks, costs and trade-offs.",
    model_name="gemini/gemini-2.5-pro",
    max_loops=1,
    print_on=False,
)

workflow = SequentialWorkflow(agents=[researcher, analyst], max_loops=1, channel=channel)
workflow.run("Assess home battery storage for a utility's product team.")
print(f"{len(channel.messages())} messages on the channel")

import os

from swarms import Agent, HttpChannel

channel = HttpChannel(
    os.environ["CHANNEL_URL"],
    headers={"Authorization": f"Bearer {os.environ['CHANNEL_TOKEN']}"},
)

reviewer = Agent(
    agent_name="Reviewer",
    system_prompt="You check the analysis in the conversation for gaps and unsupported claims.",
    model_name="anthropic/claude-opus-5-5",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)
print(reviewer.run("Review the analysis above. List anything that needs a source."))

"""The same channel server in FastAPI. Run: uvicorn channel_server_fastapi:app --port 8765"""

import hmac
import os
import sqlite3
import threading

from fastapi import Depends, FastAPI, Header, HTTPException, Path, Response
from pydantic import BaseModel

TOKEN = os.environ["CHANNEL_TOKEN"]
db = sqlite3.connect(os.environ.get("CHANNEL_DB", "channels.db"), check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute(
    "CREATE TABLE IF NOT EXISTS messages (id INTEGER PRIMARY KEY AUTOINCREMENT,"
    " channel TEXT NOT NULL, agent_name TEXT NOT NULL, content TEXT NOT NULL,"
    " created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)"
)
db.commit()
lock = threading.Lock()
app = FastAPI()
Name = Path(pattern=r"^[A-Za-z0-9_.-]{1,100}$")


class Message(BaseModel):
    agent_name: str
    content: str


def check_token(authorization: str = Header(default="")):
    if not hmac.compare_digest(authorization.encode(), f"Bearer {TOKEN}".encode()):
        raise HTTPException(status_code=401, detail="missing or wrong bearer token")


@app.get("/channels/{name}", dependencies=[Depends(check_token)])
def read(name: str = Name) -> list[Message]:
    with lock:
        rows = db.execute(
            "SELECT agent_name, content FROM messages WHERE channel = ? ORDER BY id", (name,)
        ).fetchall()
    return [Message(agent_name=a, content=c) for a, c in rows]


@app.post("/channels/{name}", status_code=201, dependencies=[Depends(check_token)])
def append(message: Message, name: str = Name) -> Message:
    with lock:
        db.execute(
            "INSERT INTO messages (channel, agent_name, content) VALUES (?, ?, ?)",
            (name, message.agent_name, message.content),
        )
        db.commit()
    return message


@app.delete("/channels/{name}", status_code=204, dependencies=[Depends(check_token)])
def clear(name: str = Name) -> Response:
    with lock:
        db.execute("DELETE FROM messages WHERE channel = ?", (name,))
        db.commit()
    return Response(status_code=204)

pip install -U swarms