Modpol/agentic-govbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove legacy agent file (preserved in git history)

Browse Source 
Git provides version history - no need to keep old files in the codebase.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Nathan Schneider
2026-02-08 18:58:03 -07:00
parent a92236e528
commit 422f0859f4
1 changed files with 0 additions and 659 deletions
Download Patch File Download Diff File Expand all files Collapse all files

659
src/govbot/agent_legacy.py
View File

@@ -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)