Examples cover: - File operations with risk levels - API calls with cost tracking - Decision-making with reasoning - Multi-agent collaboration - Incident investigation MIT license: Build on this. Make AGI safe.
164 lines
4.8 KiB
Python
164 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Advanced examples of Agent Provenance Chain usage.
|
|
"""
|
|
|
|
from apc import create_agent_chain
|
|
import json
|
|
|
|
|
|
def example_file_operations():
|
|
"""Example: Signing file operations."""
|
|
print("Example 1: File Operations\n" + "=" * 50)
|
|
|
|
chain = create_agent_chain("file-agent")
|
|
|
|
# Sign a file read
|
|
chain.sign_action(
|
|
"file_read",
|
|
{"path": "/etc/passwd", "bytes_read": 2048},
|
|
{"reasoning": "Checking user accounts for security audit"}
|
|
)
|
|
|
|
# Sign a file write
|
|
chain.sign_action(
|
|
"file_write",
|
|
{"path": "/var/log/agent.log", "bytes_written": 156},
|
|
{"reasoning": "Logging operation results", "risk_level": "low"}
|
|
)
|
|
|
|
print(f"✅ Signed {len(chain.get_chain())} file operations\n")
|
|
|
|
|
|
def example_api_calls():
|
|
"""Example: Signing API interactions."""
|
|
print("Example 2: API Calls\n" + "=" * 50)
|
|
|
|
chain = create_agent_chain("api-agent")
|
|
|
|
# External API call
|
|
chain.sign_action(
|
|
"api_call",
|
|
{
|
|
"url": "https://api.openai.com/v1/chat/completions",
|
|
"method": "POST",
|
|
"status": 200,
|
|
"tokens_used": 1523
|
|
},
|
|
{
|
|
"reasoning": "Generating response to user query",
|
|
"data_sensitivity": "user_provided",
|
|
"cost_usd": 0.03
|
|
}
|
|
)
|
|
|
|
print(f"✅ Signed {len(chain.get_chain())} API calls\n")
|
|
|
|
|
|
def example_decision_making():
|
|
"""Example: Signing agent decisions."""
|
|
print("Example 3: Decision Making\n" + "=" * 50)
|
|
|
|
chain = create_agent_chain("decision-agent")
|
|
|
|
# Complex decision with reasoning
|
|
chain.sign_action(
|
|
"decision",
|
|
{
|
|
"question": "Should I delete user data?",
|
|
"answer": "no",
|
|
"confidence": 0.95,
|
|
"alternatives_considered": ["yes", "ask_user", "archive"]
|
|
},
|
|
{
|
|
"reasoning": "Deletion requires explicit user consent per privacy policy",
|
|
"policy_ref": "PRIVACY-001",
|
|
"risk_level": "high"
|
|
}
|
|
)
|
|
|
|
print(f"✅ Signed {len(chain.get_chain())} decisions\n")
|
|
|
|
|
|
def example_multi_agent():
|
|
"""Example: Multi-agent scenario."""
|
|
print("Example 4: Multi-Agent Collaboration\n" + "=" * 50)
|
|
|
|
agent_a = create_agent_chain("agent-a")
|
|
agent_b = create_agent_chain("agent-b")
|
|
|
|
# Agent A requests help
|
|
agent_a.sign_action(
|
|
"request_collaboration",
|
|
{"target_agent": "agent-b", "task": "data_analysis"},
|
|
{"reasoning": "Task too complex for solo operation"}
|
|
)
|
|
|
|
# Agent B accepts
|
|
agent_b.sign_action(
|
|
"accept_collaboration",
|
|
{"requesting_agent": "agent-a", "task": "data_analysis"},
|
|
{"reasoning": "Have spare capacity and required skills"}
|
|
)
|
|
|
|
# Agent B performs work
|
|
agent_b.sign_action(
|
|
"complete_task",
|
|
{"task": "data_analysis", "result": "summary.json"},
|
|
{"reasoning": "Analysis complete, returning results"}
|
|
)
|
|
|
|
# Agent A verifies
|
|
agent_a.sign_action(
|
|
"verify_result",
|
|
{"from_agent": "agent-b", "verified": True},
|
|
{"reasoning": "Results match expected format and quality"}
|
|
)
|
|
|
|
print(f"✅ Agent A: {len(agent_a.get_chain())} actions")
|
|
print(f"✅ Agent B: {len(agent_b.get_chain())} actions")
|
|
print("✅ Both chains independently verifiable\n")
|
|
|
|
|
|
def example_incident_investigation():
|
|
"""Example: Investigating an incident."""
|
|
print("Example 5: Incident Investigation\n" + "=" * 50)
|
|
|
|
chain = create_agent_chain("prod-agent")
|
|
|
|
# Simulate an incident
|
|
chain.sign_action(
|
|
"exec",
|
|
{"command": "rm /tmp/important.db", "exit_code": 0},
|
|
{"reasoning": "Cleanup temporary files", "risk_level": "medium"}
|
|
)
|
|
|
|
# Later: investigate
|
|
print("Investigating: Why was important.db deleted?")
|
|
print()
|
|
|
|
full_chain = chain.get_chain()
|
|
for action in full_chain:
|
|
if "important.db" in json.dumps(action):
|
|
print(f"Found in action at {action['iso_time']}:")
|
|
print(f" Reasoning: {action['context'].get('reasoning')}")
|
|
print(f" Signed by: {action['agent']}")
|
|
print(f" Signature: {action['signature'][:32]}...")
|
|
print()
|
|
print("✅ Cryptographic proof of what happened and why\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
example_file_operations()
|
|
example_api_calls()
|
|
example_decision_making()
|
|
example_multi_agent()
|
|
example_incident_investigation()
|
|
|
|
print("=" * 50)
|
|
print("All examples demonstrate:")
|
|
print(" • Cryptographic signing of actions")
|
|
print(" • Context preservation (reasoning)")
|
|
print(" • Independent verification")
|
|
print(" • Audit trail for accountability")
|
|
print("=" * 50)
|