Multi-agent coordination

Agents don't message each other directly — they share state on a Blackboard. Reads are targeted, writes are logged, and the loop itself burns no LLM tokens. Source: ragforge/coordination/.

How it works

The system uses stigmergy: each agent reads the board, decides if it should fire via a trigger_fn(board) → bool, and writes results back. The Orchestrator runs a deterministic loop:

Check each agent's trigger function.
Fire all eligible agents in sequence.
Repeat until goal(board) is met, quiescence, or max_steps.

Each entry carries key, value, author, timestamp, tags, and an auto-incrementing version for conflict detection. The persistent backend is SQLite (WAL mode).

Working example — 3 agents

From examples/multi_agent_coordination.py.

python

from ragforge.coordination.blackboard import InMemoryBlackboard
from ragforge.coordination.agent import Agent, Orchestrator, AgentResult

board = InMemoryBlackboard("research-task")
board.write("question", "What is RAGForge's approach to coordination?", author="user")

def researcher_trigger(board):
    return board.has_key("question") and not board.has_key("findings")

def researcher_action(board, agent_id):
    question = board.read("question")
    findings = "RAGForge uses blackboard-based stigmergy..."
    board.write("findings", findings, author=agent_id,
                tags={"confidence": 0.85, "status": "pending_review"})
    return AgentResult(agent_id=agent_id, entries_read=["question"],
                       entries_written=["findings"], tokens_used=120)

def critic_trigger(board):
    entries = board.read_by_tag("status", lambda v: v == "pending_review")
    return len(entries) > 0 and not board.has_key("review")

def critic_action(board, agent_id):
    findings = board.read("findings")
    if findings.tags.get("confidence", 0) > 0.7:
        board.write("review", "Approved.", author=agent_id,
                    tags={"status": "approved"})
    return AgentResult(agent_id=agent_id, entries_read=["findings"],
                       entries_written=["review"], tokens_used=60)

def writer_trigger(board):
    review = board.read("review")
    return review and review.tags.get("status") == "approved" and not board.has_key("final_answer")

def writer_action(board, agent_id):
    board.write("final_answer", "RAGForge uses stigmergy...", author=agent_id,
                tags={"status": "complete"})
    return AgentResult(agent_id=agent_id, entries_read=["findings", "review"],
                       entries_written=["final_answer"], tokens_used=180)

agents = [
    Agent("researcher", researcher_trigger, researcher_action),
    Agent("critic",     critic_trigger,     critic_action),
    Agent("writer",     writer_trigger,     writer_action),
]

goal = lambda board: board.has_key("final_answer")
orch = Orchestrator(board, agents, goal=goal, max_steps=10)
result = orch.run()

print(f"Steps: {len(result.steps)}  Tokens: {result.total_tokens}  Cost: ${result.total_cost_usd:.4f}")
print(board.read("final_answer").value)

Run from the CLI

bash

ragforge agents run config.py --max-steps 20 --persist
ragforge agents benchmark config.py
ragforge agents board research-task

HTTP API

See Coordination endpoints for the full REST surface.

← Previous

Evaluation

Migration