中文文档

Agent Protocol

A lightweight multi-agent collaboration framework built with pure Python and asyncio.

Features

• Agent Registry — Agents register capabilities and are discoverable by other agents
• Structured Tasks — Typed task definitions with priority, budget, deadline, and status tracking
• Smart Dispatcher — Capability-based routing, load balancing, priority queues, timeout & retry
• Message Bus — Async point-to-point and broadcast messaging with persistence
• Shared State — Key-value state board with event watchers, visible to all agents
• Task Pipeline — Chain tasks sequentially with automatic output-to-input passing
• Scoring System — Track agent performance history for intelligent dispatch
• Budget Control — Token and cost limits with automatic enforcement
• Event System — Local handlers + webhook HTTP callbacks
• DAG Scheduler — Dependency graph with parallel execution, cycle detection, ASCII visualization
• Retry & Fault Tolerance — Exponential backoff, fallback agents, dead letter queue
• Agent Sandbox — Isolated execution with resource limits and violation detection
• Monitor Dashboard — Real-time ASCII dashboard and JSON report export
• Protocol Versioning — Version negotiation, backward compatibility, graceful degradation
• Persistence — State snapshots, rollback, and auto-recovery
• CLI Tool — python -m agent_protocol demo|status|tasks|run

Quick Start

import asyncio
from agent_protocol import BaseAgent, Task, AgentRegistry, Dispatcher, MessageBus, SharedState

class MyAgent(BaseAgent):
    async def handle_task(self, task: Task) -> dict:
        return {"result": f"Processed: {task.input['data']}"}

async def main():
    registry = AgentRegistry()
    bus = MessageBus()
    state = SharedState()
    dispatcher = Dispatcher(registry, bus, state)

    agent = MyAgent("my-agent", capabilities=["process"])
    registry.register(agent)

    task = Task(type="process", input={"data": "hello"})
    result = await dispatcher.execute(task)
    print(result.result)  # {"result": "Processed: hello"}

asyncio.run(main())

DAG Dependency Graph

from agent_protocol import DAGScheduler

dag = DAGScheduler()
dag.add_node("collect", "collect", input_data={"source": "db"})
dag.add_node("clean", "clean", dependencies=["collect"])
dag.add_node("analyze", "analyze", dependencies=["clean"])
dag.add_node("report", "report", dependencies=["analyze"])

# Visualize
print(dag.visualize())

# Execute with parallel support
results = await dag.execute(dispatcher)

Retry & Fallback

from agent_protocol import RetryPolicy, FaultTolerantExecutor, DeadLetterQueue

policy = RetryPolicy(max_retries=3, base_delay=1.0, exponential_base=2.0)
executor = FaultTolerantExecutor(default_policy=policy)
executor.set_fallback("primary-agent", "backup-agent")

result = await executor.execute_with_retry(task, agent, registry=registry)

Agent Sandbox

from agent_protocol import SandboxManager, ResourceLimits

sandbox = SandboxManager()
limits = ResourceLimits(max_memory_mb=256, max_execution_time=30.0)
sandbox.create_sandbox("my-agent", limits=limits)

result = await sandbox.execute_sandboxed(agent, task)

Monitor Dashboard

from agent_protocol import MonitorDashboard

monitor = MonitorDashboard(registry=registry, scoring=scoring)
print(monitor.render_dashboard())  # ASCII table
report = monitor.export_json()     # JSON report

Protocol Versioning

from agent_protocol import VersionNegotiator, VersionedTask

negotiator = VersionNegotiator(current_version="1.0")
negotiator.register_agent_version("agent-1", "1.0")
result = negotiator.negotiate("agent-1")  # {"status": "match", ...}

Persistence & Snapshots

from agent_protocol import PersistenceManager

pm = PersistenceManager(persist_dir=".agent_data")
pm.save_system(agents=[...], tasks=[...], scoring={...})
pm.create_snapshot("v1", data={...}, description="Before upgrade")
pm.rollback("v1")  # Restore previous state

Running the Demos

# Basic demo
python examples/demo.py

# Advanced demo (DAG, retry, fallback, monitoring)
python examples/demo_advanced.py

# Simple example
python examples/simple.py

Running Tests

pip install pytest pytest-asyncio
python -m pytest tests/ -v

Architecture

agent_protocol/
├── core.py          # BaseAgent, Task, Message, enums
├── registry.py      # Agent discovery and management
├── dispatcher.py    # Task routing with priority queues
├── bus.py           # Async message bus
├── state.py         # Shared key-value state
├── pipeline.py      # Sequential task chains
├── scoring.py       # Agent performance tracking
├── budget.py        # Token/cost budget enforcement
├── events.py        # Event system with webhooks
├── dag.py           # DAG dependency scheduler
├── retry.py         # Retry policies, fallback, DLQ
├── sandbox.py       # Agent isolation & resource limits
├── monitor.py       # Dashboard & reporting
├── versioning.py    # Protocol version negotiation
├── persistence.py   # State snapshots & recovery
├── cli.py           # Command-line interface
└── utils.py         # Logging and helpers

License

MIT