AutoGen Quickstart

pip install visiq "autogen-agentchat>=0.7,<0.8" "autogen-core>=0.7,<0.8"

import functools
import inspect

from visiq import Governor, ToolBlocked  # pip install visiq

gov = Governor(agent_id="support-bot")


def scenario_search(query: str) -> list:
    # A stand-in retrieval source — your real RAG store returns the same shape:
    # a list of {"page_content", "metadata"} dicts gate_documents can filter.
    return [
        {"page_content": f"Q3 revenue was $4.2M. (matched: {query})",
         "metadata": {"classification": "internal"}},
    ]


def governed(name, fn):
    # Route one tool body through the gate. gate_tool decides BEFORE the body
    # runs: a deny raises ToolBlocked; a mask verdict hands the callback ONLY the
    # redacted arguments, never the originals.
    sig = inspect.signature(fn)

    @functools.wraps(fn)
    def wrapper(*a, **kw):
        bound = sig.bind(*a, **kw)
        bound.apply_defaults()
        args = dict(bound.arguments)
        try:
            return gov.gate_tool(name, args, lambda effective: fn(**effective))
        except ToolBlocked as e:
            return f"[BLOCKED BY POLICY] {e.reason}"

    wrapper.__name__ = name
    return wrapper


def issue_refund(customer_id: str, amount: int) -> str:
    return f"Refunded ${amount} to {customer_id}"


def search_knowledge(query: str) -> str:
    # Retrieval governance: drop / redact documents before the model sees them.
    docs = gov.gate_documents(scenario_search(query), query=query)
    return "\n\n".join(d["page_content"] for d in docs)

from autogen_core.tools import FunctionTool

refund_tool = FunctionTool(governed("issue_refund", issue_refund),
                           description="Issue a refund to a customer")
search_tool = FunctionTool(search_knowledge,
                           description="Search the company knowledge base")

gov.start(tools=[{"name": "issue_refund"}, {"name": "search_knowledge"}])
gov.flush()  # deliver buffered audit events before the process exits