用 SQLite 作为 AI 智能体的共享记忆

PRAGMA journal_mode=WAL;

CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY,
    channel TEXT NOT NULL,
    agent_name TEXT NOT NULL,
    content TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS messages_channel ON messages (channel, id);

pip install -U swarms

from swarms import Agent, SqliteChannel

channel = SqliteChannel("agent_memory.db", name="ev-batteries")

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,
)
writer = Agent(
    agent_name="Writer",
    system_prompt="Turn the facts in the conversation into a three-sentence summary.",
    model_name="anthropic/claude-sonnet-5-5",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)

task = "Summarize the state of solid-state EV batteries."
researcher.run(task)
print(writer.run(task))

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

from swarms import Agent, SqliteChannel

# A later run, in a new process: same file, same conversation name.
channel = SqliteChannel("agent_memory.db", name="ev-batteries")
print(len(channel.messages()), "messages already in the conversation")

editor = Agent(
    agent_name="Editor",
    system_prompt="You edit drafts for length and clarity.",
    model_name="gpt-5.4",
    max_loops=1,
    output_type="final",
    print_on=False,
    channel=channel,
)

print(editor.run("Cut the summary above to one sentence."))
print(len(channel.messages()), "messages now")

from swarms import Agent, SqliteChannel

DB = "agent_memory.db"
tickets = {
    "ticket-1041": "My March invoice charged me twice.",
    "ticket-1042": "The export button returns a 500 error.",
}

for name, text in tickets.items():
    channel = SqliteChannel(DB, name=name)  # one conversation per ticket, one file
    triage = Agent(
        agent_name="Triage",
        system_prompt="Classify the ticket as billing, bug or other, and say why in one line.",
        model_name="groq/openai/gpt-oss-120b",  # fast and cheap for high-volume triage
        max_loops=1,
        output_type="final",
        print_on=False,
        user_name="Customer",  # the task is posted under this name
        channel=channel,
    )
    triage.run(text)
    channel.close()

print(SqliteChannel.list_conversations(DB))
print(SqliteChannel(DB, name="ticket-1042").messages())

from swarms import Agent, SequentialWorkflow, SqliteChannel

channel = SqliteChannel("agent_memory.db", name="semiconductor-brief")

researcher = Agent(
    agent_name="Researcher",
    system_prompt="Collect the key facts and figures as a bullet list.",
    model_name="gemini/gemini-2.5-pro",  # long context for source material
    max_loops=1,
    print_on=False,
)
analyst = Agent(
    agent_name="Analyst",
    system_prompt="Interpret the facts above: trends, risks, open questions.",
    model_name="deepseek/deepseek-v4-pro",
    max_loops=1,
    print_on=False,
)
writer = Agent(
    agent_name="Writer",
    system_prompt="Write a one-page brief from the conversation.",
    model_name="anthropic/claude-sonnet-5-5",
    max_loops=1,
    print_on=False,
)

workflow = SequentialWorkflow(
    agents=[researcher, analyst, writer],
    max_loops=1,
    channel=channel,
)
workflow.run("Brief me on the US semiconductor sector.")

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

print("Writer's latest message:", channel.fetch_messages(agent_name="Writer", limit=1)[0]["id"])

import sqlite3

# Read-only connection: analytics can never modify agent memory.
conn = sqlite3.connect("file:agent_memory.db?mode=ro", uri=True)

print("Messages per speaker, per conversation:")
for row in conn.execute(
    """
    SELECT channel, agent_name, COUNT(*) AS n
    FROM messages
    GROUP BY channel, agent_name
    ORDER BY channel, n DESC
    """
):
    print(" ", row)

print("Every message that mentions a 500 error:")
for row in conn.execute(
    """
    SELECT channel, id, agent_name, substr(content, 1, 60)
    FROM messages
    WHERE content LIKE ?
    ORDER BY id
    """,
    ("%500 error%",),
):
    print(" ", row)

print("Latest message in each conversation:")
for row in conn.execute(
    """
    SELECT m.channel, m.agent_name, m.created_at
    FROM messages AS m
    JOIN (SELECT channel, MAX(id) AS id FROM messages GROUP BY channel) AS last
      ON m.id = last.id
    ORDER BY m.id
    """
):
    print(" ", row)

print("Conversations active in the last 24 hours (created_at is UTC):")
for row in conn.execute(
    """
    SELECT channel, COUNT(*)
    FROM messages
    WHERE created_at >= datetime('now', '-1 day')
    GROUP BY channel
    """
):
    print(" ", row)

conn.close()

import os
import sqlite3
from concurrent.futures import ProcessPoolExecutor

from swarms import Agent, SqliteChannel

DB = "support.db"
MODELS = [
    "gpt-5.4-mini",
    "anthropic/claude-haiku-5-5",
    "mistral/mistral-large-latest",
    "groq/openai/gpt-oss-120b",
]


def handle_ticket(job):
    ticket_id, text, model = job
    # Each process opens its own connection; never pass a channel between processes.
    channel = SqliteChannel(DB, name=ticket_id)
    ops_log = SqliteChannel(DB, name="ops-log")  # one conversation every process writes to
    try:
        agent = Agent(
            agent_name="Support-Agent",
            system_prompt="Draft a short, friendly first reply to the customer.",
            model_name=model,
            max_loops=1,
            output_type="final",
            print_on=False,
            user_name="Customer",
            channel=channel,
        )
        agent.run(text)
        ops_log.post("Support-Agent", f"{ticket_id} answered by {model} in pid {os.getpid()}")
    finally:
        channel.close()
        ops_log.close()
    return ticket_id


if __name__ == "__main__":
    # Open the file once up front so the table and WAL mode exist before workers start.
    SqliteChannel(DB, name="ops-log").close()

    jobs = [
        (f"ticket-{2000 + i}", f"Customer message number {i}", MODELS[i % len(MODELS)])
        for i in range(8)
    ]
    with ProcessPoolExecutor(max_workers=4) as pool:
        finished = list(pool.map(handle_ticket, jobs))

    conn = sqlite3.connect(DB)
    print("tickets answered:", len(finished))
    print("conversations:", len(SqliteChannel.list_conversations(DB)))
    print(
        "processes that wrote to ops-log:",
        conn.execute(
            "SELECT COUNT(DISTINCT substr(content, instr(content, 'pid '))) "
            "FROM messages WHERE channel = 'ops-log'"
        ).fetchone()[0],
    )
    print(
        "rows per ticket:",
        conn.execute(
            "SELECT MIN(n), MAX(n) FROM (SELECT COUNT(*) AS n FROM messages "
            "WHERE channel LIKE 'ticket-%' GROUP BY channel)"
        ).fetchone(),
    )
    conn.close()

import json
import sqlite3

from swarms import SqliteChannel

DB = "agent_memory.db"

# Archive one finished conversation to JSON Lines, then remove it from the database.
ticket = SqliteChannel(DB, name="ticket-1041")
with open("ticket-1041.jsonl", "w") as f:
    for row in ticket.fetch_messages():
        f.write(json.dumps(row) + "\n")
ticket.clear()  # deletes this conversation's rows only
ticket.close()

# Age-based retention across every conversation is plain SQL on the same table.
conn = sqlite3.connect(DB)
with conn:
    expired = conn.execute(
        "DELETE FROM messages WHERE created_at < datetime('now', '-30 days')"
    ).rowcount
conn.execute("VACUUM")  # give the freed pages back to the filesystem
conn.close()

print("expired rows deleted:", expired)
print(SqliteChannel.list_conversations(DB))

pip install -U swarms