AGNO Integration Guide¶
Integrate ACGP with AGNO for multi-agent orchestration governance.
Overview¶
AGNO provides practical multi-agent patterns with simpler abstractions than rolling your own. ACGP adds governance to agents, teams, and routing decisions.
Installation¶
Basic Usage¶
from agno import Agent
from acgp_agno import GovernedAgent
from acgp import GovernanceSteward
# Your AGNO agent
agent = Agent(name="assistant", role="helpful assistant")
# Add governance
steward = GovernanceSteward(blueprint_file="blueprint.yaml")
governed_agent = GovernedAgent(agent, steward)
# Execute with governance
result = governed_agent.run("Process this request")
Features¶
Agent Governance¶
governed_agent = GovernedAgent(agent, steward, agent_name="assistant")
result = governed_agent.run("Execute task")
Team Governance¶
from acgp_agno import GovernedTeam
team = Team(agents=[agent1, agent2, agent3])
governed_team = GovernedTeam(team, steward)
result = governed_team.run("Multi-agent task")
Router Governance¶
from acgp_agno import GovernedRouter
router = MyAGNORouter()
governed_router = GovernedRouter(router, steward)
destination = governed_router.route(input_data)
Memory Integration¶
from acgp_agno import ACGPMemoryIntegration
memory = ACGPMemoryIntegration(agno_memory_backend)
agent = GovernedAgent(agent, steward, memory_integration=memory)
# Access history
history = memory.get_intervention_history()
trust_debt = memory.get_trust_debt()
Async Support¶
For async/await support:
from acgp_agno import AsyncGovernedAgent, AsyncGovernedTeam
# Async agent
async_agent = AsyncGovernedAgent(agent, steward)
result = await async_agent.run("Async task")
# Async team
async_team = AsyncGovernedTeam(team, steward)
result = await async_team.run("Multi-agent async task")
Advanced Features¶
Override Blocked Actions¶
# After human approval
result = governed_agent.run_with_override(
trace_id="blocked_trace_id",
task="Previously blocked task"
)
Get Governance Metrics¶
metrics = governed_agent.get_metrics()
print(f"Interventions: {metrics['intervention_count']}")
print(f"Trust debt: {metrics['trust_debt']}")
Exception Handling¶
from acgp_agno import InterventionError, AgentGovernanceError
try:
result = governed_agent.run("Risky task")
except InterventionError as e:
print(f"Action {e.decision}: {e.message}")
print(f"Trace ID: {e.trace_id}")
Examples¶
Example implementations available in the acgp-agno package:
- Basic Agent: Simple AGNO agent with governance
- Multi-Agent Team: Team of agents working together
- Router Governance: Governed routing decisions
- Async Operations: Async agent and team examples