1202 lines
42 KiB
Python
1202 lines
42 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.responses import FileResponse
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy.future import select
|
||
from typing import List
|
||
from pydantic import BaseModel
|
||
from enum import Enum
|
||
from pathlib import Path
|
||
import uuid
|
||
import asyncio
|
||
import os
|
||
|
||
from .database import get_db, engine, Base
|
||
from .models import ContextNode, Experiment, Signal, SignalFeedback
|
||
from .services.fetcher import fetch_hn_stories
|
||
from .services.analyzer import analyze_impact
|
||
from .services.llm import llm_service
|
||
|
||
app = FastAPI(title="Neural Sync API")
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"], # In production, specify allowed origins
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# Signal Priority Enum
|
||
class SignalPriority(str, Enum):
|
||
CRITICAL = "critical" # impact >= 0.9: Immediate processing + notification
|
||
HIGH = "high" # 0.7 <= impact < 0.9: Async processing + notification
|
||
MEDIUM = "medium" # 0.5 <= impact < 0.7: Async processing only
|
||
LOW = "low" # impact < 0.5: Record only, no processing
|
||
|
||
def classify_signal_priority(impact_score: float, confidence: float = 0.0) -> SignalPriority:
|
||
"""
|
||
Classify signal priority based on impact score and confidence.
|
||
|
||
Priority Matrix:
|
||
- CRITICAL: impact >= 0.9 (regardless of confidence)
|
||
- HIGH: 0.7 <= impact < 0.9 AND confidence >= 0.6
|
||
- MEDIUM: 0.5 <= impact < 0.7 OR (0.7 <= impact < 0.9 AND confidence < 0.6)
|
||
- LOW: impact < 0.5
|
||
"""
|
||
if impact_score >= 0.9:
|
||
return SignalPriority.CRITICAL
|
||
elif impact_score >= 0.7 and confidence >= 0.6:
|
||
return SignalPriority.HIGH
|
||
elif impact_score >= 0.5 or (impact_score >= 0.7 and confidence < 0.6):
|
||
return SignalPriority.MEDIUM
|
||
else:
|
||
return SignalPriority.LOW
|
||
|
||
# Pydantic Models
|
||
class SignalCreate(BaseModel):
|
||
source: str
|
||
content: str
|
||
impact_score: float
|
||
priority: SignalPriority = None # Will be auto-calculated if not provided
|
||
|
||
class ExperimentCreate(BaseModel):
|
||
title: str
|
||
hypothesis: str
|
||
target_metric: str
|
||
linked_context_nodes: List[uuid.UUID] = []
|
||
|
||
class ExperimentUpdate(BaseModel):
|
||
current_score: float
|
||
|
||
class ContextNodeCreate(BaseModel):
|
||
content: str
|
||
|
||
from .services.plane import plane_service
|
||
|
||
class TaskBreakdownRequest(BaseModel):
|
||
description: str
|
||
sync_to_plane: bool = False
|
||
|
||
class TaskItem(BaseModel):
|
||
task_name: str
|
||
type: str
|
||
estimated_hours: float
|
||
|
||
from typing import List, Optional
|
||
|
||
class TaskBreakdownResponse(BaseModel):
|
||
tasks: List[TaskItem]
|
||
sync_result: Optional[dict] = None
|
||
|
||
# Endpoints
|
||
|
||
@app.on_event("startup")
|
||
async def startup():
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
|
||
@app.get("/signals/")
|
||
async def list_signals(db: AsyncSession = Depends(get_db)):
|
||
result = await db.execute(select(Signal).order_by(Signal.impact_score.desc()))
|
||
signals = result.scalars().all()
|
||
return signals
|
||
|
||
@app.post("/signals/", response_model=dict)
|
||
async def create_signal(signal: SignalCreate, db: AsyncSession = Depends(get_db)):
|
||
new_signal = Signal(
|
||
source=signal.source,
|
||
content=signal.content,
|
||
impact_score=signal.impact_score
|
||
)
|
||
db.add(new_signal)
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
return {"id": str(new_signal.id), "status": "received"}
|
||
|
||
from fastapi import FastAPI, Depends, HTTPException, Request
|
||
|
||
# ... existing imports ...
|
||
|
||
# Pydantic Models for Webhook
|
||
class WebhookPayload(BaseModel):
|
||
# This structure depends on the Git provider (GitHub/GitLab)
|
||
# We'll support a generic structure or try to parse flexibly
|
||
commits: List[dict] = []
|
||
repository: dict = {}
|
||
ref: str = ""
|
||
|
||
# ... existing models ...
|
||
|
||
from .services.lark import LarkService
|
||
|
||
lark_service = LarkService()
|
||
|
||
@app.post("/webhooks/git")
|
||
async def handle_git_webhook(payload: dict, db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
Receives Git webhooks.
|
||
1. Extract Task ID and Code Diff.
|
||
2. AI Code Audit.
|
||
3. If Passed -> Update Plane & Signal status.
|
||
"""
|
||
import re
|
||
import json
|
||
|
||
print(f"Received Webhook Payload: {payload}")
|
||
|
||
commits = payload.get("commits", [])
|
||
if not commits:
|
||
return {"status": "ignored", "reason": "No commits found"}
|
||
|
||
updated_tasks = []
|
||
|
||
for commit in commits:
|
||
message = commit.get("message", "")
|
||
# Mocking diff retrieval - in real life, fetch from commit.url or diff_url
|
||
# Here we assume the payload might simulate a diff or we rely on message/added files
|
||
code_diff = commit.get("patch") or str(commit.get("modified", []))
|
||
|
||
# Regex to find Task IDs like #PROJ-123 or #123 or #Task-Name
|
||
# Matching task names is harder, let's stick to ID or simple name match if ID not found
|
||
task_matches = re.findall(r'#([A-Za-z0-9-]+-\d+|[A-Za-z0-9]+-\d+)', message)
|
||
|
||
for task_id in task_matches:
|
||
print(f"Processing Task ID: {task_id}")
|
||
|
||
# 1. Find the Signal that generated this task (to get task details/context)
|
||
# We search in the JSON column 'tasks_generated'
|
||
# Postgres JSON search is powerful but for MVP let's do a simple check or assume we have context.
|
||
# Actually, to Audit properly, we need the Task Description.
|
||
# Let's search specifically for this task_id in all signals.
|
||
|
||
# Optimized search: Select signals where tasks_generated contains the task_id
|
||
# For MVP with SQLite/Simple Postgres usage: Fetch recent signals and check in python
|
||
|
||
result = await db.execute(select(Signal).order_by(Signal.created_at.desc()).limit(20))
|
||
signals = result.scalars().all()
|
||
|
||
target_task = None
|
||
target_signal = None
|
||
|
||
for sig in signals:
|
||
if not sig.tasks_generated: continue
|
||
try:
|
||
tasks = json.loads(sig.tasks_generated)
|
||
for t in tasks:
|
||
# Match by Plane ID or Key
|
||
if t.get("plane_key") == task_id or t.get("plane_id") == task_id:
|
||
target_task = t
|
||
target_signal = sig
|
||
break
|
||
except:
|
||
continue
|
||
if target_task: break
|
||
|
||
task_name = target_task.get("task_name") if target_task else f"Task {task_id}"
|
||
|
||
# 2. AI Code Audit
|
||
audit_result = llm_service.audit_code_change(task_name, message, code_diff)
|
||
|
||
if audit_result.get("passed"):
|
||
print(f"✅ Audit Passed for {task_id}")
|
||
# 3. Update Plane
|
||
plane_res = await plane_service.update_issue_status(task_id, status="Done")
|
||
|
||
# 4. Update Signal DB (Mark task as Done in JSON)
|
||
if target_signal and target_task:
|
||
tasks = json.loads(target_signal.tasks_generated)
|
||
for t in tasks:
|
||
if t.get("plane_key") == task_id or t.get("plane_id") == task_id:
|
||
t["status"] = "Done"
|
||
t["audit_feedback"] = audit_result.get("reason")
|
||
# Record completion time for metrics
|
||
if "completed_at" not in t:
|
||
from datetime import datetime
|
||
t["completed_at"] = datetime.utcnow().isoformat()
|
||
|
||
target_signal.tasks_generated = json.dumps(tasks)
|
||
await db.commit()
|
||
|
||
updated_tasks.append({
|
||
"task_id": task_id,
|
||
"status": "Done",
|
||
"audit": audit_result
|
||
})
|
||
else:
|
||
print(f"❌ Audit Failed for {task_id}: {audit_result.get('reason')}")
|
||
updated_tasks.append({
|
||
"task_id": task_id,
|
||
"status": "Failed",
|
||
"audit": audit_result
|
||
})
|
||
|
||
return {"status": "processed", "updated_tasks": updated_tasks}
|
||
|
||
@app.post("/webhooks/lark")
|
||
async def handle_lark_webhook(request: Request):
|
||
"""
|
||
Receives Lark Bot Webhook events.
|
||
When a message is received:
|
||
1. Extract content.
|
||
2. Call AI Task Breakdown.
|
||
3. Sync to Plane.
|
||
4. Reply with a Card Message.
|
||
"""
|
||
try:
|
||
data = await request.json()
|
||
except Exception:
|
||
return {"status": "error", "message": "Invalid JSON"}
|
||
|
||
# Handle Challenge (Lark verification)
|
||
if "challenge" in data:
|
||
return {"challenge": data["challenge"]}
|
||
|
||
event = data.get("event", {})
|
||
if not event:
|
||
return {"status": "ignored"}
|
||
|
||
# Check if it's a message event
|
||
if event.get("type") == "message" or event.get("msg_type") == "text":
|
||
message_content = event.get("text_without_at_bot") or event.get("content", "")
|
||
# Parse JSON string content if needed, Lark sends JSON string for 'content'
|
||
if isinstance(message_content, str) and message_content.startswith("{"):
|
||
import json
|
||
try:
|
||
content_obj = json.loads(message_content)
|
||
message_content = content_obj.get("text", "")
|
||
except:
|
||
pass
|
||
|
||
message_content = message_content.strip()
|
||
chat_id = event.get("open_chat_id") or event.get("chat_id") # V1 vs V2 events
|
||
|
||
if not message_content or not chat_id:
|
||
return {"status": "ignored", "reason": "No content or chat_id"}
|
||
|
||
print(f"Received Lark Message: {message_content} from {chat_id}")
|
||
|
||
# Call Task Breakdown Engine
|
||
try:
|
||
tasks = llm_service.generate_task_breakdown(message_content)
|
||
|
||
# Sync to Plane
|
||
if tasks:
|
||
await plane_service.sync_tasks(tasks)
|
||
|
||
# Send Card Reply
|
||
# In simulation, this will just print
|
||
await lark_service.send_card_message(chat_id, message_content, tasks)
|
||
|
||
except Exception as e:
|
||
print(f"Error processing Lark message: {e}")
|
||
|
||
return {"status": "processed"}
|
||
@app.post("/signals/ingest", response_model=dict)
|
||
async def ingest_signal(signal: SignalCreate, background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
Ingests external signals (RSS/Crawler) with priority-based processing.
|
||
|
||
Processing Strategy by Priority:
|
||
- CRITICAL (impact >= 0.9): Immediate processing + notification
|
||
- HIGH (0.7 <= impact < 0.9): Async processing + notification
|
||
- MEDIUM (0.5 <= impact < 0.7): Async processing only
|
||
- LOW (impact < 0.5): Record only, no processing
|
||
"""
|
||
# 1. Calculate initial priority
|
||
initial_priority = signal.priority or classify_signal_priority(signal.impact_score)
|
||
|
||
# 2. Save Signal with priority
|
||
new_signal = Signal(
|
||
source=signal.source,
|
||
content=signal.content,
|
||
impact_score=signal.impact_score,
|
||
priority=initial_priority.value
|
||
)
|
||
db.add(new_signal)
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
|
||
# 3. Priority-based processing
|
||
if initial_priority == SignalPriority.LOW:
|
||
# LOW priority: Record only, no processing
|
||
print(f"[PRIORITY] LOW priority signal recorded: {new_signal.id}")
|
||
return {
|
||
"id": str(new_signal.id),
|
||
"priority": initial_priority.value,
|
||
"status": "recorded",
|
||
"message": "Low priority signal recorded without processing"
|
||
}
|
||
|
||
# For MEDIUM, HIGH, and CRITICAL: process the signal
|
||
# Define async processing function
|
||
async def process_signal_internal():
|
||
# RAG: Find relevant context from knowledge base
|
||
relevant_context = await llm_service.find_relevant_context(
|
||
signal_content=signal.content,
|
||
db_session=db,
|
||
top_k=3,
|
||
min_similarity=0.3
|
||
)
|
||
|
||
# Strategic Assessment with RAG context and multi-model voting
|
||
strategy_result = llm_service.evaluate_with_voting(
|
||
signal_content=signal.content,
|
||
relevant_context=relevant_context,
|
||
impact_score=signal.impact_score
|
||
)
|
||
|
||
is_strategic = strategy_result.get("is_strategic", False)
|
||
decision_proposal = strategy_result.get("decision_proposal", {})
|
||
strategy_desc = decision_proposal.get("recommendation", "")
|
||
tasks = strategy_result.get("tasks", [])
|
||
confidence = strategy_result.get("confidence", 0.0)
|
||
|
||
# Re-calculate priority with confidence
|
||
final_priority = classify_signal_priority(signal.impact_score, confidence)
|
||
new_signal.priority = final_priority.value
|
||
|
||
# Process based on priority
|
||
sync_result = None
|
||
notification_sent = False
|
||
|
||
if final_priority in [SignalPriority.CRITICAL, SignalPriority.HIGH]:
|
||
if is_strategic and tasks:
|
||
# Sync to Plane
|
||
print(f"[PRIORITY] {final_priority.value.upper()}: Strategic Signal Detected! Strategy: {strategy_desc}")
|
||
sync_result = await plane_service.sync_tasks(tasks, is_urgent=(final_priority == SignalPriority.CRITICAL))
|
||
|
||
# Lark Strategic Alert (for HIGH and CRITICAL)
|
||
default_chat_id = "oc_1234567890"
|
||
await lark_service.send_strategic_alert(default_chat_id, signal.content, decision_proposal, tasks)
|
||
notification_sent = True
|
||
|
||
# Update signal impact score to reflect strategic importance
|
||
new_signal.impact_score = 1.0
|
||
|
||
# Store strategic analysis
|
||
import json
|
||
analysis_data = {
|
||
"decision_proposal": decision_proposal,
|
||
"competitiveness_radar": strategy_result.get("competitiveness_radar"),
|
||
"detected_tech_stacks": strategy_result.get("detected_tech_stacks", []),
|
||
"related_knowledge_nodes": strategy_result.get("related_knowledge_nodes", []),
|
||
"rag_context_used": len(relevant_context) if relevant_context else 0,
|
||
"rag_context_sources": [ctx['id'] for ctx in (relevant_context or [])]
|
||
}
|
||
|
||
if "_voting" in strategy_result:
|
||
voting_info = strategy_result["_voting"]
|
||
analysis_data["voting"] = {
|
||
"enabled": voting_info.get("enabled"),
|
||
"models_used": voting_info.get("models_used"),
|
||
"strategic_votes": voting_info.get("strategic_votes"),
|
||
"total_votes": voting_info.get("total_votes"),
|
||
"unanimous": voting_info.get("unanimous")
|
||
}
|
||
|
||
new_signal.strategic_analysis = json.dumps(analysis_data)
|
||
new_signal.tasks_generated = json.dumps(tasks)
|
||
new_signal.processed = True
|
||
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
|
||
return {
|
||
"id": str(new_signal.id),
|
||
"priority": final_priority.value,
|
||
"is_strategic": is_strategic,
|
||
"confidence": confidence,
|
||
"decision_proposal": decision_proposal,
|
||
"rag_context_used": len(relevant_context) if relevant_context else 0,
|
||
"voting": strategy_result.get("_voting"),
|
||
"tasks_generated": len(tasks),
|
||
"sync_result": sync_result,
|
||
"notification_sent": notification_sent
|
||
}
|
||
|
||
# Process based on priority
|
||
if initial_priority == SignalPriority.CRITICAL:
|
||
# CRITICAL: Process immediately
|
||
result = await process_signal_internal()
|
||
result["processing_mode"] = "synchronous"
|
||
return result
|
||
else:
|
||
# MEDIUM and HIGH: Process in background
|
||
print(f"[PRIORITY] {initial_priority.value.upper()}: Signal queued for background processing: {new_signal.id}")
|
||
background_tasks.add_task(process_signal_internal)
|
||
return {
|
||
"id": str(new_signal.id),
|
||
"priority": initial_priority.value,
|
||
"status": "queued",
|
||
"processing_mode": "background",
|
||
"message": f"{initial_priority.value.upper()} priority signal queued for processing"
|
||
}
|
||
|
||
@app.post("/signals/fetch", response_model=dict)
|
||
async def trigger_signal_fetch(background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
Manually trigger the AI Agent to fetch and analyze signals from the web.
|
||
This now includes full AI strategic analysis for each signal.
|
||
"""
|
||
# 1. Fetch raw stories
|
||
raw_stories = await fetch_hn_stories(limit=30)
|
||
|
||
new_signals_count = 0
|
||
processed_count = 0
|
||
|
||
for story in raw_stories:
|
||
# 2. Check for duplicates (Simple check by content)
|
||
result = await db.execute(select(Signal).where(Signal.content == story['content']))
|
||
if result.scalars().first():
|
||
print(f"Duplicate signal skipped: {story['content'][:30]}...")
|
||
continue
|
||
|
||
# 3. Analyze Impact (keyword-based)
|
||
impact = analyze_impact(story['content'])
|
||
|
||
# 4. Create signal and trigger full AI analysis
|
||
if impact > 0.4: # Only save relevant stuff
|
||
# Use the same logic as /signals/ingest
|
||
initial_priority = classify_signal_priority(impact)
|
||
|
||
new_signal = Signal(
|
||
source=story['source'],
|
||
content=story['content'],
|
||
impact_score=impact,
|
||
priority=initial_priority.value
|
||
)
|
||
db.add(new_signal)
|
||
await db.commit()
|
||
await db.refresh(new_signal)
|
||
new_signals_count += 1
|
||
|
||
# Skip AI analysis for LOW priority
|
||
if initial_priority == SignalPriority.LOW:
|
||
print(f"[FETCH] LOW priority signal saved without analysis: {new_signal.id}")
|
||
continue
|
||
|
||
# Queue for AI analysis (same logic as ingest_signal)
|
||
async def process_fetched_signal():
|
||
# RAG: Find relevant context from knowledge base
|
||
relevant_context = await llm_service.find_relevant_context(
|
||
signal_content=story['content'],
|
||
db_session=db,
|
||
top_k=3,
|
||
min_similarity=0.3
|
||
)
|
||
|
||
# Strategic Assessment with RAG context
|
||
strategy_result = llm_service.evaluate_with_voting(
|
||
signal_content=story['content'],
|
||
relevant_context=relevant_context,
|
||
impact_score=impact
|
||
)
|
||
|
||
is_strategic = strategy_result.get("is_strategic", False)
|
||
decision_proposal = strategy_result.get("decision_proposal", {})
|
||
confidence = strategy_result.get("confidence", 0.0)
|
||
|
||
# Re-calculate priority with confidence
|
||
final_priority = classify_signal_priority(impact, confidence)
|
||
new_signal.priority = final_priority.value
|
||
|
||
# Store strategic analysis
|
||
import json
|
||
analysis_data = {
|
||
"decision_proposal": decision_proposal,
|
||
"competitiveness_radar": strategy_result.get("competitiveness_radar"),
|
||
"detected_tech_stacks": strategy_result.get("detected_tech_stacks", []),
|
||
"related_knowledge_nodes": strategy_result.get("related_knowledge_nodes", []),
|
||
"rag_context_used": len(relevant_context) if relevant_context else 0,
|
||
"rag_context_sources": [ctx['id'] for ctx in (relevant_context or [])],
|
||
"confidence": confidence,
|
||
"is_strategic": is_strategic
|
||
}
|
||
|
||
if "_voting" in strategy_result:
|
||
voting_info = strategy_result["_voting"]
|
||
analysis_data["voting"] = {
|
||
"enabled": voting_info.get("enabled"),
|
||
"models_used": voting_info.get("models_used"),
|
||
"strategic_votes": voting_info.get("strategic_votes"),
|
||
"total_votes": voting_info.get("total_votes"),
|
||
"unanimous": voting_info.get("unanimous")
|
||
}
|
||
|
||
new_signal.strategic_analysis = json.dumps(analysis_data)
|
||
tasks = strategy_result.get("tasks", [])
|
||
new_signal.tasks_generated = json.dumps(tasks)
|
||
new_signal.processed = True
|
||
|
||
await db.commit()
|
||
nonlocal processed_count
|
||
processed_count += 1
|
||
print(f"[FETCH] Signal analyzed: {new_signal.id}, strategic={is_strategic}")
|
||
|
||
# Process in background for MEDIUM/HIGH priority
|
||
if initial_priority in [SignalPriority.MEDIUM, SignalPriority.HIGH]:
|
||
background_tasks.add_task(process_fetched_signal)
|
||
else:
|
||
# CRITICAL: process immediately
|
||
await process_fetched_signal()
|
||
|
||
# Auto-stop Ollama after all analysis is complete to save resources
|
||
_stop_ollama()
|
||
|
||
return {
|
||
"status": "success",
|
||
"new_signals_count": new_signals_count,
|
||
"processed_count": processed_count,
|
||
"message": f"Fetched {new_signals_count} signals, {processed_count} analyzed"
|
||
}
|
||
|
||
def _stop_ollama():
|
||
"""Stop Ollama service to save resources."""
|
||
import subprocess
|
||
import signal
|
||
import os
|
||
|
||
killed = False
|
||
|
||
# Method 1: Try to find and kill ollama serve processes
|
||
try:
|
||
result = subprocess.run(
|
||
["pgrep", "-f", "ollama"],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
if result.returncode == 0:
|
||
pids = result.stdout.strip().split('\n')
|
||
for pid in pids:
|
||
if pid:
|
||
try:
|
||
os.kill(int(pid), signal.SIGTERM)
|
||
killed = True
|
||
except ProcessLookupError:
|
||
pass
|
||
except Exception as e:
|
||
print(f"[AI] pgrep failed: {e}")
|
||
|
||
# Method 2: Fallback to pkill -9 (force kill)
|
||
if not killed:
|
||
try:
|
||
result = subprocess.run(
|
||
["pkill", "-9", "-f", "ollama"],
|
||
capture_output=True,
|
||
timeout=5
|
||
)
|
||
killed = result.returncode == 0
|
||
except Exception as e:
|
||
print(f"[AI] pkill failed: {e}")
|
||
|
||
if killed:
|
||
print("[AI] Ollama stopped - analysis complete")
|
||
|
||
@app.get("/experiments/")
|
||
async def list_experiments(db: AsyncSession = Depends(get_db)):
|
||
result = await db.execute(select(Experiment))
|
||
experiments = result.scalars().all()
|
||
return experiments
|
||
|
||
@app.post("/experiments/")
|
||
async def create_experiment(experiment: ExperimentCreate, db: AsyncSession = Depends(get_db)):
|
||
new_experiment = Experiment(
|
||
title=experiment.title,
|
||
hypothesis=experiment.hypothesis,
|
||
target_metric=experiment.target_metric,
|
||
linked_context_nodes=experiment.linked_context_nodes,
|
||
status="running" # Auto-start for now
|
||
)
|
||
db.add(new_experiment)
|
||
await db.commit()
|
||
await db.refresh(new_experiment)
|
||
return new_experiment
|
||
|
||
@app.patch("/experiments/{experiment_id}")
|
||
async def update_experiment_score(experiment_id: uuid.UUID, update: ExperimentUpdate, db: AsyncSession = Depends(get_db)):
|
||
result = await db.execute(select(Experiment).where(Experiment.id == experiment_id))
|
||
experiment = result.scalars().first()
|
||
if not experiment:
|
||
raise HTTPException(status_code=404, detail="Experiment not found")
|
||
|
||
experiment.current_score = update.current_score
|
||
|
||
# Auto-update status based on score (Simple Logic)
|
||
# In a real scenario, this threshold might be dynamic or configured per experiment
|
||
if update.current_score >= 0.9:
|
||
experiment.status = "success"
|
||
elif update.current_score < 0.5:
|
||
experiment.status = "failed"
|
||
else:
|
||
experiment.status = "running"
|
||
|
||
await db.commit()
|
||
await db.refresh(experiment)
|
||
return experiment
|
||
|
||
@app.get("/context_nodes/")
|
||
async def list_context_nodes(db: AsyncSession = Depends(get_db)):
|
||
result = await db.execute(select(ContextNode))
|
||
nodes = result.scalars().all()
|
||
# Simplified serialization to avoid Vector issues
|
||
return [{
|
||
"id": node.id,
|
||
"content": node.content,
|
||
"status": node.status,
|
||
"version": node.version,
|
||
"has_embedding": node.embedding is not None,
|
||
"ai_insights": node.ai_insights
|
||
} for node in nodes]
|
||
|
||
@app.post("/context_nodes/")
|
||
async def create_context_node(node: ContextNodeCreate, db: AsyncSession = Depends(get_db)):
|
||
# Generate Embedding using Ollama
|
||
embedding = llm_service.get_embedding(node.content)
|
||
|
||
# AI Insight Extraction
|
||
insights = llm_service.extract_context_insights(node.content)
|
||
|
||
new_node = ContextNode(
|
||
content=node.content,
|
||
embedding=embedding if embedding else None,
|
||
ai_insights=insights
|
||
)
|
||
db.add(new_node)
|
||
await db.commit()
|
||
await db.refresh(new_node)
|
||
|
||
# Return a simplified response to avoid serialization issues with Vector type
|
||
return {
|
||
"id": new_node.id,
|
||
"content": new_node.content,
|
||
"status": new_node.status,
|
||
"version": new_node.version,
|
||
"has_embedding": new_node.embedding is not None,
|
||
"ai_insights": new_node.ai_insights
|
||
}
|
||
|
||
@app.post("/tasks/breakdown", response_model=TaskBreakdownResponse)
|
||
async def breakdown_tasks(request: TaskBreakdownRequest):
|
||
"""
|
||
AI Task Breakdown Engine.
|
||
Receives a natural language description and returns a list of technical tasks.
|
||
If sync_to_plane=True, it will create issues in Plane (simulated if no API key).
|
||
"""
|
||
tasks = llm_service.generate_task_breakdown(request.description)
|
||
if not tasks:
|
||
raise HTTPException(status_code=500, detail="Failed to generate task breakdown")
|
||
|
||
sync_result = None
|
||
if request.sync_to_plane:
|
||
sync_result = await plane_service.sync_tasks(tasks)
|
||
|
||
return {"tasks": tasks, "sync_result": sync_result}
|
||
|
||
# Feedback endpoints
|
||
class SignalFeedbackCreate(BaseModel):
|
||
signal_id: uuid.UUID
|
||
was_strategic: bool # User's actual judgment
|
||
reason: str = "" # Optional explanation
|
||
user_id: str = "" # Optional user identifier
|
||
|
||
class SignalFeedbackResponse(BaseModel):
|
||
id: uuid.UUID
|
||
signal_id: uuid.UUID
|
||
was_strategic: bool
|
||
ai_predicted_strategic: bool
|
||
ai_confidence: float
|
||
is_correct: bool # Whether AI prediction matched user feedback
|
||
|
||
@app.post("/signals/{signal_id}/feedback", response_model=SignalFeedbackResponse)
|
||
async def submit_signal_feedback(
|
||
signal_id: uuid.UUID,
|
||
feedback: SignalFeedbackCreate,
|
||
db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""
|
||
Submit user feedback on AI strategic analysis.
|
||
This helps the system learn from corrections.
|
||
"""
|
||
# 1. Verify signal exists
|
||
result = await db.execute(select(Signal).where(Signal.id == signal_id))
|
||
signal = result.scalar_one_or_none()
|
||
|
||
if not signal:
|
||
raise HTTPException(status_code=404, detail="Signal not found")
|
||
|
||
# 2. Extract AI prediction from strategic_analysis
|
||
ai_predicted_strategic = False
|
||
ai_confidence = 0.0
|
||
|
||
if signal.strategic_analysis:
|
||
try:
|
||
import json
|
||
analysis = json.loads(signal.strategic_analysis)
|
||
decision_proposal = analysis.get("decision_proposal", {})
|
||
# Try different ways the prediction might be stored
|
||
ai_predicted_strategic = analysis.get("is_strategic", signal.impact_score >= 0.9)
|
||
ai_confidence = analysis.get("confidence", 0.0)
|
||
except:
|
||
# Fallback to impact_score
|
||
ai_predicted_strategic = signal.impact_score >= 0.9
|
||
|
||
# 3. Create feedback record
|
||
new_feedback = SignalFeedback(
|
||
signal_id=signal_id,
|
||
was_strategic=feedback.was_strategic,
|
||
reason=feedback.reason,
|
||
user_id=feedback.user_id,
|
||
ai_predicted_strategic=ai_predicted_strategic,
|
||
ai_confidence=ai_confidence
|
||
)
|
||
|
||
db.add(new_feedback)
|
||
await db.commit()
|
||
await db.refresh(new_feedback)
|
||
|
||
# 4. Calculate if AI was correct
|
||
is_correct = ai_predicted_strategic == feedback.was_strategic
|
||
|
||
# 5. Log for monitoring
|
||
print(f"[FEEDBACK] Signal {signal_id}: AI predicted {ai_predicted_strategic}, user says {feedback.was_strategic} - {'✓ Correct' if is_correct else '✗ Wrong'}")
|
||
if feedback.reason:
|
||
print(f"[FEEDBACK] Reason: {feedback.reason}")
|
||
|
||
return SignalFeedbackResponse(
|
||
id=new_feedback.id,
|
||
signal_id=new_feedback.signal_id,
|
||
was_strategic=new_feedback.was_strategic,
|
||
ai_predicted_strategic=new_feedback.ai_predicted_strategic,
|
||
ai_confidence=new_feedback.ai_confidence,
|
||
is_correct=is_correct
|
||
)
|
||
|
||
@app.get("/signals/{signal_id}/feedbacks", response_model=List[SignalFeedbackResponse])
|
||
async def get_signal_feedbacks(
|
||
signal_id: uuid.UUID,
|
||
db: AsyncSession = Depends(get_db)
|
||
):
|
||
"""
|
||
Get all feedback for a specific signal.
|
||
"""
|
||
result = await db.execute(
|
||
select(SignalFeedback)
|
||
.where(SignalFeedback.signal_id == signal_id)
|
||
.order_by(SignalFeedback.created_at.desc())
|
||
)
|
||
feedbacks = result.scalars().all()
|
||
|
||
return [
|
||
SignalFeedbackResponse(
|
||
id=f.id,
|
||
signal_id=f.signal_id,
|
||
was_strategic=f.was_strategic,
|
||
ai_predicted_strategic=f.ai_predicted_strategic,
|
||
ai_confidence=f.ai_confidence or 0.0,
|
||
is_correct=f.ai_predicted_strategic == f.was_strategic
|
||
)
|
||
for f in feedbacks
|
||
]
|
||
|
||
@app.get("/feedbacks/stats")
|
||
async def get_feedback_stats(db: AsyncSession = Depends(get_db)):
|
||
"""
|
||
Get overall feedback statistics for monitoring AI performance.
|
||
"""
|
||
result = await db.execute(select(SignalFeedback))
|
||
all_feedbacks = result.scalars().all()
|
||
|
||
total = len(all_feedbacks)
|
||
if total == 0:
|
||
return {
|
||
"total_feedbacks": 0,
|
||
"accuracy": 0.0,
|
||
"true_positives": 0,
|
||
"false_positives": 0,
|
||
"true_negatives": 0,
|
||
"false_negatives": 0
|
||
}
|
||
|
||
correct = sum(1 for f in all_feedbacks if f.ai_predicted_strategic == f.was_strategic)
|
||
true_positives = sum(1 for f in all_feedbacks if f.ai_predicted_strategic and f.was_strategic)
|
||
false_positives = sum(1 for f in all_feedbacks if f.ai_predicted_strategic and not f.was_strategic)
|
||
true_negatives = sum(1 for f in all_feedbacks if not f.ai_predicted_strategic and not f.was_strategic)
|
||
false_negatives = sum(1 for f in all_feedbacks if not f.ai_predicted_strategic and f.was_strategic)
|
||
|
||
return {
|
||
"total_feedbacks": total,
|
||
"accuracy": correct / total,
|
||
"true_positives": true_positives,
|
||
"false_positives": false_positives,
|
||
"true_negatives": true_negatives,
|
||
"false_negatives": false_negatives
|
||
}
|
||
|
||
@app.get("/ai/status")
|
||
async def get_ai_status():
|
||
"""
|
||
检查 AI 服务状态
|
||
返回当前提供商类型和运行状态
|
||
"""
|
||
import subprocess
|
||
import json
|
||
import os
|
||
|
||
provider = os.getenv("LLM_PROVIDER", "ollama").lower()
|
||
|
||
# 云端 API 始终就绪
|
||
if provider in ["deepseek", "openai", "qwen"]:
|
||
api_key = os.getenv(f"{provider.upper()}_API_KEY")
|
||
return {
|
||
"status": "ready",
|
||
"provider": provider,
|
||
"provider_type": "cloud",
|
||
"cloud_service": provider,
|
||
"api_key_configured": bool(api_key),
|
||
"service_name": provider.capitalize()
|
||
}
|
||
|
||
# 本地 Ollama 需要检查
|
||
try:
|
||
result = subprocess.run(
|
||
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}", "http://127.0.0.1:11434/api/tags"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=3
|
||
)
|
||
|
||
if result.stdout.strip() == "200":
|
||
result2 = subprocess.run(
|
||
["curl", "-s", "http://127.0.0.1:11434/api/tags"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=3
|
||
)
|
||
if result2.returncode == 0:
|
||
data = json.loads(result2.stdout)
|
||
models = data.get("models", [])
|
||
model_names = [m.get("name", "unknown") for m in models]
|
||
return {
|
||
"status": "ready",
|
||
"provider": "ollama",
|
||
"provider_type": "local",
|
||
"ollama_running": True,
|
||
"available_models": model_names,
|
||
"model_count": len(model_names),
|
||
"service_name": "Ollama"
|
||
}
|
||
except Exception as e:
|
||
print(f"[AI STATUS] Error: {e}")
|
||
|
||
return {
|
||
"status": "not_ready",
|
||
"provider": "ollama",
|
||
"provider_type": "local",
|
||
"ollama_running": False,
|
||
"available_models": [],
|
||
"model_count": 0,
|
||
"service_name": "Ollama",
|
||
"error": "Ollama 服务未运行，请点击启动"
|
||
}
|
||
|
||
@app.post("/ai/stop")
|
||
async def stop_ai_service():
|
||
"""
|
||
Manually stop Ollama service to save resources and reduce heat.
|
||
Tries multiple methods to ensure all Ollama processes are terminated.
|
||
"""
|
||
import subprocess
|
||
killed_any = False
|
||
errors = []
|
||
|
||
# Method 1: Kill ollama serve process (the actual model server)
|
||
try:
|
||
result = subprocess.run(
|
||
["pkill", "-9", "-f", "ollama serve"],
|
||
capture_output=True,
|
||
timeout=5
|
||
)
|
||
if result.returncode == 0:
|
||
print("[AI] Killed 'ollama serve' process")
|
||
killed_any = True
|
||
except Exception as e:
|
||
errors.append(f"pkill ollama serve: {e}")
|
||
|
||
# Method 2: Kill Ollama.app main process (macOS)
|
||
try:
|
||
result = subprocess.run(
|
||
["pkill", "-9", "-f", "Ollama.app"],
|
||
capture_output=True,
|
||
timeout=5
|
||
)
|
||
if result.returncode == 0:
|
||
print("[AI] Killed Ollama.app process")
|
||
killed_any = True
|
||
except Exception as e:
|
||
errors.append(f"pkill Ollama.app: {e}")
|
||
|
||
# Method 3: Fallback - try killall (macOS/Linux)
|
||
if not killed_any:
|
||
try:
|
||
result = subprocess.run(
|
||
["killall", "-9", "ollama"],
|
||
capture_output=True,
|
||
timeout=5
|
||
)
|
||
if result.returncode == 0:
|
||
print("[AI] Killed ollama via killall")
|
||
killed_any = True
|
||
except Exception as e:
|
||
errors.append(f"killall ollama: {e}")
|
||
|
||
# Method 4: Last resort - kill by port
|
||
try:
|
||
# Find process using port 11434
|
||
result = subprocess.run(
|
||
["lsof", "-ti", ":11434"],
|
||
capture_output=True,
|
||
timeout=5
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
pid = result.stdout.strip().decode()
|
||
subprocess.run(["kill", "-9", pid], capture_output=True, timeout=5)
|
||
print(f"[AI] Killed process {pid} using port 11434")
|
||
killed_any = True
|
||
except Exception as e:
|
||
# Port method is optional, don't record as error
|
||
pass
|
||
|
||
if killed_any:
|
||
return {
|
||
"status": "stopped",
|
||
"message": "Ollama service stopped successfully"
|
||
}
|
||
elif errors:
|
||
return {
|
||
"status": "error",
|
||
"message": f"Failed to stop Ollama: {'; '.join(errors)}"
|
||
}
|
||
else:
|
||
return {
|
||
"status": "not_running",
|
||
"message": "Ollama was not running"
|
||
}
|
||
|
||
|
||
@app.post("/ai/start")
|
||
async def start_ai_service():
|
||
"""
|
||
启动本地 Ollama 服务
|
||
对于云端 API (DeepSeek/OpenAI/Qwen)，返回无需启动
|
||
"""
|
||
import subprocess
|
||
import os
|
||
|
||
provider = os.getenv("LLM_PROVIDER", "ollama").lower()
|
||
|
||
# 云端 API 不需要启动
|
||
if provider in ["deepseek", "openai", "qwen"]:
|
||
return {
|
||
"status": "not_applicable",
|
||
"message": f"使用 {provider.upper()} 云端服务，无需启动本地服务",
|
||
"provider": provider,
|
||
"provider_type": "cloud"
|
||
}
|
||
|
||
# 检查 Ollama 是否已运行
|
||
try:
|
||
result = subprocess.run(
|
||
["lsof", "-ti", ":11434"],
|
||
capture_output=True,
|
||
timeout=2
|
||
)
|
||
if result.returncode == 0:
|
||
return {
|
||
"status": "already_running",
|
||
"message": "Ollama 已在运行",
|
||
"provider": "ollama",
|
||
"provider_type": "local"
|
||
}
|
||
except:
|
||
pass
|
||
|
||
# 启动 Ollama（尝试多种方式）
|
||
started = False
|
||
error_msgs = []
|
||
|
||
# 方法1: macOS Ollama.app
|
||
try:
|
||
subprocess.Popen(
|
||
["open", "-a", "Ollama"],
|
||
capture_output=True
|
||
)
|
||
started = True
|
||
except Exception as e:
|
||
error_msgs.append(f"Ollama.app: {e}")
|
||
|
||
# 方法2: 命令行 ollama serve
|
||
if not started:
|
||
try:
|
||
subprocess.Popen(
|
||
["ollama", "serve"],
|
||
capture_output=True,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL
|
||
)
|
||
started = True
|
||
except Exception as e:
|
||
error_msgs.append(f"ollama serve: {e}")
|
||
|
||
if started:
|
||
return {
|
||
"status": "starting",
|
||
"message": "正在启动 Ollama，请稍候...",
|
||
"provider": "ollama",
|
||
"provider_type": "local"
|
||
}
|
||
else:
|
||
return {
|
||
"status": "error",
|
||
"message": f"启动失败: {'; '.join(error_msgs)}",
|
||
"provider": "ollama",
|
||
"provider_type": "local"
|
||
}
|
||
|
||
|
||
# LLM 配置管理
|
||
class LLMConfig(BaseModel):
|
||
provider: str = "deepseek" # ollama, deepseek, openai, qwen
|
||
api_key: Optional[str] = None
|
||
deepseek_api_key: Optional[str] = None
|
||
openai_api_key: Optional[str] = None
|
||
qwen_api_key: Optional[str] = None
|
||
|
||
|
||
@app.get("/ai/config")
|
||
async def get_ai_config():
|
||
"""
|
||
获取当前 LLM 配置
|
||
"""
|
||
from dotenv import load_dotenv
|
||
import os
|
||
|
||
# 重新加载环境变量
|
||
env_path = Path(__file__).parent.parent / ".env"
|
||
load_dotenv(env_path)
|
||
|
||
provider = os.getenv("LLM_PROVIDER", "ollama")
|
||
|
||
return {
|
||
"provider": provider,
|
||
"ollama_host": os.getenv("OLLAMA_HOST", "http://127.0.0.1:11434"),
|
||
"has_deepseek_key": bool(os.getenv("DEEPSEEK_API_KEY")),
|
||
"has_openai_key": bool(os.getenv("OPENAI_API_KEY")),
|
||
"has_qwen_key": bool(os.getenv("QWEN_API_KEY")),
|
||
}
|
||
|
||
|
||
@app.post("/ai/config")
|
||
async def update_ai_config(config: LLMConfig):
|
||
"""
|
||
更新 LLM 配置（需要重启后端生效）
|
||
"""
|
||
from dotenv import load_dotenv
|
||
import os
|
||
|
||
env_path = Path(__file__).parent.parent / ".env"
|
||
|
||
# 读取现有 .env 文件
|
||
env_vars = {}
|
||
if env_path.exists():
|
||
load_dotenv(env_path)
|
||
# 保留现有配置
|
||
for key, value in os.environ.items():
|
||
if key.startswith(("DEEPSEEK_", "OPENAI_", "QWEN_", "OLLAMA_", "LLM_")):
|
||
env_vars[key] = value
|
||
|
||
# 更新配置
|
||
env_vars["LLM_PROVIDER"] = config.provider
|
||
|
||
# 保存 API 密钥
|
||
if config.deepseek_api_key:
|
||
env_vars["DEEPSEEK_API_KEY"] = config.deepseek_api_key
|
||
if config.openai_api_key:
|
||
env_vars["OPENAI_API_KEY"] = config.openai_api_key
|
||
if config.qwen_api_key:
|
||
env_vars["QWEN_API_KEY"] = config.qwen_api_key
|
||
|
||
# 写入 .env 文件
|
||
with open(env_path, "w") as f:
|
||
for key, value in env_vars.items():
|
||
# 只保存 LLM 相关的环境变量
|
||
if key.startswith(("DEEPSEEK_", "OPENAI_", "QWEN_", "OLLAMA_", "LLM_")):
|
||
f.write(f"{key}={value}\n")
|
||
|
||
print(f"[AI] LLM 配置已更新: provider={config.provider}")
|
||
|
||
return {
|
||
"status": "success",
|
||
"message": "配置已保存，请重启后端服务生效",
|
||
"provider": config.provider,
|
||
"requires_restart": True
|
||
}
|
||
|
||
|
||
# 前端静态文件服务
|
||
FRONTEND_DIR = Path(__file__).parent.parent.parent / "frontend" / "dist"
|
||
|
||
@app.get("/")
|
||
async def serve_frontend():
|
||
"""服务前端首页"""
|
||
index_path = FRONTEND_DIR / "index.html"
|
||
if index_path.exists():
|
||
return FileResponse(index_path)
|
||
return {"message": "Frontend not found", "frontend_dir": str(FRONTEND_DIR)}
|
||
|
||
@app.get("/{path:path}")
|
||
async def serve_frontend_paths(path: str):
|
||
"""服务前端其他路由"""
|
||
# 先尝试直接文件
|
||
file_path = FRONTEND_DIR / path
|
||
if file_path.exists() and file_path.is_file():
|
||
return FileResponse(file_path)
|
||
# 回退到 index.html (SPA 路由)
|
||
return FileResponse(FRONTEND_DIR / "index.html")
|
||
|
||
|
||
@app.post("/admin/build-frontend")
|
||
async def build_frontend(background_tasks: BackgroundTasks):
|
||
"""构建前端 (仅用于部署)"""
|
||
import subprocess
|
||
|
||
frontend_path = FRONTEND_DIR.parent
|
||
|
||
async def run_build():
|
||
try:
|
||
result = subprocess.run(
|
||
["npm", "install"],
|
||
cwd=str(frontend_path),
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=300
|
||
)
|
||
result = subprocess.run(
|
||
["npm", "run", "build"],
|
||
cwd=str(frontend_path),
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=300
|
||
)
|
||
return result.returncode == 0, result.stdout, result.stderr
|
||
except Exception as e:
|
||
return False, "", str(e)
|
||
|
||
success, stdout, stderr = await run_build()
|
||
|
||
if success:
|
||
return {"status": "success", "message": "前端构建成功"}
|
||
else:
|
||
return {"status": "error", "message": stderr or "构建失败"}
|