439 lines
14 KiB
Python
439 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
AI-powered agents for extracting structured information from discussions.
|
|
|
|
Supports multiple AI providers via CLI commands:
|
|
- Claude Code: claude -p "prompt"
|
|
- Gemini CLI: gemini "prompt"
|
|
- OpenAI Codex: codex "prompt"
|
|
- Or direct API: anthropic library
|
|
|
|
These agents process natural language discussions and extract:
|
|
- Questions (answered, open, partial)
|
|
- Action items (TODO → assignment → DONE)
|
|
- Decisions (ADR-style with rationale)
|
|
- @mentions (routing and tracking)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
|
|
def get_ai_config() -> dict[str, str]:
|
|
"""
|
|
Get AI provider configuration from environment or git config.
|
|
|
|
Environment variables (highest priority):
|
|
CDEV_AI_PROVIDER: "claude-cli", "claude-api", "gemini-cli", "codex-cli"
|
|
CDEV_AI_COMMAND: Command template, e.g., "claude -p {prompt}"
|
|
|
|
Git config (fallback):
|
|
cascadingdev.aiprovider
|
|
cascadingdev.aicommand
|
|
|
|
Returns dict with 'provider' and 'command' keys.
|
|
"""
|
|
# Try environment first
|
|
provider = os.environ.get("CDEV_AI_PROVIDER")
|
|
command = os.environ.get("CDEV_AI_COMMAND")
|
|
|
|
# Fall back to git config
|
|
if not provider or not command:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "config", "--get", "cascadingdev.aiprovider"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False
|
|
)
|
|
if result.returncode == 0:
|
|
provider = result.stdout.strip()
|
|
|
|
result = subprocess.run(
|
|
["git", "config", "--get", "cascadingdev.aicommand"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False
|
|
)
|
|
if result.returncode == 0:
|
|
command = result.stdout.strip()
|
|
except Exception:
|
|
pass
|
|
|
|
# Default to claude-cli if nothing configured
|
|
if not provider:
|
|
provider = "claude-cli"
|
|
if not command:
|
|
command = "claude -p '{prompt}'"
|
|
|
|
return {"provider": provider, "command": command}
|
|
|
|
|
|
def call_ai_cli(prompt: str, content: str) -> dict[str, Any] | None:
|
|
"""
|
|
Call AI via CLI command (claude, gemini, codex, etc.).
|
|
|
|
The command template can use {prompt} and {content} placeholders.
|
|
Content is passed via a temporary file in .git/ai-agents-temp/ to avoid
|
|
shell escaping issues while remaining accessible to Claude CLI.
|
|
|
|
Returns parsed JSON response or None on failure.
|
|
"""
|
|
config = get_ai_config()
|
|
command_template = config["command"]
|
|
|
|
# Create temporary file in .git/ai-agents-temp/ (accessible to Claude CLI)
|
|
# Using .git/ ensures it's gitignored and Claude has permission
|
|
temp_dir = Path.cwd() / ".git" / "ai-agents-temp"
|
|
temp_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Use PID for unique filename to avoid conflicts
|
|
content_file = temp_dir / f"discussion-{os.getpid()}.md"
|
|
content_file.write_text(content, encoding='utf-8')
|
|
|
|
try:
|
|
# Build full prompt
|
|
full_prompt = f"{prompt}\n\nPlease read the discussion content from: {content_file}\n\nReturn ONLY valid JSON, no other text."
|
|
|
|
# Format command
|
|
# Support both {prompt} placeholder and direct append
|
|
if "{prompt}" in command_template:
|
|
command = command_template.format(prompt=full_prompt)
|
|
else:
|
|
# Assume command expects prompt as argument
|
|
command = f"{command_template} '{full_prompt}'"
|
|
|
|
# Execute command
|
|
result = subprocess.run(
|
|
command,
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
check=False
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
sys.stderr.write(f"[agents] warning: AI command failed with code {result.returncode}\n")
|
|
if result.stderr:
|
|
sys.stderr.write(f"[agents] {result.stderr[:200]}\n")
|
|
return None
|
|
|
|
response_text = result.stdout.strip()
|
|
|
|
# Try to parse as JSON
|
|
# Look for JSON in code blocks or raw
|
|
if "```json" in response_text:
|
|
json_start = response_text.find("```json") + 7
|
|
json_end = response_text.find("```", json_start)
|
|
response_text = response_text[json_start:json_end].strip()
|
|
elif "```" in response_text:
|
|
json_start = response_text.find("```") + 3
|
|
json_end = response_text.find("```", json_start)
|
|
response_text = response_text[json_start:json_end].strip()
|
|
|
|
return json.loads(response_text)
|
|
|
|
except subprocess.TimeoutExpired:
|
|
sys.stderr.write("[agents] warning: AI command timed out after 60s\n")
|
|
return None
|
|
except json.JSONDecodeError as e:
|
|
sys.stderr.write(f"[agents] warning: Failed to parse AI response as JSON: {e}\n")
|
|
sys.stderr.write(f"[agents] Response preview: {response_text[:200]}...\n")
|
|
return None
|
|
except Exception as e:
|
|
sys.stderr.write(f"[agents] error calling AI: {e}\n")
|
|
return None
|
|
finally:
|
|
# Clean up temp file
|
|
try:
|
|
content_file.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def call_ai_api(prompt: str, content: str) -> dict[str, Any] | None:
|
|
"""
|
|
Call AI via direct API (anthropic library).
|
|
|
|
Returns parsed JSON response or None on failure.
|
|
Requires ANTHROPIC_API_KEY environment variable.
|
|
"""
|
|
api_key = os.environ.get("ANTHROPIC_API_KEY")
|
|
if not api_key:
|
|
return None
|
|
|
|
try:
|
|
import anthropic
|
|
except ImportError:
|
|
return None
|
|
|
|
try:
|
|
client = anthropic.Anthropic(api_key=api_key)
|
|
|
|
message = client.messages.create(
|
|
model="claude-3-5-sonnet-20241022",
|
|
max_tokens=4096,
|
|
messages=[{
|
|
"role": "user",
|
|
"content": f"{prompt}\n\n<discussion_content>\n{content}\n</discussion_content>"
|
|
}]
|
|
)
|
|
|
|
# Extract text from response
|
|
response_text = ""
|
|
for block in message.content:
|
|
if hasattr(block, "text"):
|
|
response_text += block.text
|
|
|
|
# Try to parse as JSON
|
|
if "```json" in response_text:
|
|
json_start = response_text.find("```json") + 7
|
|
json_end = response_text.find("```", json_start)
|
|
response_text = response_text[json_start:json_end].strip()
|
|
elif "```" in response_text:
|
|
json_start = response_text.find("```") + 3
|
|
json_end = response_text.find("```", json_start)
|
|
response_text = response_text[json_start:json_end].strip()
|
|
|
|
return json.loads(response_text)
|
|
|
|
except Exception as e:
|
|
sys.stderr.write(f"[agents] error calling API: {e}\n")
|
|
return None
|
|
|
|
|
|
def call_ai(prompt: str, content: str) -> dict[str, Any] | None:
|
|
"""
|
|
Call configured AI provider (CLI or API).
|
|
|
|
First tries CLI-based provider (claude, gemini, codex).
|
|
Falls back to direct API if ANTHROPIC_API_KEY is set.
|
|
|
|
Returns parsed JSON response or None on failure.
|
|
"""
|
|
config = get_ai_config()
|
|
provider = config["provider"]
|
|
|
|
# Try CLI-based providers first
|
|
if provider.endswith("-cli") or not os.environ.get("ANTHROPIC_API_KEY"):
|
|
result = call_ai_cli(prompt, content)
|
|
if result is not None:
|
|
return result
|
|
|
|
# Fall back to API if available
|
|
if os.environ.get("ANTHROPIC_API_KEY"):
|
|
return call_ai_api(prompt, content)
|
|
|
|
# No AI available
|
|
if provider.endswith("-cli"):
|
|
sys.stderr.write(f"[agents] warning: AI CLI provider '{provider}' not responding, skipping AI processing\n")
|
|
else:
|
|
sys.stderr.write("[agents] warning: No AI provider configured, skipping AI processing\n")
|
|
return None
|
|
|
|
|
|
def normalize_discussion(content: str) -> dict[str, Any] | None:
|
|
"""
|
|
Agent 1: Normalize natural language discussion to structured format.
|
|
|
|
Extracts votes, questions, decisions, action items in consistent format.
|
|
"""
|
|
prompt = """You are a discussion normalizer. Extract structured information from the discussion content.
|
|
|
|
Return a JSON object with these fields:
|
|
{
|
|
"votes": [
|
|
{"participant": "Name", "vote": "READY|CHANGES|REJECT", "line": "original line text"}
|
|
],
|
|
"questions": [
|
|
{"participant": "Name", "question": "text", "line": "original line"}
|
|
],
|
|
"decisions": [
|
|
{"participant": "Name", "decision": "text", "rationale": "why", "line": "original line"}
|
|
],
|
|
"action_items": [
|
|
{"participant": "Name", "action": "text", "status": "TODO|ASSIGNED|DONE", "assignee": "Name or null", "line": "original line"}
|
|
],
|
|
"mentions": [
|
|
{"from": "Name", "to": "Name or @all", "context": "surrounding text"}
|
|
]
|
|
}
|
|
|
|
Look for patterns like:
|
|
- VOTE: READY/CHANGES/REJECT
|
|
- Q: or Question: or ending with ?
|
|
- DECISION: or "we decided to..."
|
|
- TODO: or ACTION: or "we need to..."
|
|
- DONE: or "completed..."
|
|
- @Name or @all for mentions
|
|
|
|
Only include items that are clearly present. Return empty arrays for missing categories."""
|
|
|
|
return call_ai(prompt, content)
|
|
|
|
|
|
def track_questions(content: str, existing_questions: list[dict[str, Any]]) -> dict[str, Any] | None:
|
|
"""
|
|
Agent 2: Track questions and their answers.
|
|
|
|
Identifies new questions and matches answers to existing questions.
|
|
"""
|
|
existing_q_json = json.dumps(existing_questions, indent=2)
|
|
|
|
prompt = f"""You are a question tracking agent. Process new discussion content and track questions.
|
|
|
|
Existing tracked questions:
|
|
{existing_q_json}
|
|
|
|
For the new content:
|
|
1. Identify new questions
|
|
2. Match answers to existing questions (by context/participant names)
|
|
3. Mark questions as: OPEN, PARTIAL (some answers), or ANSWERED
|
|
|
|
Return JSON:
|
|
{{
|
|
"new_questions": [
|
|
{{"participant": "Name", "question": "text", "status": "OPEN"}}
|
|
],
|
|
"updated_questions": [
|
|
{{"question_id": "index in existing list", "status": "OPEN|PARTIAL|ANSWERED", "answer": "text if answered", "answered_by": "Name"}}
|
|
]
|
|
}}
|
|
|
|
Look for:
|
|
- Lines ending with ?
|
|
- Q: or Question: markers
|
|
- A: or Answer: or Re: markers (these answer previous questions)
|
|
- @Name mentions (directing questions to someone)"""
|
|
|
|
return call_ai(prompt, content)
|
|
|
|
|
|
def track_action_items(content: str, existing_items: list[dict[str, Any]]) -> dict[str, Any] | None:
|
|
"""
|
|
Agent 3: Track action items through their lifecycle.
|
|
|
|
Tracks: TODO → ASSIGNED → DONE
|
|
"""
|
|
existing_items_json = json.dumps(existing_items, indent=2)
|
|
|
|
prompt = f"""You are an action item tracking agent. Process new discussion content and track action items.
|
|
|
|
Existing tracked items:
|
|
{existing_items_json}
|
|
|
|
For the new content:
|
|
1. Identify new action items (TODO:, ACTION:, "we need to...", "should...")
|
|
2. Track assignments (@Name you handle X, "I'll do Y")
|
|
3. Track completions (DONE:, "completed", "finished")
|
|
|
|
Return JSON:
|
|
{{
|
|
"new_items": [
|
|
{{"participant": "Name", "action": "text", "status": "TODO", "assignee": null}}
|
|
],
|
|
"updated_items": [
|
|
{{"item_id": "index in existing list", "status": "TODO|ASSIGNED|DONE", "assignee": "Name or null", "completed_by": "Name or null"}}
|
|
]
|
|
}}
|
|
|
|
Status transitions:
|
|
- TODO: Just identified, no assignee
|
|
- ASSIGNED: Someone took ownership (@Name: I'll handle X)
|
|
- DONE: Marked complete (DONE:, "completed this")"""
|
|
|
|
return call_ai(prompt, content)
|
|
|
|
|
|
def track_decisions(content: str, existing_decisions: list[dict[str, Any]]) -> dict[str, Any] | None:
|
|
"""
|
|
Agent 4: Track architectural decisions (ADR-style).
|
|
|
|
Extracts decisions with rationale, tracks reversals.
|
|
"""
|
|
existing_decisions_json = json.dumps(existing_decisions, indent=2)
|
|
|
|
prompt = f"""You are a decision logging agent. Extract architectural and design decisions from discussions.
|
|
|
|
Existing decisions:
|
|
{existing_decisions_json}
|
|
|
|
For the new content, identify decisions with:
|
|
- What was decided
|
|
- Who made/supported the decision
|
|
- Rationale (why)
|
|
- Any alternatives considered
|
|
- Whether this supersedes/reverses a previous decision
|
|
|
|
Return JSON:
|
|
{{
|
|
"new_decisions": [
|
|
{{
|
|
"participant": "Name",
|
|
"decision": "We will use X",
|
|
"rationale": "Because Y",
|
|
"alternatives": ["Option A", "Option B"],
|
|
"supporters": ["Name1", "Name2"]
|
|
}}
|
|
],
|
|
"updated_decisions": [
|
|
{{
|
|
"decision_id": "index in existing list",
|
|
"status": "ACTIVE|SUPERSEDED|REVERSED",
|
|
"superseded_by": "new decision text"
|
|
}}
|
|
]
|
|
}}
|
|
|
|
Look for:
|
|
- DECISION: markers
|
|
- "We decided to..."
|
|
- "Going with X because..."
|
|
- Vote consensus (multiple READY votes on a proposal)
|
|
- "Actually, let's change to..." (reversals)"""
|
|
|
|
return call_ai(prompt, content)
|
|
|
|
|
|
def extract_mentions(content: str) -> list[dict[str, str]]:
|
|
"""
|
|
Extract @mentions from content (does not require Claude).
|
|
|
|
Returns list of mentions with context.
|
|
"""
|
|
import re
|
|
|
|
mentions = []
|
|
lines = content.splitlines()
|
|
|
|
for line in lines:
|
|
# Find all @mentions in the line
|
|
pattern = r'@(\w+|all)'
|
|
matches = re.finditer(pattern, line)
|
|
|
|
for match in matches:
|
|
mentioned = match.group(1)
|
|
|
|
# Extract participant name if line starts with "- Name:"
|
|
from_participant = None
|
|
stripped = line.strip()
|
|
if stripped.startswith('-') or stripped.startswith('*'):
|
|
parts = stripped[1:].split(':', 1)
|
|
if len(parts) == 2:
|
|
from_participant = parts[0].strip()
|
|
|
|
mentions.append({
|
|
"from": from_participant or "unknown",
|
|
"to": mentioned,
|
|
"context": line.strip()
|
|
})
|
|
|
|
return mentions
|