Agent Zero
317 lines
11 KiB
Python
317 lines
11 KiB
Python
"""ARQ background jobs for the AI Proactive plugin.
|
|
|
|
Deep analysis job runs after context-change for deeper analysis:
|
|
- Mail-thread summary generation
|
|
- Finding similar contacts via unified_search
|
|
- Analyzing open tasks
|
|
- Generating extended suggestion with more context
|
|
- Pushing additional suggestion if confidence > threshold
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import litellm
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db import create_db_session
|
|
from app.core.notifications import create_notification
|
|
from app.models.audit import AuditLog
|
|
from app.models.contact import Contact
|
|
from app.plugins.builtins.ai_proactive.models import ProactiveSuggestion, ProactiveSettings
|
|
from app.plugins.builtins.ai_proactive.services import (
|
|
_serialize_row,
|
|
generate_suggestion,
|
|
get_user_settings,
|
|
push_suggestion,
|
|
)
|
|
from app.plugins.builtins.mail.models import Mail
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
OLLAMA_API_KEY = os.environ.get('API_KEY_OLLAMA_CLOUD', '')
|
|
|
|
DEEP_SYSTEM_PROMPT = """Du bist ein proaktiver KI-Assistent für ein CRM. Du führst eine Tiefenanalyse durch.
|
|
|
|
Analysiere den erweiterten Kontext und generiere einen detaillierten Vorschlag.
|
|
Antworte mit JSON:
|
|
{
|
|
"suggestion_type": "info|warning|action|insight",
|
|
"title": "Kurzer Titel (max 200 Zeichen)",
|
|
"content": "Detaillierte Beschreibung mit konkreten Empfehlungen",
|
|
"confidence": 0.0-1.0,
|
|
"actions": [{"method": "GET|POST|PUT|DELETE", "path": "/api/v1/...", "body": {}, "description": "..."}]
|
|
}
|
|
|
|
Fokussiere auf:
|
|
- Muster in der Kommunikationshistorie
|
|
- Beziehungen zwischen Entities
|
|
- Handlungsempfehlungen mit hoher Relevanz
|
|
- Antworte nur mit gültigem JSON"""
|
|
|
|
|
|
async def deep_analysis(
|
|
ctx: dict[str, Any],
|
|
entity_type: str,
|
|
entity_id: str,
|
|
user_id: str,
|
|
tenant_id: str,
|
|
) -> None:
|
|
"""Deep Analysis Background-Job.
|
|
|
|
Enqueued after context-change for deeper analysis:
|
|
- Mail-thread summary generation
|
|
- Similar contacts via unified_search
|
|
- Open tasks analysis
|
|
- Extended suggestion generation with more context
|
|
- If confidence > threshold: save + push additional suggestion
|
|
"""
|
|
try:
|
|
eid = uuid.UUID(entity_id)
|
|
uid = uuid.UUID(user_id)
|
|
tid = uuid.UUID(tenant_id)
|
|
except (ValueError, TypeError):
|
|
logger.warning("deep_analysis: invalid UUID parameters")
|
|
return
|
|
|
|
async with create_db_session(tid) as db:
|
|
# Get settings
|
|
settings = await get_user_settings(db, tid, uid)
|
|
if not settings.enabled:
|
|
return
|
|
|
|
# Gather extended context (more mails, more activities)
|
|
extended_context: dict[str, Any] = {
|
|
"entity_type": entity_type,
|
|
"entity_id": entity_id,
|
|
"deep_analysis": True,
|
|
}
|
|
|
|
if entity_type == "contact":
|
|
# Extended: last 30 mails
|
|
mail_result = await db.execute(
|
|
select(Mail)
|
|
.where(Mail.contact_id == eid)
|
|
.where(Mail.tenant_id == tid)
|
|
.order_by(Mail.received_at.desc())
|
|
.limit(30)
|
|
)
|
|
extended_context["mails"] = [
|
|
_serialize_row(m) for m in mail_result.scalars().all()
|
|
]
|
|
|
|
# Extended: last 50 audit entries
|
|
audit_result = await db.execute(
|
|
select(AuditLog)
|
|
.where(AuditLog.entity_id == eid)
|
|
.where(AuditLog.tenant_id == tid)
|
|
.order_by(AuditLog.timestamp.desc())
|
|
.limit(50)
|
|
)
|
|
extended_context["activities"] = [
|
|
_serialize_row(a) for a in audit_result.scalars().all()
|
|
]
|
|
|
|
# Contact data
|
|
contact_result = await db.execute(
|
|
select(Contact)
|
|
.where(Contact.id == eid)
|
|
.where(Contact.tenant_id == tid)
|
|
.limit(1)
|
|
)
|
|
contact = contact_result.scalar_one_or_none()
|
|
extended_context["contact"] = _serialize_row(contact) if contact else None
|
|
|
|
elif entity_type == "company":
|
|
from app.models.company import Company, CompanyContact
|
|
|
|
comp_result = await db.execute(
|
|
select(Company)
|
|
.where(Company.id == eid)
|
|
.where(Company.tenant_id == tid)
|
|
.limit(1)
|
|
)
|
|
company = comp_result.scalar_one_or_none()
|
|
extended_context["company"] = _serialize_row(company) if company else None
|
|
|
|
# All mails for company
|
|
mail_result = await db.execute(
|
|
select(Mail)
|
|
.where(Mail.company_id == eid)
|
|
.where(Mail.tenant_id == tid)
|
|
.order_by(Mail.received_at.desc())
|
|
.limit(30)
|
|
)
|
|
extended_context["mails"] = [
|
|
_serialize_row(m) for m in mail_result.scalars().all()
|
|
]
|
|
|
|
elif entity_type == "mail":
|
|
mail_result = await db.execute(
|
|
select(Mail)
|
|
.where(Mail.id == eid)
|
|
.where(Mail.tenant_id == tid)
|
|
.limit(1)
|
|
)
|
|
mail = mail_result.scalar_one_or_none()
|
|
extended_context["mail"] = _serialize_row(mail) if mail else None
|
|
|
|
if mail and mail.thread_id:
|
|
thread_result = await db.execute(
|
|
select(Mail)
|
|
.where(Mail.thread_id == mail.thread_id)
|
|
.where(Mail.tenant_id == tid)
|
|
.order_by(Mail.received_at.asc())
|
|
.limit(50)
|
|
)
|
|
extended_context["thread"] = [
|
|
_serialize_row(m) for m in thread_result.scalars().all()
|
|
]
|
|
|
|
# Similar entities via unified_search
|
|
try:
|
|
from app.plugins.builtins.unified_search.search_engine import (
|
|
find_similar_all_types,
|
|
)
|
|
|
|
extended_context["similar"] = await find_similar_all_types(
|
|
db, entity_type, eid, tid, limit=5
|
|
)
|
|
except Exception:
|
|
extended_context["similar"] = {}
|
|
|
|
# Generate extended suggestion with deep analysis prompt
|
|
model = settings.model or "ollama/deepseek-v4-flash"
|
|
|
|
# Get API key from DB (like ai_assistant does) or fall back to env
|
|
from app.plugins.builtins.ai_proactive.services import _get_llm_api_key
|
|
api_key, api_base, provider_type = await _get_llm_api_key(db, tid)
|
|
if not api_key:
|
|
api_key = os.environ.get('API_KEY_OLLAMA_CLOUD', '') or None
|
|
|
|
# Build model string with provider prefix
|
|
model = settings.model or "ollama/deepseek-v4-flash"
|
|
if provider_type:
|
|
model_parts = model.split("/", 1)
|
|
model = f"{provider_type}/{model_parts[-1]}"
|
|
|
|
litellm_kwargs: dict[str, Any] = dict(
|
|
model=model,
|
|
messages=[
|
|
{"role": "system", "content": DEEP_SYSTEM_PROMPT},
|
|
{
|
|
"role": "user",
|
|
"content": json.dumps(
|
|
extended_context, default=str, ensure_ascii=False
|
|
),
|
|
},
|
|
],
|
|
temperature=0.3,
|
|
max_tokens=1500,
|
|
response_format={"type": "json_object"},
|
|
)
|
|
if api_key:
|
|
litellm_kwargs["api_key"] = api_key
|
|
if api_base:
|
|
litellm_kwargs["api_base"] = api_base
|
|
|
|
try:
|
|
response = await litellm.acompletion(**litellm_kwargs)
|
|
content = response.choices[0].message.content
|
|
if not content:
|
|
return
|
|
# Strip markdown code fences if present
|
|
content = content.strip()
|
|
if content.startswith("```"):
|
|
content = content.split("\n", 1)[-1] if "\n" in content else content[3:]
|
|
if content.endswith("```"):
|
|
content = content[:-3].strip()
|
|
suggestion_data = json.loads(content)
|
|
if not suggestion_data.get("title") or not suggestion_data.get("content"):
|
|
return
|
|
if not isinstance(suggestion_data.get("actions"), list):
|
|
suggestion_data["actions"] = []
|
|
confidence = suggestion_data.get("confidence", 0.5)
|
|
try:
|
|
confidence = float(confidence)
|
|
except (TypeError, ValueError):
|
|
confidence = 0.5
|
|
suggestion_data["confidence"] = max(0.0, min(1.0, confidence))
|
|
valid_types = {"info", "warning", "action", "insight"}
|
|
if suggestion_data.get("suggestion_type") not in valid_types:
|
|
suggestion_data["suggestion_type"] = "insight"
|
|
except Exception:
|
|
logger.exception("deep_analysis: LLM call failed")
|
|
return
|
|
|
|
# Confidence threshold check
|
|
if suggestion_data["confidence"] < settings.confidence_threshold:
|
|
logger.debug(
|
|
"deep_analysis: confidence %s below threshold %s",
|
|
suggestion_data["confidence"],
|
|
settings.confidence_threshold,
|
|
)
|
|
return
|
|
|
|
# Save extended suggestion
|
|
suggestion = ProactiveSuggestion(
|
|
tenant_id=tid,
|
|
user_id=uid,
|
|
entity_type=entity_type,
|
|
entity_id=eid,
|
|
suggestion_type=suggestion_data["suggestion_type"],
|
|
title=f"[Tiefenanalyse] {suggestion_data['title']}",
|
|
content=suggestion_data["content"],
|
|
confidence=suggestion_data["confidence"],
|
|
actions=suggestion_data["actions"],
|
|
context_snapshot=extended_context,
|
|
)
|
|
db.add(suggestion)
|
|
await db.flush()
|
|
|
|
suggestion_dict = {
|
|
"id": str(suggestion.id),
|
|
"entity_type": suggestion.entity_type,
|
|
"entity_id": str(suggestion.entity_id) if suggestion.entity_id else None,
|
|
"suggestion_type": suggestion.suggestion_type,
|
|
"title": suggestion.title,
|
|
"content": suggestion.content,
|
|
"confidence": suggestion.confidence,
|
|
"actions": suggestion.actions,
|
|
"created_at": suggestion.created_at.isoformat()
|
|
if suggestion.created_at
|
|
else None,
|
|
"is_dismissed": suggestion.is_dismissed,
|
|
"is_acted_upon": suggestion.is_acted_upon,
|
|
}
|
|
|
|
# Push via SSE
|
|
await push_suggestion(str(uid), suggestion_dict)
|
|
|
|
# Notification for urgent suggestions
|
|
if suggestion.suggestion_type == "warning":
|
|
try:
|
|
await create_notification(
|
|
db,
|
|
tid,
|
|
uid,
|
|
type="ai_suggestion_urgent",
|
|
title=suggestion.title,
|
|
body=suggestion.content[:200],
|
|
)
|
|
except Exception:
|
|
logger.exception("deep_analysis: notification failed")
|
|
|
|
await db.commit()
|
|
logger.info(
|
|
"deep_analysis: generated suggestion %s for %s/%s",
|
|
suggestion.id,
|
|
entity_type,
|
|
entity_id,
|
|
)
|