profit/agent-governance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
agent-governance/agents/llm-planner/agent.py
profit 77655c298c Initial commit: Agent Governance System Phase 8 
Phase 8 Production Hardening with complete governance infrastructure:

- Vault integration with tiered policies (T0-T4)
- DragonflyDB state management
- SQLite audit ledger
- Pipeline DSL and templates
- Promotion/revocation engine
- Checkpoint system for session persistence
- Health manager and circuit breaker for fault tolerance
- GitHub/Slack integrations
- Architectural test pipeline with bug watcher, suggestion engine, council review
- Multi-agent chaos testing framework

Test Results:
- Governance tests: 68/68 passing
- E2E workflow: 16/16 passing
- Phase 2 Vault: 14/14 passing
- Integration tests: 27/27 passing

Coverage: 57.6% average across 12 phases

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-23 22:07:06 -05:00

528 lines
18 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""
LLM Planner Agent - Tier 0 Observer
====================================
A compliant agent that uses OpenRouter LLMs for planning tasks.
Follows the Agent Foundation document constraints.
Capabilities:
- Read documentation and inventory from Vault
- Generate implementation plans using LLM
- Produce structured, auditable outputs
- Log all actions to governance ledger
"""
import json
import sqlite3
import subprocess
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional
from openai import OpenAI
from pydantic import BaseModel, Field
# =============================================================================
# Agent Metadata (Section 4 of Foundation Doc)
# =============================================================================
AGENT_METADATA = {
"agent_id": "llm-planner-001",
"agent_role": "observer",
"owner": "system",
"version": "0.1.0",
"tier": 0,
"input_contract": "TaskRequest",
"output_contract": "AgentOutput",
"allowed_side_effects": [
"read_docs",
"read_inventory",
"read_logs",
"generate_plan",
"llm_inference"
],
"forbidden_actions": [
"ssh",
"create_vm",
"modify_vm",
"delete_vm",
"run_ansible",
"run_terraform",
"write_secrets",
"execute_shell",
"modify_files"
],
"confidence_reporting": True,
"confidence_threshold": 0.7,
}
# =============================================================================
# Structured Types (Section 6 of Foundation Doc)
# =============================================================================
class Decision(str, Enum):
EXECUTE = "EXECUTE"
SKIP = "SKIP"
ESCALATE = "ESCALATE"
INSUFFICIENT_INFORMATION = "INSUFFICIENT_INFORMATION"
ERROR = "ERROR"
class SideEffect(BaseModel):
type: str
target: str
reversible: bool = True
class ErrorInfo(BaseModel):
type: str
message: str
triggering_input: str
partial_progress: str
recommended_action: str
class AgentOutput(BaseModel):
"""Required output format per Foundation Doc Section 6"""
agent_id: str
version: str
timestamp: str
action: str
decision: Decision
confidence: float = Field(ge=0.0, le=1.0)
assumptions: list[str] = []
dependencies: list[str] = []
side_effects: list[SideEffect] = []
notes_for_humans: str = ""
error: Optional[ErrorInfo] = None
# Extended fields for LLM agent
llm_model: Optional[str] = None
llm_response: Optional[str] = None
plan: Optional[dict] = None
class TaskRequest(BaseModel):
"""Input schema for task requests"""
task_type: str # e.g., "generate_plan", "analyze", "summarize"
description: str
context: dict = {}
constraints: list[str] = []
# =============================================================================
# Vault Integration
# =============================================================================
class VaultClient:
def __init__(self, addr: str = "https://127.0.0.1:8200"):
self.addr = addr
self.token = self._load_token()
def _load_token(self) -> str:
with open("/opt/vault/init-keys.json") as f:
return json.load(f)["root_token"]
def read_secret(self, path: str) -> dict:
result = subprocess.run([
"curl", "-sk",
"-H", f"X-Vault-Token: {self.token}",
f"{self.addr}/v1/secret/data/{path}"
], capture_output=True, text=True)
data = json.loads(result.stdout)
if "data" in data and "data" in data["data"]:
return data["data"]["data"]
raise ValueError(f"Failed to read secret: {path}")
def get_openrouter_key(self) -> str:
return self.read_secret("api-keys/openrouter")["api_key"]
def get_inventory(self) -> dict:
return self.read_secret("inventory/proxmox")
def get_docs(self, doc_name: str) -> dict:
return self.read_secret(f"docs/{doc_name}")
# =============================================================================
# Ledger Integration
# =============================================================================
class Ledger:
def __init__(self, db_path: str = "/opt/agent-governance/ledger/governance.db"):
self.db_path = db_path
def log_action(self, output: AgentOutput, success: bool):
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO agent_actions
(timestamp, agent_id, agent_version, tier, action, decision, confidence,
success, error_type, error_message)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
output.timestamp,
output.agent_id,
output.version,
AGENT_METADATA["tier"],
output.action,
output.decision.value,
output.confidence,
1 if success else 0,
output.error.type if output.error else None,
output.error.message if output.error else None
))
# Update metrics
cursor.execute("""
INSERT INTO agent_metrics (agent_id, current_tier, total_runs, last_active_at, compliant_runs, consecutive_compliant)
VALUES (?, ?, 1, ?, ?, ?)
ON CONFLICT(agent_id) DO UPDATE SET
total_runs = total_runs + 1,
compliant_runs = CASE WHEN ? = 1 THEN compliant_runs + 1 ELSE compliant_runs END,
consecutive_compliant = CASE WHEN ? = 1 THEN consecutive_compliant + 1 ELSE 0 END,
last_active_at = ?,
updated_at = ?
""", (
output.agent_id, AGENT_METADATA["tier"], output.timestamp,
1 if success else 0, 1 if success else 0,
1 if success else 0, 1 if success else 0,
output.timestamp, output.timestamp
))
conn.commit()
conn.close()
# =============================================================================
# LLM Planner Agent
# =============================================================================
class LLMPlannerAgent:
"""
Tier 0 Observer Agent with LLM capabilities.
Invariant: If the agent cannot explain why it took an action,
that action is invalid. (Foundation Doc Section 1)
"""
def __init__(self, model: str = "anthropic/claude-3.5-haiku"):
self.metadata = AGENT_METADATA
self.model = model
self.vault = VaultClient()
self.ledger = Ledger()
# Initialize OpenRouter client
api_key = self.vault.get_openrouter_key()
self.llm = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=api_key
)
print(f"[INIT] Agent {self.metadata['agent_id']} v{self.metadata['version']} initialized")
print(f"[INIT] LLM Model: {self.model}")
print(f"[INIT] Tier: {self.metadata['tier']} ({self.metadata['agent_role']})")
def _now(self) -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _validate_action(self, action: str) -> bool:
"""Section 3.3: Bounded Authority - validate action is allowed"""
if action in self.metadata["forbidden_actions"]:
return False
if action in self.metadata["allowed_side_effects"]:
return True
# Default deny for unknown actions
return False
def _create_error_output(self, action: str, error: ErrorInfo) -> AgentOutput:
return AgentOutput(
agent_id=self.metadata["agent_id"],
version=self.metadata["version"],
timestamp=self._now(),
action=action,
decision=Decision.ERROR,
confidence=0.0,
error=error,
notes_for_humans="Action failed - see error details"
)
def _check_confidence(self, confidence: float) -> bool:
"""Section 7: Confidence below threshold = no irreversible actions"""
return confidence >= self.metadata["confidence_threshold"]
def generate_plan(self, request: TaskRequest) -> AgentOutput:
"""Generate an implementation plan using LLM"""
action = "generate_plan"
# Validate action is allowed
if not self._validate_action(action):
error = ErrorInfo(
type="FORBIDDEN_ACTION",
message=f"Action '{action}' is not permitted for this agent",
triggering_input=request.description,
partial_progress="None",
recommended_action="Escalate to higher tier agent"
)
output = self._create_error_output(action, error)
self.ledger.log_action(output, success=False)
return output
# Build context from Vault
try:
inventory = self.vault.get_inventory()
context_info = f"Infrastructure: {inventory.get('cluster', 'unknown')} with pools: {inventory.get('pools', 'unknown')}"
except Exception as e:
context_info = "Infrastructure context unavailable"
# Construct prompt
system_prompt = f"""You are a Tier 0 Observer agent in an infrastructure automation system.
Your role is to generate implementation plans - you CANNOT execute anything.
Agent Constraints:
- You can ONLY generate plans, not execute them
- Plans must be reversible and include rollback steps
- You must identify uncertainties and flag them
- All assumptions must be explicitly stated
Infrastructure Context:
{context_info}
Output your plan as JSON with this structure:
{{
"title": "Plan title",
"summary": "Brief summary",
"confidence": 0.0-1.0,
"assumptions": ["assumption 1", "assumption 2"],
"uncertainties": ["uncertainty 1"],
"steps": [
{{"step": 1, "action": "description", "reversible": true, "rollback": "how to undo"}}
],
"estimated_tier_required": 0-4,
"risks": ["risk 1"]
}}"""
user_prompt = f"""Task: {request.description}
Additional Context: {json.dumps(request.context) if request.context else 'None'}
Constraints: {', '.join(request.constraints) if request.constraints else 'None'}
Generate a detailed implementation plan."""
# Call LLM
try:
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
max_tokens=2000,
temperature=0.3 # Lower temperature for more deterministic planning
)
llm_response = response.choices[0].message.content
# Try to parse the plan JSON
try:
# Extract JSON from response
json_start = llm_response.find('{')
json_end = llm_response.rfind('}') + 1
if json_start >= 0 and json_end > json_start:
plan = json.loads(llm_response[json_start:json_end])
confidence = plan.get("confidence", 0.5)
else:
plan = {"raw_response": llm_response}
confidence = 0.5
except json.JSONDecodeError:
plan = {"raw_response": llm_response}
confidence = 0.5
# Check confidence threshold
decision = Decision.EXECUTE if self._check_confidence(confidence) else Decision.ESCALATE
output = AgentOutput(
agent_id=self.metadata["agent_id"],
version=self.metadata["version"],
timestamp=self._now(),
action=action,
decision=decision,
confidence=confidence,
assumptions=plan.get("assumptions", []),
dependencies=[],
side_effects=[
SideEffect(type="llm_inference", target=self.model, reversible=True)
],
notes_for_humans=f"Plan generated. Estimated tier required: {plan.get('estimated_tier_required', 'unknown')}",
llm_model=self.model,
llm_response=llm_response[:500] + "..." if len(llm_response) > 500 else llm_response,
plan=plan
)
self.ledger.log_action(output, success=True)
return output
except Exception as e:
error = ErrorInfo(
type="LLM_ERROR",
message=str(e),
triggering_input=request.description,
partial_progress="Context loaded, LLM call failed",
recommended_action="Check API key and connectivity"
)
output = self._create_error_output(action, error)
self.ledger.log_action(output, success=False)
return output
def analyze(self, request: TaskRequest) -> AgentOutput:
"""Analyze a topic and provide insights"""
action = "read_docs" # Analysis is a form of reading
if not self._validate_action(action):
error = ErrorInfo(
type="FORBIDDEN_ACTION",
message=f"Action '{action}' is not permitted",
triggering_input=request.description,
partial_progress="None",
recommended_action="Check agent permissions"
)
output = self._create_error_output(action, error)
self.ledger.log_action(output, success=False)
return output
try:
response = self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a technical analyst. Provide clear, structured analysis. Be explicit about uncertainties."},
{"role": "user", "content": request.description}
],
max_tokens=1500,
temperature=0.4
)
llm_response = response.choices[0].message.content
output = AgentOutput(
agent_id=self.metadata["agent_id"],
version=self.metadata["version"],
timestamp=self._now(),
action=action,
decision=Decision.EXECUTE,
confidence=0.8,
assumptions=["Analysis based on provided context only"],
dependencies=[],
side_effects=[
SideEffect(type="llm_inference", target=self.model, reversible=True)
],
notes_for_humans="Analysis complete",
llm_model=self.model,
llm_response=llm_response
)
self.ledger.log_action(output, success=True)
return output
except Exception as e:
error = ErrorInfo(
type="LLM_ERROR",
message=str(e),
triggering_input=request.description,
partial_progress="None",
recommended_action="Retry or check connectivity"
)
output = self._create_error_output(action, error)
self.ledger.log_action(output, success=False)
return output
def run(self, request: TaskRequest) -> AgentOutput:
"""Main entry point - routes to appropriate handler"""
print(f"\n[TASK] Received: {request.task_type}")
print(f"[TASK] Description: {request.description[:100]}...")
handlers = {
"generate_plan": self.generate_plan,
"plan": self.generate_plan,
"analyze": self.analyze,
"analysis": self.analyze,
}
handler = handlers.get(request.task_type)
if handler:
return handler(request)
# Unknown task type
error = ErrorInfo(
type="UNKNOWN_TASK_TYPE",
message=f"Unknown task type: {request.task_type}",
triggering_input=json.dumps(request.model_dump()),
partial_progress="None",
recommended_action=f"Use one of: {list(handlers.keys())}"
)
output = self._create_error_output(request.task_type, error)
self.ledger.log_action(output, success=False)
return output
# =============================================================================
# CLI Interface
# =============================================================================
def main():
import argparse
parser = argparse.ArgumentParser(description="LLM Planner Agent")
parser.add_argument("task_type", choices=["plan", "analyze"], help="Type of task")
parser.add_argument("description", help="Task description")
parser.add_argument("--model", default="anthropic/claude-3.5-haiku", help="OpenRouter model")
parser.add_argument("--json", action="store_true", help="Output raw JSON")
args = parser.parse_args()
agent = LLMPlannerAgent(model=args.model)
request = TaskRequest(
task_type=args.task_type,
description=args.description
)
output = agent.run(request)
if args.json:
print(output.model_dump_json(indent=2))
else:
print("\n" + "="*60)
print(f"Decision: {output.decision.value}")
print(f"Confidence: {output.confidence}")
print(f"Model: {output.llm_model}")
print("="*60)
if output.plan:
print("\n📋 PLAN:")
print(json.dumps(output.plan, indent=2))
elif output.llm_response:
print("\n📝 RESPONSE:")
print(output.llm_response)
if output.error:
print("\n❌ ERROR:")
print(f" Type: {output.error.type}")
print(f" Message: {output.error.message}")
print(f" Recommendation: {output.error.recommended_action}")
if output.assumptions:
print("\n⚠️ ASSUMPTIONS:")
for a in output.assumptions:
print(f" - {a}")
print("\n" + "="*60)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink