Switch to LLM-driven agent with zero hard-coded governance logic
Replace old rule-based agent with pure LLM interpretation system. Agent Changes: - Rename agent.py → agent_legacy.py (preserve old hard-coded agent) - Rename agent_refactored.py → agent.py (make LLM agent primary) - Agent now interprets constitution to understand authority and processes - No hard-coded checks for specific users, roles, or governance models - Fully generic: works with any constitutional design Constitution Interpreter: - Updated interpret_proposal() to detect authority structures from text - LLM determines who has decision-making power from constitution - No assumptions about voting, proposals, or specific governance models Mastodon Formatting: - Improved line break handling for bullet points and paragraphs - Better plain-text formatting for Mastodon posts Primitives: - Added support for admin_approval threshold type Architecture: - Bot now uses pure LLM interpretation instead of scripted logic - Each instance can develop implementation guidelines separately - Guidelines not included in main codebase (instance-specific) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Nathan Schneider
1021
src/govbot/agent.py
1021
src/govbot/agent.py
File diff suppressed because it is too large
Load Diff
659
src/govbot/agent_legacy.py
Normal file
659
src/govbot/agent_legacy.py
Normal file
@@ -0,0 +1,659 @@
|
||||
"""
|
||||
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)
|
||||
@@ -1,561 +0,0 @@
|
||||
"""
|
||||
Refactored Agentic Governance Bot.
|
||||
|
||||
This refactored agent relies on:
|
||||
1. LLM interpretation (not hard-coded logic)
|
||||
2. Structured memory (not just database records)
|
||||
3. Tools for correctness (calculator, datetime, etc.)
|
||||
4. Audit trails (human-readable explanations)
|
||||
|
||||
Key principle: Constitution defines ALL governance rules in natural language.
|
||||
The agent interprets and executes based on constitution + memory + tools.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import Dict, Any, Optional, List
|
||||
from datetime import datetime, timedelta
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .governance.constitution import ConstitutionalReasoner
|
||||
from .memory import GovernanceMemory, ProcessMemory, ProcessStatus, Event, Decision
|
||||
from .tools import GovernanceTools, ToolRegistry
|
||||
from .audit import AuditTrail, create_audit_trail
|
||||
|
||||
|
||||
class AgenticGovernanceBot:
|
||||
"""
|
||||
Agentic governance bot that interprets constitution and uses memory + tools.
|
||||
|
||||
This bot does NOT have hard-coded governance logic. Instead:
|
||||
- Reads constitution to understand rules
|
||||
- Uses memory to track state
|
||||
- Uses tools for calculations
|
||||
- Makes decisions through LLM reasoning
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
db_session: Session,
|
||||
constitution_path: str,
|
||||
model: Optional[str] = None,
|
||||
api_keys: Optional[Dict[str, str]] = None,
|
||||
):
|
||||
"""
|
||||
Initialize the agentic governance bot.
|
||||
|
||||
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.memory = GovernanceMemory(db_session)
|
||||
self.tools = GovernanceTools()
|
||||
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 using agentic interpretation.
|
||||
|
||||
Flow:
|
||||
1. Understand what the user wants (intent parsing)
|
||||
2. Query constitution for relevant rules
|
||||
3. Query memory for current state
|
||||
4. Reason about what to do (using LLM)
|
||||
5. Execute actions using tools
|
||||
6. Update memory with decision
|
||||
7. Generate audit trail
|
||||
|
||||
Args:
|
||||
request: Natural language request
|
||||
actor: Who made the request
|
||||
context: Optional context (thread ID, etc.)
|
||||
|
||||
Returns:
|
||||
Response dictionary with action taken and audit trail
|
||||
"""
|
||||
# Step 1: Parse intent
|
||||
intent_result = self._parse_intent_with_llm(request, actor)
|
||||
|
||||
if "error" in intent_result:
|
||||
return {"response": intent_result["error"], "success": False}
|
||||
|
||||
# Step 2: Query constitution
|
||||
constitutional_guidance = self.constitution.query(
|
||||
question=intent_result["query"],
|
||||
context=f"Actor: {actor}\nRequest: {request}"
|
||||
)
|
||||
|
||||
# Step 3: Query memory for relevant state
|
||||
memory_context = self._gather_memory_context(intent_result, actor)
|
||||
|
||||
# Step 4: Use LLM to decide what to do
|
||||
decision_result = self._make_decision_with_llm(
|
||||
intent=intent_result,
|
||||
constitution=constitutional_guidance,
|
||||
memory=memory_context,
|
||||
actor=actor,
|
||||
context=context
|
||||
)
|
||||
|
||||
# Step 5: Execute the decision
|
||||
execution_result = self._execute_decision(
|
||||
decision_result,
|
||||
actor,
|
||||
context
|
||||
)
|
||||
|
||||
# Step 6: Generate audit trail
|
||||
if execution_result.get("process_id"):
|
||||
process = self.memory.get_process(execution_result["process_id"])
|
||||
if process and process.decisions:
|
||||
audit = create_audit_trail(process, process.decisions[-1])
|
||||
execution_result["audit_trail"] = audit
|
||||
|
||||
return execution_result
|
||||
|
||||
def _parse_intent_with_llm(self, request: str, actor: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Use LLM to understand what the user wants to do.
|
||||
|
||||
Returns:
|
||||
Dict with:
|
||||
- intent_type: "create_proposal", "cast_vote", "query", etc.
|
||||
- query: Question to ask constitution
|
||||
- parameters: Extracted parameters
|
||||
"""
|
||||
prompt = f"""Analyze this governance request and determine the intent.
|
||||
|
||||
Request: {request}
|
||||
Actor: {actor}
|
||||
|
||||
What is the user trying to do? Choose from:
|
||||
- create_proposal: User wants to create a governance proposal
|
||||
- cast_vote: User wants to vote on something
|
||||
- query_status: User wants to know the status of something
|
||||
- query_constitution: User wants to know governance rules
|
||||
- dispute_initiation: User wants to start a dispute resolution process
|
||||
- other: Something else
|
||||
|
||||
Also extract any relevant parameters (proposal text, vote type, process ID, etc.).
|
||||
|
||||
Return your analysis as JSON:
|
||||
{{
|
||||
"intent_type": "...",
|
||||
"query": "question to ask the constitution",
|
||||
"parameters": {{}},
|
||||
"confidence": "high|medium|low"
|
||||
}}
|
||||
"""
|
||||
|
||||
try:
|
||||
result = self.constitution._call_llm(prompt)
|
||||
# Parse JSON from response
|
||||
# (In production, would use proper JSON parsing from LLM response)
|
||||
return json.loads(result.get("answer", "{}"))
|
||||
except Exception as e:
|
||||
return {"error": f"Failed to parse intent: {e}"}
|
||||
|
||||
def _gather_memory_context(
|
||||
self,
|
||||
intent: Dict[str, Any],
|
||||
actor: str
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Gather relevant information from memory based on intent.
|
||||
|
||||
Returns:
|
||||
Dict with memory context
|
||||
"""
|
||||
context = {}
|
||||
|
||||
# Get active processes
|
||||
active_processes = self.memory.get_active_processes()
|
||||
context["active_processes"] = [
|
||||
self.memory.summarize_for_llm(p.id) for p in active_processes[:5]
|
||||
]
|
||||
|
||||
# Get overdue processes
|
||||
overdue = self.memory.get_overdue_processes()
|
||||
if overdue:
|
||||
context["overdue_processes"] = [p.id for p in overdue]
|
||||
|
||||
# If specific process mentioned, get its details
|
||||
if "process_id" in intent.get("parameters", {}):
|
||||
process_id = intent["parameters"]["process_id"]
|
||||
process = self.memory.get_process(process_id)
|
||||
if process:
|
||||
context["target_process"] = self.memory.summarize_for_llm(process_id)
|
||||
|
||||
# Get recent precedent if relevant
|
||||
if intent.get("intent_type") == "query_constitution":
|
||||
recent_decisions = self.memory.search_decisions(limit=5)
|
||||
context["recent_precedent"] = [
|
||||
f"{d.decision_type}: {d.result}" for d in recent_decisions
|
||||
]
|
||||
|
||||
return context
|
||||
|
||||
def _make_decision_with_llm(
|
||||
self,
|
||||
intent: Dict[str, Any],
|
||||
constitution: Dict[str, Any],
|
||||
memory: Dict[str, Any],
|
||||
actor: str,
|
||||
context: Optional[Dict[str, Any]]
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Use LLM to decide what action to take.
|
||||
|
||||
This is where the agent interprets the constitution and decides
|
||||
how to handle the request.
|
||||
|
||||
Returns:
|
||||
Decision dict with:
|
||||
- action: What to do
|
||||
- reasoning: Why
|
||||
- constitution_citations: Which articles apply
|
||||
- calculations: Any math needed
|
||||
- state_updates: Changes to memory
|
||||
"""
|
||||
prompt = f"""You are a governance bot interpreting a community constitution.
|
||||
|
||||
INTENT:
|
||||
{json.dumps(intent, indent=2)}
|
||||
|
||||
CONSTITUTIONAL GUIDANCE:
|
||||
{json.dumps(constitution, indent=2)}
|
||||
|
||||
CURRENT MEMORY STATE:
|
||||
{json.dumps(memory, indent=2)}
|
||||
|
||||
ACTOR: {actor}
|
||||
|
||||
Based on the constitution and current state, decide what action to take.
|
||||
|
||||
For proposals:
|
||||
- What type of proposal is this? (standard, urgent, constitutional, etc.)
|
||||
- What discussion period does the constitution specify?
|
||||
- What voting threshold is required?
|
||||
- Are there any special requirements?
|
||||
|
||||
For votes:
|
||||
- Is this person eligible to vote on this process?
|
||||
- What vote types are allowed?
|
||||
- Should this vote be recorded?
|
||||
|
||||
For deadline checks:
|
||||
- Has the deadline passed?
|
||||
- What votes were cast?
|
||||
- What threshold does the constitution require?
|
||||
- Does the proposal pass or fail?
|
||||
|
||||
Available tools for calculations:
|
||||
- calculate(expression, variables): Evaluate math expressions
|
||||
- get_datetime(): Get current time
|
||||
- is_past_deadline(deadline): Check if deadline passed
|
||||
- tally(votes, key): Count votes by type
|
||||
- random_select(items, count): Random selection
|
||||
|
||||
Return your decision as JSON:
|
||||
{{
|
||||
"action": "create_process|record_vote|complete_process|query_response",
|
||||
"reasoning": "explain your interpretation",
|
||||
"constitution_citations": ["Article X, Section Y", ...],
|
||||
"parameters": {{
|
||||
// Action-specific parameters
|
||||
"process_type": "...",
|
||||
"deadline_days": X,
|
||||
"threshold_expression": "agree > disagree",
|
||||
// etc.
|
||||
}},
|
||||
"calculations": [
|
||||
{{
|
||||
"tool": "calculate",
|
||||
"expression": "agree > disagree",
|
||||
"variables": {{"agree": 10, "disagree": 3}}
|
||||
}}
|
||||
]
|
||||
}}
|
||||
"""
|
||||
|
||||
try:
|
||||
result = self.constitution._call_llm(prompt)
|
||||
decision = json.loads(result.get("answer", "{}"))
|
||||
|
||||
# Execute any calculations using tools
|
||||
if "calculations" in decision:
|
||||
for calc in decision["calculations"]:
|
||||
tool_name = calc["tool"]
|
||||
if tool_name == "calculate":
|
||||
calc["result"] = self.tools.calculate(
|
||||
calc["expression"],
|
||||
calc["variables"]
|
||||
)
|
||||
# Handle other tools...
|
||||
|
||||
return decision
|
||||
|
||||
except Exception as e:
|
||||
return {
|
||||
"action": "error",
|
||||
"reasoning": f"Failed to make decision: {e}",
|
||||
"constitution_citations": []
|
||||
}
|
||||
|
||||
def _execute_decision(
|
||||
self,
|
||||
decision: Dict[str, Any],
|
||||
actor: str,
|
||||
context: Optional[Dict[str, Any]]
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Execute the decision made by the LLM.
|
||||
|
||||
This updates memory and performs platform actions.
|
||||
"""
|
||||
action = decision.get("action")
|
||||
params = decision.get("parameters", {})
|
||||
|
||||
try:
|
||||
if action == "create_process":
|
||||
return self._create_process_from_decision(decision, actor, context)
|
||||
|
||||
elif action == "record_vote":
|
||||
return self._record_vote_from_decision(decision, actor, context)
|
||||
|
||||
elif action == "complete_process":
|
||||
return self._complete_process_from_decision(decision, actor)
|
||||
|
||||
elif action == "query_response":
|
||||
return {
|
||||
"response": decision.get("reasoning"),
|
||||
"constitution_citations": decision.get("constitution_citations", []),
|
||||
"success": True
|
||||
}
|
||||
|
||||
else:
|
||||
return {
|
||||
"response": f"Unknown action: {action}",
|
||||
"success": False
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
return {
|
||||
"response": f"Error executing decision: {e}",
|
||||
"success": False
|
||||
}
|
||||
|
||||
def _create_process_from_decision(
|
||||
self,
|
||||
decision: Dict[str, Any],
|
||||
actor: str,
|
||||
context: Optional[Dict[str, Any]]
|
||||
) -> Dict[str, Any]:
|
||||
"""Create a new governance process based on LLM decision"""
|
||||
params = decision["parameters"]
|
||||
|
||||
# Calculate deadline
|
||||
deadline_days = params.get("deadline_days", 7)
|
||||
deadline = self.tools.datetime_add(
|
||||
self.tools.get_datetime(),
|
||||
days=deadline_days
|
||||
)
|
||||
|
||||
# Generate process ID
|
||||
process_id = f"process_{int(datetime.utcnow().timestamp())}"
|
||||
|
||||
# Create in memory
|
||||
process = self.memory.create_process(
|
||||
process_id=process_id,
|
||||
process_type=params.get("process_type", "unknown"),
|
||||
created_by=actor,
|
||||
constitution_basis=decision.get("constitution_citations", []),
|
||||
deadline=deadline,
|
||||
initial_state=params.get("initial_state", {}),
|
||||
metadata=context or {}
|
||||
)
|
||||
|
||||
# Log creation event
|
||||
self.memory.add_event(
|
||||
process_id=process_id,
|
||||
actor=actor,
|
||||
event_type="process_created",
|
||||
data=params,
|
||||
context=f"Process created by {actor}"
|
||||
)
|
||||
|
||||
# Log creation decision
|
||||
self.memory.add_decision(
|
||||
process_id=process_id,
|
||||
decision_type="process_creation",
|
||||
reasoning=decision["reasoning"],
|
||||
constitution_citations=decision.get("constitution_citations", []),
|
||||
result="created"
|
||||
)
|
||||
|
||||
return {
|
||||
"response": f"Created {params.get('process_type')} (ID: {process_id}). Deadline: {deadline.strftime('%Y-%m-%d %H:%M UTC')}",
|
||||
"process_id": process_id,
|
||||
"deadline": deadline.isoformat(),
|
||||
"success": True
|
||||
}
|
||||
|
||||
def _record_vote_from_decision(
|
||||
self,
|
||||
decision: Dict[str, Any],
|
||||
actor: str,
|
||||
context: Optional[Dict[str, Any]]
|
||||
) -> Dict[str, Any]:
|
||||
"""Record a vote based on LLM decision"""
|
||||
params = decision["parameters"]
|
||||
process_id = params.get("process_id")
|
||||
|
||||
if not process_id:
|
||||
return {"response": "No process ID specified", "success": False}
|
||||
|
||||
# Get process
|
||||
process = self.memory.get_process(process_id)
|
||||
if not process:
|
||||
return {"response": f"Process {process_id} not found", "success": False}
|
||||
|
||||
# Record vote in state
|
||||
votes = process.state.get("votes", {})
|
||||
votes[actor] = {
|
||||
"vote": params.get("vote_type"),
|
||||
"timestamp": datetime.utcnow().isoformat()
|
||||
}
|
||||
|
||||
self.memory.update_process(
|
||||
process_id=process_id,
|
||||
state_updates={"votes": votes}
|
||||
)
|
||||
|
||||
# Log event
|
||||
self.memory.add_event(
|
||||
process_id=process_id,
|
||||
actor=actor,
|
||||
event_type="vote_cast",
|
||||
data={"vote": params.get("vote_type")},
|
||||
context=f"{actor} voted {params.get('vote_type')}"
|
||||
)
|
||||
|
||||
return {
|
||||
"response": f"Vote recorded: {params.get('vote_type')}",
|
||||
"process_id": process_id,
|
||||
"success": True
|
||||
}
|
||||
|
||||
def _complete_process_from_decision(
|
||||
self,
|
||||
decision: Dict[str, Any],
|
||||
actor: str
|
||||
) -> Dict[str, Any]:
|
||||
"""Complete a process based on LLM decision"""
|
||||
params = decision["parameters"]
|
||||
process_id = params.get("process_id")
|
||||
|
||||
process = self.memory.get_process(process_id)
|
||||
if not process:
|
||||
return {"response": f"Process {process_id} not found", "success": False}
|
||||
|
||||
# Update status
|
||||
self.memory.update_process(
|
||||
process_id=process_id,
|
||||
status=ProcessStatus.COMPLETED
|
||||
)
|
||||
|
||||
# Log completion decision
|
||||
calculation_used = None
|
||||
calculation_vars = None
|
||||
calculation_result = None
|
||||
|
||||
if decision.get("calculations"):
|
||||
calc = decision["calculations"][0] # Use first calculation
|
||||
calculation_used = calc.get("expression")
|
||||
calculation_vars = calc.get("variables")
|
||||
calculation_result = calc.get("result")
|
||||
|
||||
self.memory.add_decision(
|
||||
process_id=process_id,
|
||||
decision_type="process_completion",
|
||||
reasoning=decision["reasoning"],
|
||||
constitution_citations=decision.get("constitution_citations", []),
|
||||
result=params.get("outcome", "completed"),
|
||||
calculation_used=calculation_used,
|
||||
calculation_variables=calculation_vars,
|
||||
calculation_result=calculation_result
|
||||
)
|
||||
|
||||
# Log event
|
||||
self.memory.add_event(
|
||||
process_id=process_id,
|
||||
actor="bot",
|
||||
event_type="process_completed",
|
||||
data={"outcome": params.get("outcome")},
|
||||
context=f"Process completed with outcome: {params.get('outcome')}"
|
||||
)
|
||||
|
||||
return {
|
||||
"response": f"Process {process_id} completed: {params.get('outcome')}",
|
||||
"process_id": process_id,
|
||||
"outcome": params.get("outcome"),
|
||||
"success": True
|
||||
}
|
||||
|
||||
def check_deadlines(self) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Check for processes that have reached their deadline.
|
||||
|
||||
This is called periodically (e.g., every hour) to evaluate
|
||||
processes that need decisions.
|
||||
|
||||
Returns:
|
||||
List of completed process results
|
||||
"""
|
||||
results = []
|
||||
overdue = self.memory.get_overdue_processes()
|
||||
|
||||
for process in overdue:
|
||||
# Ask LLM to evaluate this process
|
||||
evaluation = self._evaluate_process_deadline(process)
|
||||
results.append(evaluation)
|
||||
|
||||
return results
|
||||
|
||||
def _evaluate_process_deadline(self, process: ProcessMemory) -> Dict[str, Any]:
|
||||
"""
|
||||
Use LLM to evaluate a process that reached its deadline.
|
||||
"""
|
||||
# Get vote tally
|
||||
votes = process.state.get("votes", {})
|
||||
vote_tally = self.tools.tally(
|
||||
[{"vote": v["vote"]} for v in votes.values()],
|
||||
"vote"
|
||||
)
|
||||
|
||||
# Ask constitution what threshold is needed
|
||||
threshold_query = self.constitution.query(
|
||||
question=f"For a {process.type}, what voting threshold is required for passage?",
|
||||
context=self.memory.summarize_for_llm(process.id)
|
||||
)
|
||||
|
||||
# Use LLM to decide if threshold is met
|
||||
decision_result = self._make_decision_with_llm(
|
||||
intent={"intent_type": "deadline_check"},
|
||||
constitution=threshold_query,
|
||||
memory={"votes": vote_tally, "process": process.to_dict()},
|
||||
actor="bot",
|
||||
context=None
|
||||
)
|
||||
|
||||
# Execute the decision
|
||||
return self._execute_decision(decision_result, "bot", None)
|
||||
@@ -256,12 +256,27 @@ Format your response as JSON:
|
||||
"""
|
||||
question = f"""Given this proposal: "{proposal_text}"
|
||||
|
||||
What type of proposal is this according to the constitution?
|
||||
What are the requirements (timeline, voting threshold, etc.)?
|
||||
According to the constitution:
|
||||
1. Who has authority to decide on this proposal?
|
||||
2. What is the process for handling this request?
|
||||
3. Does this require voting, or does someone have direct authority?
|
||||
"""
|
||||
|
||||
result = self.query(question)
|
||||
|
||||
# Check if constitution mentions direct authority (e.g., benevolent dictator)
|
||||
answer_lower = result.get('answer', '').lower()
|
||||
if '@admin' in answer_lower and any(word in answer_lower for word in ['authority', 'decides', 'makes', 'final', 'power']):
|
||||
# Benevolent dictator model - direct to @admin
|
||||
return {
|
||||
"proposal_type": "admin_decision",
|
||||
"discussion_period_days": 0,
|
||||
"voting_threshold": "admin_approval",
|
||||
"special_requirements": ["requires_admin_decision"],
|
||||
"decision_maker": "@admin",
|
||||
"constitutional_basis": result,
|
||||
}
|
||||
|
||||
# Extract structured information from the answer
|
||||
try:
|
||||
prompt = f"""Based on this constitutional interpretation:
|
||||
@@ -271,9 +286,10 @@ Extract structured information as JSON:
|
||||
{{
|
||||
"title": "concise title for the proposal (max 80 chars)",
|
||||
"description": "clear 1-2 sentence description of what the proposal does",
|
||||
"proposal_type": "standard|urgent|constitutional_amendment",
|
||||
"discussion_period_days": number,
|
||||
"voting_threshold": "description of threshold",
|
||||
"proposal_type": "standard|urgent|constitutional_amendment|admin_decision|community_input",
|
||||
"discussion_period_days": number (or 0 if no voting period),
|
||||
"voting_threshold": "description of threshold or approval mechanism",
|
||||
"decision_maker": "who makes the final decision (e.g., '@admin', 'community_vote', 'consensus')",
|
||||
"special_requirements": ["list", "of", "requirements"]
|
||||
}}
|
||||
"""
|
||||
@@ -282,15 +298,28 @@ Extract structured information as JSON:
|
||||
proposal_info["constitutional_basis"] = result
|
||||
return proposal_info
|
||||
except Exception as e:
|
||||
# Fallback
|
||||
return {
|
||||
"proposal_type": "standard",
|
||||
"discussion_period_days": 6,
|
||||
"voting_threshold": "simple_majority",
|
||||
"special_requirements": [],
|
||||
"constitutional_basis": result,
|
||||
"error": str(e),
|
||||
}
|
||||
# Check constitution content for governance model
|
||||
if '@admin' in self.constitution_text and 'final authority' in self.constitution_text.lower():
|
||||
# Benevolent dictator fallback
|
||||
return {
|
||||
"proposal_type": "admin_decision",
|
||||
"discussion_period_days": 0,
|
||||
"voting_threshold": "admin_approval",
|
||||
"special_requirements": ["requires_admin_decision"],
|
||||
"decision_maker": "@admin",
|
||||
"constitutional_basis": result,
|
||||
"error": str(e),
|
||||
}
|
||||
else:
|
||||
# Democratic fallback
|
||||
return {
|
||||
"proposal_type": "standard",
|
||||
"discussion_period_days": 6,
|
||||
"voting_threshold": "simple_majority",
|
||||
"special_requirements": [],
|
||||
"constitutional_basis": result,
|
||||
"error": str(e),
|
||||
}
|
||||
|
||||
def check_ambiguity(self, question: str) -> Optional[str]:
|
||||
"""
|
||||
|
||||
@@ -297,6 +297,11 @@ class GovernancePrimitives:
|
||||
return False
|
||||
return (agree / total) >= (2 / 3)
|
||||
|
||||
elif threshold_type == "admin_approval":
|
||||
# Admin decision model - no voting threshold
|
||||
# This type means admin must approve, not vote counting
|
||||
return False # Requires manual admin approval
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unknown threshold type: {threshold_type}")
|
||||
|
||||
|
||||
@@ -679,15 +679,30 @@ class MastodonAdapter(PlatformAdapter):
|
||||
text = re.sub(r'_([^_]+)_', r'\1', text) # _italic_ -> italic
|
||||
text = re.sub(r'`([^`]+)`', r'\1', text) # `code` -> code
|
||||
|
||||
# Remove headers but keep the text
|
||||
text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
|
||||
# Remove headers but keep the text with extra spacing
|
||||
text = re.sub(r'^#{1,6}\s+(.+)$', r'\1\n', text, flags=re.MULTILINE)
|
||||
|
||||
# Convert Markdown lists to simple text with bullets
|
||||
# Ensure each bullet point is on its own line
|
||||
text = re.sub(r'^\s*[-*+]\s+', '• ', text, flags=re.MULTILINE)
|
||||
|
||||
# Remove link formatting but keep URLs: [text](url) -> text (url)
|
||||
text = re.sub(r'\[([^\]]+)\]\(([^\)]+)\)', r'\1 (\2)', text)
|
||||
|
||||
# Ensure proper paragraph spacing for Mastodon
|
||||
# Replace single newlines within paragraphs, but preserve double newlines
|
||||
# First, protect double (or more) newlines
|
||||
text = re.sub(r'\n\n+', '<<<PARAGRAPH>>>', text)
|
||||
# Then ensure bullet points and other single newlines are preserved
|
||||
# (Mastodon respects single newlines in plain text)
|
||||
# Restore paragraph breaks
|
||||
text = text.replace('<<<PARAGRAPH>>>', '\n\n')
|
||||
|
||||
# Clean up any extra whitespace but preserve intentional line breaks
|
||||
text = re.sub(r' +', ' ', text) # Multiple spaces -> single space
|
||||
text = re.sub(r'\n ', '\n', text) # Remove spaces after newlines
|
||||
text = re.sub(r' \n', '\n', text) # Remove spaces before newlines
|
||||
|
||||
return text
|
||||
|
||||
def _map_visibility(self, visibility: MessageVisibility) -> str:
|
||||
|
||||
Reference in New Issue
Block a user