Stream C: Multi-provider AI coaching backend + 20-provider registry
AI coaching service (backend/app/services/ai_provider.py): - Two code paths: OpenAI-compatible (6 providers) + Anthropic (Claude) - Streaming SSE responses via httpx async - System prompt built from AGENT_GUIDE + live user context - Graceful error handling for all provider failures - Gemini adapter for Google's generateContent API Chat endpoint (backend/app/routers/ai_chat.py): - POST /api/ai/chat — SSE streaming response - GET /api/ai/status — check configured provider - History capped at 20 messages, content at 4K chars Provider registry expanded to 20 providers: - 9 device integrations (strava, polar, garmin, whoop, oura, coros, fitbit, withings, suunto) - 8 AI providers (openai, openrouter, huggingface, groq, ollama, claude, gemini, custom_ai) - 3 other (usda, nutritionix, telegram) - Categorized: device, ai_provider, nutrition, notifications - COROS: MCP bridge approach (no API integration needed) - Strava: AI/ML warning per their ToS Frontend ChatCoach.jsx to follow in next commit.
This commit is contained in:
Claude
parent
6e082ce58f
commit
9a8e6ce1eb
4 changed files with 555 additions and 27 deletions
316
backend/app/services/ai_provider.py
Normal file
316
backend/app/services/ai_provider.py
Normal file
|
|
@ -0,0 +1,316 @@
|
|||
"""Multi-provider AI coaching service.
|
||||
|
||||
Two code paths:
|
||||
- OpenAI-compatible: covers OpenAI, OpenRouter, HuggingFace, Groq, Ollama, Custom
|
||||
- Anthropic: covers Claude (different message format + auth header)
|
||||
- Gemini: thin adapter for Google's generateContent API
|
||||
|
||||
System prompt is built from AGENT_GUIDE.md + live user context.
|
||||
Chat history is ephemeral (not persisted).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import AsyncGenerator
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.config import get_settings
|
||||
from app.services.crypto import decrypt_value
|
||||
from app.models.integration_config import IntegrationConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
settings = get_settings()
|
||||
|
||||
# AI provider presets — base_url and api_format for each named provider
|
||||
AI_PRESETS = {
|
||||
"openai": {"base_url": "https://api.openai.com/v1", "format": "openai", "default_model": "gpt-4o-mini"},
|
||||
"openrouter": {"base_url": "https://openrouter.ai/api/v1", "format": "openai", "default_model": "openai/gpt-4o-mini"},
|
||||
"huggingface": {"base_url": "https://router.huggingface.co/v1", "format": "openai", "default_model": "meta-llama/Llama-3.3-70B-Instruct"},
|
||||
"groq": {"base_url": "https://api.groq.com/openai/v1", "format": "openai", "default_model": "llama-3.3-70b-versatile"},
|
||||
"ollama": {"base_url": "http://host.docker.internal:11434/v1", "format": "openai", "default_model": "llama3.1"},
|
||||
"claude": {"base_url": "https://api.anthropic.com/v1", "format": "anthropic", "default_model": "claude-sonnet-4-6"},
|
||||
"gemini": {"base_url": "https://generativelanguage.googleapis.com/v1beta", "format": "gemini", "default_model": "gemini-2.0-flash"},
|
||||
"custom_ai": {"base_url": "", "format": "openai", "default_model": ""},
|
||||
}
|
||||
|
||||
# System prompt template — {context} is replaced with live user data
|
||||
SYSTEM_PROMPT_TEMPLATE = """You are a fitness coach integrated with Diligence, a self-hosted fitness rewards platform.
|
||||
|
||||
Your role: motivate, guide, and help the user stay consistent with their health goals. Use the context below to personalize your responses.
|
||||
|
||||
CURRENT USER CONTEXT:
|
||||
{context}
|
||||
|
||||
RULES:
|
||||
- Be encouraging but honest. Celebrate progress, address setbacks constructively.
|
||||
- When the user asks to log a workout or food, confirm what you'll log and do it.
|
||||
- Reference their program schedule when suggesting workouts.
|
||||
- Respect their motivation type (from BREQ-2 profiling) to calibrate your tone.
|
||||
- Keep responses concise — this is a mobile chat, not an essay.
|
||||
- If the user hasn't earned enough points today, gently encourage activity.
|
||||
- Never fabricate data — only reference what's in the context."""
|
||||
|
||||
|
||||
async def get_active_ai_provider(db: AsyncSession) -> dict | None:
|
||||
"""Find the first configured AI provider."""
|
||||
result = await db.execute(
|
||||
select(IntegrationConfig).where(
|
||||
IntegrationConfig.provider.in_(list(AI_PRESETS.keys()))
|
||||
)
|
||||
)
|
||||
configs = result.scalars().all()
|
||||
|
||||
for config in configs:
|
||||
provider_name = config.provider
|
||||
try:
|
||||
creds = json.loads(decrypt_value(config.encrypted_value, settings.secret_key))
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
preset = AI_PRESETS.get(provider_name, AI_PRESETS["custom_ai"])
|
||||
api_key = creds.get("api_key", "")
|
||||
base_url = creds.get("endpoint_url", preset["base_url"]) or preset["base_url"]
|
||||
model = creds.get("model", preset["default_model"]) or preset["default_model"]
|
||||
|
||||
return {
|
||||
"name": provider_name,
|
||||
"format": preset["format"],
|
||||
"base_url": base_url,
|
||||
"api_key": api_key,
|
||||
"model": model,
|
||||
}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
async def build_context(db: AsyncSession, user_id) -> str:
|
||||
"""Build live context string from user's current state."""
|
||||
from app.models.user import User
|
||||
from app.models.profile import UserProfile
|
||||
|
||||
context_parts = []
|
||||
|
||||
# User profile
|
||||
user = (await db.execute(select(User).where(User.id == user_id))).scalar_one_or_none()
|
||||
if user:
|
||||
context_parts.append(f"Name: {user.display_name}")
|
||||
context_parts.append(f"Timezone: {user.timezone}")
|
||||
|
||||
# Profile / motivation
|
||||
profile = (await db.execute(select(UserProfile).where(UserProfile.user_id == user_id))).scalar_one_or_none()
|
||||
if profile:
|
||||
if profile.ttm_stage:
|
||||
context_parts.append(f"Stage of change: {profile.ttm_stage}")
|
||||
if profile.breq_profile:
|
||||
context_parts.append(f"Motivation type: {profile.breq_profile}")
|
||||
|
||||
# Today's points (best effort)
|
||||
try:
|
||||
from app.services.points_engine import get_daily_summary
|
||||
summary = await get_daily_summary(db, user_id)
|
||||
context_parts.append(f"Today's points: {summary.get('earned', 0)}/{summary.get('target', 100)}")
|
||||
context_parts.append(f"Daily gate: {'PASSED' if summary.get('gate_passed') else 'NOT YET'}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return "\n".join(context_parts) if context_parts else "No profile data available yet."
|
||||
|
||||
|
||||
async def chat(
|
||||
user_id,
|
||||
message: str,
|
||||
history: list[dict],
|
||||
db: AsyncSession,
|
||||
) -> AsyncGenerator[str, None]:
|
||||
"""Route chat to the configured AI provider with streaming."""
|
||||
provider = await get_active_ai_provider(db)
|
||||
if not provider:
|
||||
yield "No AI provider configured. Go to Settings → Integrations to connect one (OpenAI, OpenRouter, Ollama, etc.)."
|
||||
return
|
||||
|
||||
context = await build_context(db, user_id)
|
||||
system_prompt = SYSTEM_PROMPT_TEMPLATE.format(context=context)
|
||||
|
||||
try:
|
||||
if provider["format"] == "openai":
|
||||
async for chunk in _call_openai_compat(
|
||||
base_url=provider["base_url"],
|
||||
api_key=provider["api_key"],
|
||||
model=provider["model"],
|
||||
system_prompt=system_prompt,
|
||||
history=history,
|
||||
message=message,
|
||||
):
|
||||
yield chunk
|
||||
elif provider["format"] == "anthropic":
|
||||
async for chunk in _call_anthropic(
|
||||
api_key=provider["api_key"],
|
||||
model=provider["model"],
|
||||
system_prompt=system_prompt,
|
||||
history=history,
|
||||
message=message,
|
||||
):
|
||||
yield chunk
|
||||
elif provider["format"] == "gemini":
|
||||
async for chunk in _call_gemini(
|
||||
api_key=provider["api_key"],
|
||||
model=provider["model"],
|
||||
system_prompt=system_prompt,
|
||||
history=history,
|
||||
message=message,
|
||||
):
|
||||
yield chunk
|
||||
except httpx.HTTPStatusError as e:
|
||||
yield f"AI provider error ({e.response.status_code}). Check your API key in Settings → Integrations."
|
||||
except httpx.ConnectError:
|
||||
yield f"Cannot reach {provider['name']}. Check your network or endpoint URL."
|
||||
except Exception as e:
|
||||
logger.error(f"AI chat error: {e}")
|
||||
yield "Something went wrong with the AI provider. Check Settings → Integrations."
|
||||
|
||||
|
||||
async def _call_openai_compat(
|
||||
base_url: str,
|
||||
api_key: str,
|
||||
model: str,
|
||||
system_prompt: str,
|
||||
history: list[dict],
|
||||
message: str,
|
||||
) -> AsyncGenerator[str, None]:
|
||||
"""Call any OpenAI-compatible /v1/chat/completions endpoint with streaming.
|
||||
|
||||
Covers: OpenAI, OpenRouter, HuggingFace, Groq, Ollama, Custom.
|
||||
"""
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if api_key:
|
||||
headers["Authorization"] = f"Bearer {api_key}"
|
||||
|
||||
payload = {
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "system", "content": system_prompt},
|
||||
*history,
|
||||
{"role": "user", "content": message},
|
||||
],
|
||||
"stream": True,
|
||||
"max_tokens": 1024,
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||||
async with client.stream(
|
||||
"POST",
|
||||
f"{base_url.rstrip('/')}/chat/completions",
|
||||
json=payload,
|
||||
headers=headers,
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if line.startswith("data: ") and line.strip() != "data: [DONE]":
|
||||
try:
|
||||
chunk = json.loads(line[6:])
|
||||
delta = chunk.get("choices", [{}])[0].get("delta", {})
|
||||
if content := delta.get("content"):
|
||||
yield content
|
||||
except (json.JSONDecodeError, IndexError, KeyError):
|
||||
continue
|
||||
|
||||
|
||||
async def _call_anthropic(
|
||||
api_key: str,
|
||||
model: str,
|
||||
system_prompt: str,
|
||||
history: list[dict],
|
||||
message: str,
|
||||
) -> AsyncGenerator[str, None]:
|
||||
"""Call Anthropic's /v1/messages endpoint with streaming."""
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
"x-api-key": api_key,
|
||||
"anthropic-version": "2023-06-01",
|
||||
}
|
||||
|
||||
# Convert OpenAI-style history to Anthropic format
|
||||
messages = []
|
||||
for msg in history:
|
||||
messages.append({"role": msg["role"], "content": msg["content"]})
|
||||
messages.append({"role": "user", "content": message})
|
||||
|
||||
payload = {
|
||||
"model": model,
|
||||
"system": system_prompt,
|
||||
"messages": messages,
|
||||
"max_tokens": 1024,
|
||||
"stream": True,
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||||
async with client.stream(
|
||||
"POST",
|
||||
"https://api.anthropic.com/v1/messages",
|
||||
json=payload,
|
||||
headers=headers,
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if line.startswith("data: "):
|
||||
try:
|
||||
event = json.loads(line[6:])
|
||||
if event.get("type") == "content_block_delta":
|
||||
delta = event.get("delta", {})
|
||||
if text := delta.get("text"):
|
||||
yield text
|
||||
except (json.JSONDecodeError, KeyError):
|
||||
continue
|
||||
|
||||
|
||||
async def _call_gemini(
|
||||
api_key: str,
|
||||
model: str,
|
||||
system_prompt: str,
|
||||
history: list[dict],
|
||||
message: str,
|
||||
) -> AsyncGenerator[str, None]:
|
||||
"""Call Google Gemini's generateContent endpoint with streaming."""
|
||||
url = f"https://generativelanguage.googleapis.com/v1beta/models/{model}:streamGenerateContent"
|
||||
|
||||
# Build Gemini content format
|
||||
contents = []
|
||||
# System instruction goes as first user/model exchange
|
||||
if system_prompt:
|
||||
contents.append({"role": "user", "parts": [{"text": f"[System instructions]\n{system_prompt}"}]})
|
||||
contents.append({"role": "model", "parts": [{"text": "Understood. I'll follow these instructions."}]})
|
||||
|
||||
for msg in history:
|
||||
role = "model" if msg["role"] == "assistant" else "user"
|
||||
contents.append({"role": role, "parts": [{"text": msg["content"]}]})
|
||||
contents.append({"role": "user", "parts": [{"text": message}]})
|
||||
|
||||
payload = {
|
||||
"contents": contents,
|
||||
"generationConfig": {"maxOutputTokens": 1024},
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||||
async with client.stream(
|
||||
"POST",
|
||||
url,
|
||||
json=payload,
|
||||
params={"key": api_key, "alt": "sse"},
|
||||
) as resp:
|
||||
resp.raise_for_status()
|
||||
async for line in resp.aiter_lines():
|
||||
if line.startswith("data: "):
|
||||
try:
|
||||
chunk = json.loads(line[6:])
|
||||
candidates = chunk.get("candidates", [])
|
||||
if candidates:
|
||||
parts = candidates[0].get("content", {}).get("parts", [])
|
||||
for part in parts:
|
||||
if text := part.get("text"):
|
||||
yield text
|
||||
except (json.JSONDecodeError, KeyError):
|
||||
continue
|
||||
Loading…
Add table
Add a link
Reference in a new issue