Specialists that think, collaborate, and act — with policy.

# pip install pegesis
from pegesis import Agent, tool, Memory, Policy

# 1) Declare a tool (auditable, typed I/O)
@tool(name="work_orders.create", version="1.0.0")
def create_work_order(site: str, zone: int, note: str) -> dict:
    # Implementation lives in your infra; SDK records the call and return.
    return {"status": "queued", "site": site, "zone": zone, "note": note}

# 2) Load policy & attach memory (short- and long-term)
policy = Policy.from_file("policies/facilities.json")
mem    = Memory.window(capacity=1024).with_vector_index("events", dims=384)

# 3) Register an agent with chain-of-thought style and compliance guardrails
agent = Agent(
    name="vision-guard",
    role="Detect hazards and escalate within policy",
    policy=policy,
    memory=mem,
    cot="structured"   # structured chain-of-thought trace stored in audit log
)

# 4) Subscribe to events and act under policy
@agent.on("thermal.alert")
def on_thermal(ctx):
    # Retrieve similar incidents for rationale (memory read)
    prior = agent.memory.search("events", query=ctx.signature, k=3)

    # Policy gate: only create work orders when severity meets contract
    if ctx.severity >= policy["SLA.minSeverity"]:
        # Every tool call is recorded with input/output for audit
        return create_work_order(site=ctx.site, zone=ctx.zone, note="Thermal anomaly")

    # Otherwise, store observation and exit
    agent.memory.append("events", {"site": ctx.site, "kind": "thermal", "ok": True})
    return {"status": "observed"}

# 5) Hand-off with full reasoning trace and policy context
@agent.on("vision.object.detected")
def on_object(ctx):
    if ctx.label in policy["restrictedObjects"]:
        return agent.handoff("compliance-auditor", ctx, reason="Restricted object detected")

# 6) Run once or as a service (streamed decisions with signed traces)
if __name__ == "__main__":
    agent.serve(":8081")