Autogen Integration Guide¶
Integrate ACGP with Autogen for conversational multi-agent governance.
Overview¶
Autogen enables agents to chat with each other and call functions. ACGP provides governance for messages and function calls, ensuring safe multi-agent conversations.
Installation¶
Basic Usage¶
from autogen import ConversableAgent
from acgp_autogen import GovernedConversableAgent
from acgp import GovernanceSteward
# Your Autogen agent
agent = ConversableAgent(
name="assistant",
llm_config={"model": "gpt-4"}
)
# Add governance
steward = GovernanceSteward(blueprint_file="blueprint.yaml")
governed_agent = GovernedConversableAgent(agent, steward)
# Use normally - governance automatic!
governed_agent.send("Hello", recipient)
Features¶
Message Governance¶
Evaluates messages before sending:
Function Call Governance¶
Controls tool/function execution:
from acgp_autogen import GovernedFunctionCall
def my_function(arg1, arg2):
return f"Result: {arg1} + {arg2}"
governed_func = GovernedFunctionCall(my_function, steward)
result = governed_func("a", "b") # Governed execution
Group Chat Governance¶
Govern entire group conversations:
from autogen import GroupChat
from acgp_autogen import GovernedGroupChat
group_chat = GroupChat(agents=[agent1, agent2, agent3])
governed_chat = GovernedGroupChat(group_chat, steward)
Async Support¶
from acgp_autogen import AsyncGovernedConversableAgent
async_agent = AsyncGovernedConversableAgent(agent, steward)
Advanced Patterns¶
Multi-Agent Conversations¶
engineer = GovernedConversableAgent(engineer_agent, steward)
reviewer = GovernedConversableAgent(reviewer_agent, steward)
planner = GovernedConversableAgent(planner_agent, steward)
# Governance applies to all agents in conversation
engineer.send("Let's build a feature", planner)
Reply Handlers¶
ACGP integrates via Autogen's reply handler system:
# Automatically registered when wrapping agent
governed_agent = GovernedConversableAgent(agent, steward)
# ACGP handler runs first (position=0)
Advanced Features¶
Get Governance Metrics¶
metrics = governed_agent.get_metrics()
print(f"Agent: {metrics['agent_name']}")
print(f"Total interventions: {metrics['total_interventions']}")
Exception Handling¶
from acgp_autogen import InterventionError, AutogenGovernanceError
try:
governed_agent.send("Risky message", recipient)
except InterventionError as e:
print(f"Message blocked: {e.message}")
print(f"Trace ID: {e.trace_id}")
Control Intervention Behavior¶
# Don't raise exceptions on interventions
governed_agent = GovernedConversableAgent(
agent,
steward,
raise_on_intervention=False # Log instead of raising
)
Best Practices¶
Agent Roles
Define governance rules based on agent roles (engineer, reviewer, planner).
Function Calls
Always wrap critical functions with GovernedFunctionCall for safety.
Reply Handler Position
ACGP registers as the first reply handler (position=0) to ensure all messages are evaluated.