Skip to main content

AutoGen Integration

Integrate Watchlight authorization with Microsoft AutoGen multi-agent systems.

Installation​

pip install wl-apdp[autogen]

Quick Start​

from autogen import AssistantAgent, UserProxyAgent
from wl_apdp.autogen import AuthorizedAssistant, WatchlightGroupChat

# Create authorized assistant
assistant = AuthorizedAssistant(
    name="research_assistant",
    principal_id="Agent::\"research-assistant\"",
    delegated_by="User::\"researcher\"",
    allowed_intents=["search_*", "analyze_*"],
    allowed_goals=["research"],
    llm_config={"model": "gpt-4"}
)

# User proxy
user_proxy = UserProxyAgent(
    name="user",
    human_input_mode="NEVER"
)

# Chat with authorization
result = user_proxy.initiate_chat(
    assistant,
    message="Search for recent AI papers",
    watchlight_context={
        "intent": "search_papers",
        "goal": "research"
    }
)

AuthorizedAssistant​

Extends AutoGen's AssistantAgent with authorization:

from wl_apdp.autogen import AuthorizedAssistant

assistant = AuthorizedAssistant(
    # Standard AutoGen parameters
    name="data_analyst",
    system_message="You are a data analysis expert.",
    llm_config={"model": "gpt-4"},

    # Watchlight parameters
    principal_id="Agent::\"data-analyst\"",
    delegated_by="User::\"analytics-lead\"",
    delegated_scope=["read", "execute"],

    # Allowed operations
    allowed_intents=["query_*", "analyze_*", "visualize_*"],
    allowed_goals=["data_analysis", "reporting"],

    # Trust level (1-10)
    trust_level=5
)

Authorized Function Calls​

Register Authorized Functions​

from wl_apdp.autogen import authorized_function

@authorized_function(
    resource_id="Tool::\"database-query\"",
    required_scope=["execute"],
    risk_level=2
)
def query_database(sql: str) -> str:
    """Execute a database query."""
    return execute_query(sql)

@authorized_function(
    resource_id="Tool::\"web-search\"",
    required_scope=["execute"],
    risk_level=1
)
def search_web(query: str) -> str:
    """Search the web."""
    return web_search(query)

# Register with assistant
assistant.register_function(
    function_map={
        "query_database": query_database,
        "search_web": search_web
    }
)

Function Authorization Decorator​

from wl_apdp.autogen import require_authorization

@require_authorization(
    action="execute",
    resource="Tool::\"file-write\"",
    intent_pattern="write_*"
)
def write_file(path: str, content: str):
    """Write content to a file."""
    with open(path, 'w') as f:
        f.write(content)

WatchlightGroupChat​

Multi-agent group chat with authorization:

from autogen import GroupChat, GroupChatManager
from wl_apdp.autogen import WatchlightGroupChat, AuthorizedAssistant

# Create authorized agents
researcher = AuthorizedAssistant(
    name="researcher",
    principal_id="Agent::\"researcher\"",
    allowed_goals=["research"]
)

analyst = AuthorizedAssistant(
    name="analyst",
    principal_id="Agent::\"analyst\"",
    allowed_goals=["analysis"]
)

writer = AuthorizedAssistant(
    name="writer",
    principal_id="Agent::\"writer\"",
    allowed_goals=["documentation"]
)

# Create authorized group chat
group_chat = WatchlightGroupChat(
    agents=[researcher, analyst, writer],
    watchlight_url="http://localhost:8081",

    # Delegation from group manager
    manager_principal="User::\"team-lead\"",

    # Authorization for agent selection
    authorize_speaker_selection=True,

    # Authorization for messages
    authorize_messages=True
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"model": "gpt-4"}
)

# Start chat
user_proxy.initiate_chat(
    manager,
    message="Research and analyze AI trends, then write a report"
)

Agent-to-Agent Authorization​

from wl_apdp.autogen import AuthorizedAgentChat

# Authorized chat between agents
chat = AuthorizedAgentChat(
    watchlight_url="http://localhost:8081"
)

# Agent A can only message Agent B if authorized
result = chat.send_message(
    from_agent=researcher,
    to_agent=analyst,
    message="Please analyze this data",
    context={
        "intent": "request_analysis",
        "goal": "collaborative_research"
    }
)

Delegation in AutoGen​

Hierarchical Delegation​

from wl_apdp.autogen import DelegatingAgent

# Manager agent that can delegate
manager = DelegatingAgent(
    name="project_manager",
    principal_id="Agent::\"manager\"",
    delegated_by="User::\"executive\"",
    can_delegate_to=[
        "Agent::\"researcher\"",
        "Agent::\"analyst\"",
        "Agent::\"writer\""
    ],
    max_delegation_depth=2
)

# When manager assigns tasks, delegation is automatic
manager.assign_task(
    to_agent=researcher,
    task="Research AI trends",
    delegated_scope=["read", "execute"]
)

Tracking Delegation Chains​

from wl_apdp.autogen import get_delegation_chain

# Get current delegation chain for an agent
chain = get_delegation_chain(analyst)
print(f"Delegation chain: {chain}")
# Output: [User::\"exec\" -> Agent::\"manager\" -> Agent::\"analyst\"]

Conversation Authorization​

Authorize Each Message​

from wl_apdp.autogen import AuthorizedConversation

conv = AuthorizedConversation(
    watchlight_url="http://localhost:8081",
    # Authorize every message
    authorize_all_messages=True,
    # Include conversation context
    include_history_context=True
)

# Wrap the chat
with conv.wrap(user_proxy, assistant):
    user_proxy.initiate_chat(
        assistant,
        message="Analyze sensitive data"
    )

Conversation-Level Authorization​

from wl_apdp.autogen import ConversationAuthorizer

authorizer = ConversationAuthorizer(
    watchlight_url="http://localhost:8081"
)

# Pre-authorize entire conversation
auth_result = authorizer.authorize_conversation(
    initiator="User::\"analyst\"",
    participants=["Agent::\"assistant\""],
    intent="data_analysis",
    goal="quarterly_report",
    estimated_actions=["read", "analyze", "summarize"]
)

if auth_result.allowed:
    # Proceed with conversation
    user_proxy.initiate_chat(assistant, message="...")

Tool Use Authorization​

from wl_apdp.autogen import AuthorizedToolUse

# Configure tool authorization
tool_auth = AuthorizedToolUse(
    watchlight_url="http://localhost:8081",
    tool_resources={
        "python": "Tool::\"code-execution\"",
        "bash": "Tool::\"shell-execution\"",
        "browser": "Tool::\"web-browser\""
    },
    risk_levels={
        "python": 2,
        "bash": 3,
        "browser": 1
    }
)

# Apply to assistant
assistant = AuthorizedAssistant(
    name="coder",
    principal_id="Agent::\"coder\"",
    tool_authorization=tool_auth,
    code_execution_config={
        "work_dir": "./workspace",
        "use_docker": True
    }
)

Full Example​

from autogen import UserProxyAgent, config_list_from_json
from wl_apdp.autogen import (
    AuthorizedAssistant,
    WatchlightGroupChat,
    authorized_function
)

# Load LLM config
config_list = config_list_from_json("OAI_CONFIG_LIST")
llm_config = {"config_list": config_list}

# Define authorized functions
@authorized_function(
    resource_id="Tool::\"web-search\"",
    risk_level=1
)
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

@authorized_function(
    resource_id="Tool::\"data-analysis\"",
    risk_level=2
)
def analyze_data(data: str) -> str:
    """Analyze provided data."""
    return f"Analysis of: {data}"

# Create authorized agents
researcher = AuthorizedAssistant(
    name="researcher",
    system_message="You research topics using web search.",
    principal_id="Agent::\"researcher\"",
    delegated_by="User::\"team-lead\"",
    allowed_intents=["search_*", "find_*"],
    allowed_goals=["research"],
    llm_config=llm_config
)
researcher.register_function(function_map={"search_web": search_web})

analyst = AuthorizedAssistant(
    name="analyst",
    system_message="You analyze data and provide insights.",
    principal_id="Agent::\"analyst\"",
    delegated_by="User::\"team-lead\"",
    allowed_intents=["analyze_*", "summarize_*"],
    allowed_goals=["analysis"],
    llm_config=llm_config
)
analyst.register_function(function_map={"analyze_data": analyze_data})

# User proxy
user = UserProxyAgent(
    name="user",
    human_input_mode="NEVER",
    code_execution_config=False
)

# Create authorized group chat
group_chat = WatchlightGroupChat(
    agents=[user, researcher, analyst],
    messages=[],
    max_round=10,
    watchlight_url="http://localhost:8081",
    manager_principal="User::\"team-lead\"",
    authorize_speaker_selection=True
)

manager = GroupChatManager(
    groupchat=group_chat,
    llm_config=llm_config
)

# Run
user.initiate_chat(
    manager,
    message="Research AI agent frameworks and analyze the findings",
    watchlight_context={
        "intent": "research_and_analyze",
        "goal": "research"
    }
)

Configuration​

# autogen_config.yaml
watchlight:
  url: http://localhost:8081
  api_key: ${WATCHLIGHT_API_KEY}

agents:
  researcher:
    principal_id: "Agent::\"researcher\""
    allowed_intents: ["search_*"]
    trust_level: 3

  analyst:
    principal_id: "Agent::\"analyst\""
    allowed_intents: ["analyze_*"]
    trust_level: 4

group_chat:
  authorize_speaker_selection: true
  authorize_messages: true
  max_delegation_depth: 2

Next Steps​