254 lines
10 KiB
Python
254 lines
10 KiB
Python
"""ObsiGate AI — Multi-provider AI service for editor enhancement.
|
|
|
|
Supports: DeepSeek, OpenRouter, Google Gemini.
|
|
Configured via environment variables.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
import httpx
|
|
from typing import Optional, Literal
|
|
|
|
logger = logging.getLogger("obsigate.ai")
|
|
|
|
ProviderName = Literal["deepseek", "openrouter", "gemini"]
|
|
|
|
# Provider configurations
|
|
PROVIDERS = {
|
|
"deepseek": {
|
|
"api_key": os.getenv("DEEPSEEK_API_KEY", ""),
|
|
"base_url": "https://api.deepseek.com/v1",
|
|
"model": os.getenv("DEEPSEEK_MODEL", "deepseek-chat"),
|
|
"auth_header": "Bearer {api_key}",
|
|
},
|
|
"openrouter": {
|
|
"api_key": os.getenv("OPENROUTER_API_KEY", ""),
|
|
"base_url": "https://openrouter.ai/api/v1",
|
|
"model": os.getenv("OPENROUTER_MODEL", "openai/gpt-4o-mini"),
|
|
"auth_header": "Bearer {api_key}",
|
|
},
|
|
"gemini": {
|
|
"api_key": os.getenv("GEMINI_API_KEY", ""),
|
|
"base_url": "https://generativelanguage.googleapis.com/v1beta",
|
|
"model": os.getenv("GEMINI_MODEL", "gemini-2.0-flash"),
|
|
"auth_header": None, # Uses query param ?key=
|
|
},
|
|
}
|
|
|
|
DEFAULT_PROVIDER: ProviderName = os.getenv("AI_DEFAULT_PROVIDER", "deepseek") # type: ignore
|
|
|
|
|
|
def _get_provider_config(provider: Optional[ProviderName] = None) -> dict:
|
|
"""Get provider config, falling back to default if requested provider unavailable."""
|
|
p = provider or DEFAULT_PROVIDER
|
|
if p not in PROVIDERS:
|
|
p = DEFAULT_PROVIDER
|
|
cfg = PROVIDERS[p]
|
|
if not cfg["api_key"]:
|
|
# Try next available provider
|
|
for alt in PROVIDERS:
|
|
if PROVIDERS[alt]["api_key"]:
|
|
p = alt
|
|
cfg = PROVIDERS[alt]
|
|
break
|
|
return {"name": p, **cfg}
|
|
|
|
|
|
async def _call_deepseek_openrouter(prompt: str, system: str, provider: Optional[ProviderName] = None,
|
|
temperature: float = 0.7, max_tokens: int = 2048) -> str:
|
|
"""Call OpenAI-compatible API (DeepSeek, OpenRouter)."""
|
|
cfg = _get_provider_config(provider)
|
|
headers = {
|
|
"Authorization": cfg["auth_header"].format(api_key=cfg["api_key"]),
|
|
"Content-Type": "application/json",
|
|
}
|
|
payload = {
|
|
"model": cfg["model"],
|
|
"messages": [
|
|
{"role": "system", "content": system},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
"temperature": temperature,
|
|
"max_tokens": max_tokens,
|
|
}
|
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
|
resp = await client.post(
|
|
f"{cfg['base_url']}/chat/completions",
|
|
headers=headers,
|
|
json=payload,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return data["choices"][0]["message"]["content"].strip()
|
|
|
|
|
|
async def _call_gemini(prompt: str, system: str, temperature: float = 0.7, max_tokens: int = 2048) -> str:
|
|
"""Call Google Gemini API."""
|
|
cfg = PROVIDERS["gemini"]
|
|
url = f"{cfg['base_url']}/models/{cfg['model']}:generateContent?key={cfg['api_key']}"
|
|
payload = {
|
|
"system_instruction": {"parts": [{"text": system}]},
|
|
"contents": [{"parts": [{"text": prompt}]}],
|
|
"generationConfig": {
|
|
"temperature": temperature,
|
|
"maxOutputTokens": max_tokens,
|
|
},
|
|
}
|
|
async with httpx.AsyncClient(timeout=60.0) as client:
|
|
resp = await client.post(url, json=payload)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return data["candidates"][0]["content"]["parts"][0]["text"].strip()
|
|
|
|
|
|
async def ai_complete(prompt: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Generic AI completion. Routes to appropriate provider."""
|
|
cfg = _get_provider_config(provider)
|
|
if cfg["name"] == "gemini":
|
|
return await _call_gemini(prompt, "You are a helpful assistant.")
|
|
return await _call_deepseek_openrouter(prompt, "You are a helpful assistant.", provider)
|
|
|
|
|
|
# ── Specialized AI actions ──
|
|
|
|
SYSTEM_PROMPT = """You are an AI assistant integrated into ObsiGate, a knowledge management tool.
|
|
Your responses should be direct and concise. When editing text, return ONLY the modified text,
|
|
no explanations or markdown fences."""
|
|
|
|
|
|
async def ai_improve_writing(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Improve writing quality while preserving meaning."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Improve the following text. Fix grammar, clarity, and flow. Preserve the original language and meaning.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3,
|
|
)
|
|
|
|
|
|
async def ai_fix_spelling(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Fix spelling and grammar errors."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Fix all spelling and grammar errors in this text. Return only the corrected text.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.1,
|
|
)
|
|
|
|
|
|
async def ai_make_shorter(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Make text more concise."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Make this text shorter and more concise while preserving the key information.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3,
|
|
)
|
|
|
|
|
|
async def ai_make_longer(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Expand text with more detail."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Expand this text with more detail, examples, or explanation while keeping the same tone.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.7, max_tokens=4096,
|
|
)
|
|
|
|
|
|
async def ai_simplify(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Simplify language."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Simplify this text. Use clearer, more straightforward language. Avoid jargon.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3,
|
|
)
|
|
|
|
|
|
async def ai_change_tone(text: str, tone: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Change the tone of the text."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Rewrite this text in a {tone} tone. Preserve the original meaning.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.5,
|
|
)
|
|
|
|
|
|
async def ai_translate(text: str, target_lang: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Translate text to target language."""
|
|
# Gemini is better at translation
|
|
if provider is None and PROVIDERS["gemini"]["api_key"]:
|
|
return await _call_gemini(
|
|
f"Translate the following text to {target_lang}. Return only the translation.\n\n{text}",
|
|
"You are a professional translator. Translate accurately and naturally.",
|
|
temperature=0.1,
|
|
)
|
|
return await _call_deepseek_openrouter(
|
|
f"Translate the following text to {target_lang}. Return only the translation.\n\n{text}",
|
|
"You are a professional translator. Translate accurately and naturally.",
|
|
provider, temperature=0.1,
|
|
)
|
|
|
|
|
|
async def ai_explain(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Explain the selected text."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Explain the following text clearly and concisely:\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3,
|
|
)
|
|
|
|
|
|
async def ai_summarize(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Summarize the selected text."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Summarize the following text concisely:\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3,
|
|
)
|
|
|
|
|
|
async def ai_continue_writing(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Continue writing from the selected text."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Continue writing from where this text leaves off. Match the style and tone:\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.7, max_tokens=4096,
|
|
)
|
|
|
|
|
|
async def ai_custom_rewrite(text: str, instruction: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Rewrite text based on a custom instruction."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Rewrite the following text according to this instruction: {instruction}\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.5,
|
|
)
|
|
|
|
|
|
async def ai_convert_to_list(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Convert paragraph text to a markdown list."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Convert this text into a well-organized markdown bullet list. Extract key points.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.2,
|
|
)
|
|
|
|
|
|
async def ai_convert_to_table(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Convert text to a markdown table."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Convert this information into a markdown table. Choose appropriate columns.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.2,
|
|
)
|
|
|
|
|
|
async def ai_generate_frontmatter(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Generate YAML frontmatter for a markdown document."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Generate YAML frontmatter for this markdown document. Include: titre, tags (as list), catégorie, statut, date. Return ONLY the YAML between --- markers.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3,
|
|
)
|
|
|
|
|
|
async def ai_inline_complete(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Inline completion — suggest continuation."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Complete this text naturally. Return only the completion (just the new text, no repetition):\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3, max_tokens=512,
|
|
)
|
|
|
|
|
|
async def ai_convert_to_canvas(text: str, provider: Optional[ProviderName] = None) -> str:
|
|
"""Convert text to a Mermaid diagram or canvas representation."""
|
|
return await _call_deepseek_openrouter(
|
|
f"Convert this content into a Mermaid.js diagram if applicable, or a structured outline. Choose the best format.\n\n{text}",
|
|
SYSTEM_PROMPT, provider, temperature=0.3, max_tokens=4096,
|
|
)
|