From 422f0859f437f0c361f8661fb989eaf76c064cff Mon Sep 17 00:00:00 2001 From: Nathan Schneider Date: Sun, 8 Feb 2026 18:58:03 -0700 Subject: [PATCH] Remove legacy agent file (preserved in git history) Git provides version history - no need to keep old files in the codebase. Co-Authored-By: Claude Sonnet 4.5 --- src/govbot/agent_legacy.py | 659 ------------------------------------- 1 file changed, 659 deletions(-) delete mode 100644 src/govbot/agent_legacy.py diff --git a/src/govbot/agent_legacy.py b/src/govbot/agent_legacy.py deleted file mode 100644 index 1ec1d75..0000000 --- a/src/govbot/agent_legacy.py +++ /dev/null @@ -1,659 +0,0 @@ -""" -AI Agent Orchestration for Governance Bot. - -This is the core agentic system that: -1. Receives governance requests -2. Consults the constitution (via RAG) -3. Plans appropriate actions -4. Executes using primitives -5. Maintains audit trail -""" - -import json -import subprocess -from typing import Dict, Any, Optional, List -from datetime import datetime, timedelta -from sqlalchemy.orm import Session - -from .governance.constitution import ConstitutionalReasoner -from .governance.primitives import GovernancePrimitives -from .db import queries - - -class GovernanceAgent: - """ - The AI agent that interprets requests and orchestrates governance actions. - """ - - def __init__( - self, - db_session: Session, - constitution_path: str, - model: Optional[str] = None, - api_keys: Optional[Dict[str, str]] = None, - ): - """ - Initialize the governance agent. - - Args: - db_session: Database session - constitution_path: Path to constitution file - model: LLM model to use (None for default) - api_keys: Dict with 'openai' and/or 'anthropic' API keys - """ - self.db = db_session - self.constitution = ConstitutionalReasoner(constitution_path, model, api_keys) - self.primitives = GovernancePrimitives(db_session) - self.model = model - self.api_keys = api_keys or {} - - def process_request( - self, request: str, actor: str, context: Optional[Dict[str, Any]] = None - ) -> Dict[str, Any]: - """ - Process a governance request from a user. - - This is the main agentic loop: - 1. Parse intent - 2. Consult constitution - 3. Plan actions - 4. Execute with audit trail - 5. Return response - - Args: - request: Natural language request from user - actor: Who made the request (Mastodon handle) - context: Optional context (thread ID, etc.) - - Returns: - Dict with 'response', 'actions_taken', 'process_id', etc. - """ - # Step 1: Parse intent - intent = self._parse_intent(request, actor) - - if intent.get("error"): - return {"response": intent["error"], "success": False} - - # Step 2: Consult constitution - constitutional_guidance = self.constitution.query( - question=intent["intent_description"], - context=f"Actor: {actor}, Request: {request}", - ) - - # Step 3: Check for ambiguity - if constitutional_guidance.get("confidence") == "low": - return self._handle_ambiguity( - request, actor, constitutional_guidance - ) - - # Step 4: Plan actions - action_plan = self._plan_actions( - intent, constitutional_guidance, actor, context - ) - - # Step 5: Execute plan - result = self._execute_plan(action_plan, actor) - - return result - - def _parse_intent(self, request: str, actor: str) -> Dict[str, Any]: - """ - Use AI to parse user intent from natural language. - - Args: - request: User's request - actor: Who made the request - - Returns: - Dict with 'intent_type', 'intent_description', 'parameters' - """ - prompt = f"""Parse this governance request and extract structured information. - -REQUEST: "{request}" -ACTOR: {actor} - -Identify: -1. Intent type (e.g., "create_proposal", "cast_vote", "query_constitution", "appeal", etc.) -2. Clear description of what the user wants -3. Key parameters extracted from request - -Respond with JSON: -{{ - "intent_type": "the type of intent", - "intent_description": "clear description of what user wants", - "parameters": {{ - "key": "value" - }} -}} -""" - - try: - result = self._call_llm(prompt) - parsed = self._extract_json(result) - return parsed - except Exception as e: - return {"error": f"Could not parse request: {str(e)}"} - - def _plan_actions( - self, - intent: Dict[str, Any], - constitutional_guidance: Dict[str, Any], - actor: str, - context: Optional[Dict[str, Any]], - ) -> Dict[str, Any]: - """ - Plan the sequence of primitive actions to fulfill the intent. - - Args: - intent: Parsed intent - constitutional_guidance: Constitutional interpretation - actor: Who initiated - context: Additional context - - Returns: - Action plan dictionary - """ - intent_type = intent.get("intent_type") - - # Route to specific planning function based on intent - if intent_type == "create_proposal": - return self._plan_proposal_creation( - intent, constitutional_guidance, actor, context - ) - elif intent_type == "cast_vote": - return self._plan_vote_casting( - intent, constitutional_guidance, actor, context - ) - elif intent_type == "query_constitution": - return self._plan_constitutional_query( - intent, constitutional_guidance, actor - ) - elif intent_type == "appeal": - return self._plan_appeal( - intent, constitutional_guidance, actor, context - ) - else: - # Generic planning using AI - return self._plan_generic( - intent, constitutional_guidance, actor, context - ) - - def _plan_proposal_creation( - self, - intent: Dict[str, Any], - constitutional_guidance: Dict[str, Any], - actor: str, - context: Optional[Dict[str, Any]], - ) -> Dict[str, Any]: - """Plan actions for creating a proposal""" - params = intent.get("parameters", {}) - proposal_text = params.get("proposal_text", intent.get("intent_description")) - - # Interpret proposal to determine type and requirements - proposal_info = self.constitution.interpret_proposal(proposal_text) - - # Check if the actor has direct authority to execute this - # (e.g., @admin in benevolent dictator model) - decision_maker = proposal_info.get('decision_maker', '').lower() - if decision_maker == actor.lower() or (decision_maker == '@admin' and actor.lower() in ['@admin', 'admin']): - # This person has authority - execute directly, don't create a proposal - return self._plan_direct_execution(intent, proposal_text, constitutional_guidance, actor, context) - - # Build action plan - plan = { - "intent_type": "create_proposal", - "constitutional_basis": constitutional_guidance.get("citations", []), - "actions": [ - { - "primitive": "create_process", - "args": { - "process_type": f"{proposal_info['proposal_type']}_proposal", - "creator": actor, - "deadline_days": proposal_info.get("discussion_period_days", 6), - "constitutional_basis": str(constitutional_guidance.get("citations")), - "initial_state": { - "proposal_text": proposal_text, - "title": proposal_info.get("title", proposal_text[:100]), - "description": proposal_info.get("description", proposal_text), - "proposal_type": proposal_info["proposal_type"], - "voting_threshold": proposal_info.get("voting_threshold"), - "votes": {}, - }, - "mastodon_thread_id": context.get("thread_id") - if context - else None, - }, - }, - { - "primitive": "schedule_reminder", - "args": { - "when": "deadline", # Will be calculated from process deadline - "message": f"Proposal by {actor} has reached its deadline. Counting votes.", - }, - }, - ], - "response_template": self._build_proposal_response( - proposal_text, proposal_info, constitutional_guidance, actor - ), - } - - return plan - - def _plan_direct_execution( - self, - intent: Dict[str, Any], - request_text: str, - constitutional_guidance: Dict[str, Any], - actor: str, - context: Optional[Dict[str, Any]], - ) -> Dict[str, Any]: - """Plan direct execution when actor has authority""" - # For now, acknowledge @admin's authority - # Future: implement actual rule changes, user management, etc. - plan = { - "intent_type": "admin_directive", - "constitutional_basis": constitutional_guidance.get("citations", []), - "actions": [], # No actions needed - just acknowledge - "response_template": f"""Understood. As administrator, you have the authority to implement this. - -Directive: {request_text[:250]} - -Constitutional basis: {', '.join(constitutional_guidance.get('citations', []))} - -Note: The governance system acknowledges your decision. Implementation of automated rule enforcement is forthcoming. -""", - } - return plan - - def _build_proposal_response( - self, - proposal_text: str, - proposal_info: Dict[str, Any], - constitutional_guidance: Dict[str, Any], - actor: str - ) -> str: - """Build appropriate response based on proposal type""" - proposal_type = proposal_info.get('proposal_type', 'standard') - - # Check if this is an admin decision model - if proposal_type == 'admin_decision' or proposal_info.get('decision_maker') == '@admin': - return f"""Proposal submitted: {proposal_text[:200]} - -According to the constitution, @admin holds authority to make decisions on governance matters. - -Constitutional basis: {', '.join(constitutional_guidance.get('citations', []))} - -@admin will review this proposal and announce a decision. - -Process ID: {{process_id}} -""" - else: - # Democratic model with voting - return f"""Proposal created: {proposal_text[:100]}... - -Type: {proposal_type} -Discussion period: {proposal_info.get('discussion_period_days')} days -Voting threshold: {proposal_info.get('voting_threshold')} - -Constitutional basis: {', '.join(constitutional_guidance.get('citations', []))} - -Reply with 'agree', 'disagree', 'abstain', or 'block' to vote. -Process ID: {{process_id}} -""" - - def _plan_vote_casting( - self, - intent: Dict[str, Any], - constitutional_guidance: Dict[str, Any], - actor: str, - context: Optional[Dict[str, Any]], - ) -> Dict[str, Any]: - """Plan actions for casting a vote""" - params = intent.get("parameters", {}) - vote_type = params.get("vote_type", "agree").lower() - process_id = params.get("process_id") - - # If no process_id in params, try to find it from thread context - if not process_id and context: - # Get the status ID being replied to - reply_to_id = context.get("reply_to_id") - if reply_to_id: - # Query for active processes and check if any match this thread - active_processes = queries.get_active_processes(self.db) - for proc in active_processes: - if proc.state_data: - announcement_id = proc.state_data.get("announcement_thread_id") - if announcement_id and str(announcement_id) == str(reply_to_id): - process_id = proc.id - break - - # If still not found, try the most recent active proposal - if not process_id and active_processes: - process_id = active_processes[0].id - - if not process_id: - return { - "error": "Could not identify which proposal to vote on. Please reply to a proposal announcement or specify the process ID." - } - - plan = { - "intent_type": "cast_vote", - "constitutional_basis": constitutional_guidance.get("citations", []), - "actions": [ - { - "primitive": "update_process_state", - "args": { - "process_id": process_id, - "state_updates": { - f"votes.{actor}": { - "vote": vote_type, - "timestamp": datetime.utcnow().isoformat(), - } - }, - "actor": actor, - }, - } - ], - "response_template": f"""Vote recorded: {vote_type} - -Voter: {actor} -Process: {{process_id}} -""", - } - - return plan - - def _plan_constitutional_query( - self, - intent: Dict[str, Any], - constitutional_guidance: Dict[str, Any], - actor: str, - ) -> Dict[str, Any]: - """Plan response for constitutional query""" - return { - "intent_type": "query_constitution", - "actions": [], # No state changes needed - "response_template": f"""Constitutional Interpretation: - -{constitutional_guidance['answer']} - -Citations: {', '.join(constitutional_guidance.get('citations', []))} -Confidence: {constitutional_guidance.get('confidence', 'medium')} -""", - } - - def _plan_appeal( - self, - intent: Dict[str, Any], - constitutional_guidance: Dict[str, Any], - actor: str, - context: Optional[Dict[str, Any]], - ) -> Dict[str, Any]: - """Plan actions for an appeal""" - params = intent.get("parameters", {}) - action_id = params.get("action_id") - - plan = { - "intent_type": "appeal", - "constitutional_basis": constitutional_guidance.get("citations", []), - "actions": [ - { - "primitive": "create_process", - "args": { - "process_type": "appeal", - "creator": actor, - "deadline_days": 3, - "constitutional_basis": "Article 6: Appeals", - "initial_state": { - "appealed_action_id": action_id, - "appellant": actor, - "votes": {}, - }, - }, - } - ], - "response_template": f"""Appeal initiated by {actor} - -Appealing action: {{action_id}} -Discussion period: 3 days - -Community members can vote on whether to override the action. -""", - } - - return plan - - def _plan_generic( - self, - intent: Dict[str, Any], - constitutional_guidance: Dict[str, Any], - actor: str, - context: Optional[Dict[str, Any]], - ) -> Dict[str, Any]: - """Use AI to plan generic actions""" - # This is a fallback for intents we haven't explicitly coded - prompt = f"""Based on this intent and constitutional guidance, plan the primitive actions needed. - -INTENT: {json.dumps(intent, indent=2)} - -CONSTITUTIONAL GUIDANCE: {json.dumps(constitutional_guidance, indent=2)} - -Available primitives: -- create_process(process_type, creator, deadline_days, constitutional_basis, initial_state) -- update_process_state(process_id, state_updates, actor) -- store_record(record_type, data, actor, reasoning, citation) -- schedule_reminder(when, message) - -Plan the actions as JSON: -{{ - "actions": [ - {{"primitive": "name", "args": {{...}}}} - ], - "response_template": "Message to send user (can use Markdown formatting)" -}} - -TONE: Be direct, concise, and clear. Use short paragraphs with line breaks. -Avoid formal/legalistic language AND casual interjections (no "Hey!"). -Professional but approachable. Get to the point quickly. -""" - - try: - result = self._call_llm(prompt) - plan = self._extract_json(result) - plan["intent_type"] = intent.get("intent_type") - plan["constitutional_basis"] = constitutional_guidance.get("citations", []) - return plan - except Exception as e: - return { - "error": f"Could not plan actions: {str(e)}", - "intent": intent, - "guidance": constitutional_guidance, - } - - def _execute_plan( - self, plan: Dict[str, Any], actor: str - ) -> Dict[str, Any]: - """ - Execute the planned actions using primitives. - - Args: - plan: Action plan - actor: Who initiated - - Returns: - Execution result - """ - if plan.get("error"): - return {"response": plan["error"], "success": False} - - executed_actions = [] - process_id = None - - try: - for action in plan.get("actions", []): - primitive = action["primitive"] - args = action["args"] - - # Get the primitive function - if hasattr(self.primitives, primitive): - func = getattr(self.primitives, primitive) - - # Handle special cases like deadline calculation - if "when" in args and args["when"] == "deadline": - # Calculate from process deadline - if process_id: - process = queries.get_process(self.db, process_id) - args["when"] = process.deadline - - result = func(**args) - - # Track process ID for response - if primitive == "create_process": - process_id = result - - executed_actions.append( - {"primitive": primitive, "result": result} - ) - else: - raise ValueError(f"Unknown primitive: {primitive}") - - # Build response - response_template = plan.get("response_template", "Action completed.") - response = response_template.format( - process_id=process_id, action_id=executed_actions[0].get("result") - if executed_actions - else None - ) - - return { - "response": response, - "success": True, - "process_id": process_id, - "actions_taken": executed_actions, - "constitutional_basis": plan.get("constitutional_basis"), - } - - except Exception as e: - return { - "response": f"Error executing actions: {str(e)}", - "success": False, - "partial_actions": executed_actions, - } - - def _handle_ambiguity( - self, - request: str, - actor: str, - constitutional_guidance: Dict[str, Any], - ) -> Dict[str, Any]: - """ - Handle constitutional ambiguity by requesting clarification. - - Args: - request: Original request - actor: Who made request - constitutional_guidance: The ambiguous guidance - - Returns: - Response explaining ambiguity - """ - ambiguity = constitutional_guidance.get("ambiguity", "Constitutional interpretation unclear") - - # Create clarification request - clarification = queries.create_clarification( - session=self.db, - question=f"Ambiguity in request '{request}': {ambiguity}", - ) - - response = f"""I found something unclear in the constitution regarding your request. - -Issue: {ambiguity} - -This needs community clarification. Discussion welcome. - -Clarification ID: {clarification.id} -""" - - return { - "response": response, - "success": False, - "requires_clarification": True, - "clarification_id": clarification.id, - } - - def check_deadlines(self) -> List[Dict[str, Any]]: - """ - Check for processes that have passed their deadline. - This should be called periodically by a background task. - - Returns: - List of processes that were completed - """ - overdue_processes = queries.get_processes_past_deadline(self.db) - completed = [] - - for process in overdue_processes: - # Count votes - counts = self.primitives.count_votes(process.id) - - # Determine threshold from process state - threshold_type = process.state_data.get( - "voting_threshold", "simple_majority" - ) - - # Check if passed - passed = self.primitives.check_threshold(counts, threshold_type) - - outcome = "passed" if passed else "failed" - - # Complete the process - self.primitives.complete_process( - process_id=process.id, - outcome=outcome, - reasoning=f"Vote counts: {counts}. Threshold: {threshold_type}. Result: {outcome}", - ) - - completed.append( - { - "process_id": process.id, - "outcome": outcome, - "vote_counts": counts, - } - ) - - return completed - - def _call_llm(self, prompt: str) -> str: - """Call the LLM via llm CLI""" - import os - - cmd = ["llm"] - if self.model: - cmd.extend(["-m", self.model]) - cmd.append(prompt) - - # Set up environment with API keys - env = os.environ.copy() - if self.api_keys.get('openai'): - env['OPENAI_API_KEY'] = self.api_keys['openai'] - if self.api_keys.get('anthropic'): - env['ANTHROPIC_API_KEY'] = self.api_keys['anthropic'] - - result = subprocess.run(cmd, capture_output=True, text=True, check=True, env=env) - return result.stdout.strip() - - def _extract_json(self, text: str) -> Dict[str, Any]: - """Extract JSON from LLM response""" - # Handle markdown code blocks - if "```json" in text: - start = text.find("```json") + 7 - end = text.find("```", start) - json_str = text[start:end].strip() - elif "```" in text: - start = text.find("```") + 3 - end = text.find("```", start) - json_str = text[start:end].strip() - else: - json_str = text - - return json.loads(json_str)