1472 lines
64 KiB
Python
1472 lines
64 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
import json
|
|
import logging
|
|
import re
|
|
import time
|
|
import base64
|
|
from typing import Dict, Any, List, Optional, Tuple
|
|
from modules.datamodels.datamodelChat import PromptPlaceholder, ChatDocument
|
|
from modules.services.serviceExtraction.mainServiceExtraction import ExtractionService
|
|
from modules.datamodels.datamodelAi import AiCallRequest, AiCallOptions, OperationTypeEnum, PriorityEnum, ProcessingModeEnum
|
|
from modules.datamodels.datamodelExtraction import ContentPart, DocumentIntent
|
|
from modules.datamodels.datamodelWorkflow import AiResponse, AiResponseMetadata, DocumentData
|
|
from modules.datamodels.datamodelDocument import RenderedDocument
|
|
from modules.interfaces.interfaceAiObjects import AiObjects
|
|
from modules.shared.jsonUtils import (
|
|
parseJsonWithModel
|
|
)
|
|
from .subJsonResponseHandling import JsonResponseHandler
|
|
from modules.datamodels.datamodelAi import JsonAccumulationState
|
|
from modules.services.serviceBilling.mainServiceBilling import (
|
|
getService as getBillingService,
|
|
InsufficientBalanceException,
|
|
ProviderNotAllowedException,
|
|
BillingContextError
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Rebuild the model to resolve forward references
|
|
AiCallRequest.model_rebuild()
|
|
|
|
class AiService:
|
|
"""AI service with core operations integrated."""
|
|
|
|
def __init__(self, serviceCenter=None) -> None:
|
|
"""Initialize AI service with service center access.
|
|
|
|
Args:
|
|
serviceCenter: Service center instance for accessing other services
|
|
"""
|
|
self.services = serviceCenter
|
|
# Only depend on interfaces
|
|
self.aiObjects = None # Will be initialized in create() or ensureAiObjectsInitialized()
|
|
# Submodules initialized as None - will be set in _initializeSubmodules() after aiObjects is ready
|
|
self.extractionService = None
|
|
|
|
def _initializeSubmodules(self):
|
|
"""Initialize all submodules after aiObjects is ready."""
|
|
if self.aiObjects is None:
|
|
raise RuntimeError("aiObjects must be initialized before initializing submodules")
|
|
|
|
if self.extractionService is None:
|
|
logger.info("Initializing ExtractionService...")
|
|
self.extractionService = ExtractionService(self.services)
|
|
|
|
# Initialize new submodules
|
|
from .subResponseParsing import ResponseParser
|
|
from .subDocumentIntents import DocumentIntentAnalyzer
|
|
from .subContentExtraction import ContentExtractor
|
|
from .subStructureGeneration import StructureGenerator
|
|
from .subStructureFilling import StructureFiller
|
|
from .subAiCallLooping import AiCallLooper
|
|
|
|
if not hasattr(self, 'responseParser'):
|
|
logger.info("Initializing ResponseParser...")
|
|
self.responseParser = ResponseParser(self.services)
|
|
|
|
if not hasattr(self, 'intentAnalyzer'):
|
|
logger.info("Initializing DocumentIntentAnalyzer...")
|
|
self.intentAnalyzer = DocumentIntentAnalyzer(self.services, self)
|
|
|
|
if not hasattr(self, 'contentExtractor'):
|
|
logger.info("Initializing ContentExtractor...")
|
|
self.contentExtractor = ContentExtractor(self.services, self, self.intentAnalyzer)
|
|
|
|
if not hasattr(self, 'structureGenerator'):
|
|
logger.info("Initializing StructureGenerator...")
|
|
self.structureGenerator = StructureGenerator(self.services, self)
|
|
|
|
if not hasattr(self, 'structureFiller'):
|
|
logger.info("Initializing StructureFiller...")
|
|
self.structureFiller = StructureFiller(self.services, self)
|
|
|
|
if not hasattr(self, 'aiCallLooper'):
|
|
logger.info("Initializing AiCallLooper...")
|
|
self.aiCallLooper = AiCallLooper(self.services, self, self.responseParser)
|
|
|
|
async def callAi(self, request: AiCallRequest, progressCallback=None):
|
|
"""Router: handles content parts via extractionService, text context via interface.
|
|
|
|
FAIL-SAFE BILLING at the source:
|
|
1. Pre-flight check: validates billing context is complete (RAISES if not)
|
|
2. Balance & provider check before AI call
|
|
3. billingCallback on aiObjects: records one billing transaction per model call
|
|
with exact provider + model name (set before AI call, invoked by _callWithModel)
|
|
"""
|
|
# SPEECH_TEAMS: Dedicated pipeline, bypasses standard model selection
|
|
if request.options and request.options.operationType == OperationTypeEnum.SPEECH_TEAMS:
|
|
return await self._handleSpeechTeams(request)
|
|
|
|
# FAIL-SAFE: Pre-flight billing validation (like 0 CHF credit card check)
|
|
self._preflightBillingCheck()
|
|
|
|
# Balance & provider permission checks
|
|
await self._checkBillingBeforeAiCall()
|
|
|
|
# Calculate effective allowedProviders: RBAC ∩ Workflow
|
|
effectiveProviders = self._calculateEffectiveProviders()
|
|
if effectiveProviders and request.options:
|
|
request.options = request.options.model_copy(update={'allowedProviders': effectiveProviders})
|
|
logger.debug(f"Effective allowedProviders for AI request: {effectiveProviders}")
|
|
|
|
# Set billing callback on aiObjects BEFORE the AI call
|
|
# This callback is invoked by _callWithModel() after EVERY individual model call
|
|
# For parallel content parts (e.g., 200 MB doc), each model call creates its own transaction
|
|
self.aiObjects.billingCallback = self._createBillingCallback()
|
|
|
|
try:
|
|
if hasattr(request, 'contentParts') and request.contentParts:
|
|
response = await self.extractionService.processContentPartsWithAi(
|
|
request, self.aiObjects, progressCallback
|
|
)
|
|
else:
|
|
response = await self.aiObjects.callWithTextContext(request)
|
|
finally:
|
|
# Clear callback after call completes
|
|
self.aiObjects.billingCallback = None
|
|
|
|
# Store workflow stats for analytics
|
|
self._storeAiCallStats(response, request)
|
|
|
|
return response
|
|
|
|
# =========================================================================
|
|
# SPEECH_TEAMS: Dedicated handler for Teams Meeting AI analysis
|
|
# Bypasses standard model selection. Uses a fixed fast model.
|
|
# =========================================================================
|
|
|
|
async def _handleSpeechTeams(self, request: AiCallRequest):
|
|
"""
|
|
Dedicated handler for SPEECH_TEAMS operation type.
|
|
Bypasses standard model selection and uses a fixed fast model optimized
|
|
for low-latency meeting transcript analysis.
|
|
|
|
The handler:
|
|
1. Selects a fixed fast model (no model selector)
|
|
2. Builds a specialized system prompt for meeting analysis
|
|
3. Calls the model with structured JSON output
|
|
4. Returns a SpeechTeamsResponse wrapped in AiCallResponse
|
|
|
|
Args:
|
|
request: AiCallRequest with:
|
|
- prompt: User-configured system prompt (from FeatureInstance.config.aiSystemPrompt)
|
|
- context: Accumulated transcript segments to analyze
|
|
- options.metadata: Optional dict with "botName" key
|
|
|
|
Returns:
|
|
AiCallResponse with content as JSON string (SpeechTeamsResponse format)
|
|
"""
|
|
from modules.datamodels.datamodelAi import AiCallResponse, AiModelCall, AiCallOptions, PriorityEnum
|
|
|
|
startTime = time.time()
|
|
|
|
# Billing pre-flight (SPEECH_TEAMS also needs billing)
|
|
self._preflightBillingCheck()
|
|
await self._checkBillingBeforeAiCall()
|
|
|
|
# 1. Select a fixed fast model (bypass model selector)
|
|
model = self._getSpeechTeamsModel()
|
|
if not model:
|
|
return AiCallResponse(
|
|
content=json.dumps({"shouldRespond": False, "responseText": None, "reasoning": "No suitable model available for SPEECH_TEAMS", "detectedIntent": "none"}),
|
|
modelName="error",
|
|
provider="unknown",
|
|
priceCHF=0.0,
|
|
processingTime=0.0,
|
|
bytesSent=0,
|
|
bytesReceived=0,
|
|
errorCount=1
|
|
)
|
|
|
|
# 2. Build specialized system prompt
|
|
metadata = {}
|
|
if hasattr(request.options, 'allowedProviders') and request.options.allowedProviders:
|
|
# Reuse allowedProviders field as metadata carrier if set (backward compat)
|
|
pass
|
|
botName = metadata.get("botName", "AI Assistant")
|
|
|
|
# Extract botName from context if embedded as header
|
|
contextText = request.context or ""
|
|
if contextText.startswith("BOT_NAME:"):
|
|
lines = contextText.split("\n", 1)
|
|
botName = lines[0].replace("BOT_NAME:", "").strip()
|
|
contextText = lines[1] if len(lines) > 1 else ""
|
|
|
|
userSystemPrompt = request.prompt or ""
|
|
systemPrompt = self._buildSpeechTeamsSystemPrompt(userSystemPrompt, botName)
|
|
|
|
# 3. Build messages
|
|
messages = [
|
|
{"role": "system", "content": systemPrompt},
|
|
{"role": "user", "content": contextText}
|
|
]
|
|
|
|
# 4. Call model directly (no failover loop -- single fast model)
|
|
modelCall = AiModelCall(
|
|
messages=messages,
|
|
model=model,
|
|
options=AiCallOptions(
|
|
operationType=OperationTypeEnum.SPEECH_TEAMS,
|
|
priority=PriorityEnum.SPEED,
|
|
temperature=0.3,
|
|
resultFormat="json"
|
|
)
|
|
)
|
|
|
|
# Set billing callback
|
|
self.aiObjects.billingCallback = self._createBillingCallback()
|
|
|
|
try:
|
|
inputBytes = len((systemPrompt + contextText).encode('utf-8'))
|
|
modelResponse = await model.functionCall(modelCall)
|
|
|
|
if not modelResponse.success:
|
|
raise ValueError(f"SPEECH_TEAMS model call failed: {modelResponse.error}")
|
|
|
|
content = modelResponse.content
|
|
outputBytes = len(content.encode('utf-8'))
|
|
processingTime = time.time() - startTime
|
|
priceCHF = model.calculatepriceCHF(processingTime, inputBytes, outputBytes)
|
|
|
|
response = AiCallResponse(
|
|
content=content,
|
|
modelName=model.name,
|
|
provider=model.connectorType,
|
|
priceCHF=priceCHF,
|
|
processingTime=processingTime,
|
|
bytesSent=inputBytes,
|
|
bytesReceived=outputBytes,
|
|
errorCount=0
|
|
)
|
|
|
|
# Record billing
|
|
if self.aiObjects.billingCallback:
|
|
try:
|
|
self.aiObjects.billingCallback(response)
|
|
except Exception as e:
|
|
logger.error(f"BILLING: Failed to record billing for SPEECH_TEAMS: {e}")
|
|
|
|
# Store stats
|
|
self._storeAiCallStats(response, request)
|
|
|
|
logger.info(f"SPEECH_TEAMS call completed: model={model.name}, time={processingTime:.2f}s, cost={priceCHF:.4f} CHF")
|
|
return response
|
|
|
|
except Exception as e:
|
|
processingTime = time.time() - startTime
|
|
logger.error(f"SPEECH_TEAMS call failed: {e}")
|
|
return AiCallResponse(
|
|
content=json.dumps({"shouldRespond": False, "responseText": None, "reasoning": f"Error: {str(e)}", "detectedIntent": "none"}),
|
|
modelName=model.name if model else "error",
|
|
provider=model.connectorType if model else "unknown",
|
|
priceCHF=0.0,
|
|
processingTime=processingTime,
|
|
bytesSent=0,
|
|
bytesReceived=0,
|
|
errorCount=1
|
|
)
|
|
finally:
|
|
self.aiObjects.billingCallback = None
|
|
|
|
def _getSpeechTeamsModel(self):
|
|
"""
|
|
Get the fixed fast model for SPEECH_TEAMS.
|
|
Prioritizes: GPT-4o-mini > Claude Haiku > any fast model with DATA_ANALYSE capability.
|
|
Returns the AiModel instance or None.
|
|
"""
|
|
from modules.aicore.aicoreModelRegistry import modelRegistry
|
|
|
|
availableModels = modelRegistry.getAvailableModels()
|
|
if not availableModels:
|
|
logger.error("SPEECH_TEAMS: No models available in registry")
|
|
return None
|
|
|
|
# Priority list of preferred models for SPEECH_TEAMS (fast + cheap)
|
|
_preferredModelNames = [
|
|
"gpt-4o-mini", # OpenAI: fast, cheap, good at JSON
|
|
"claude-3-5-haiku", # Anthropic: fast, cheap
|
|
"gpt-4o", # OpenAI: fallback to quality model
|
|
"claude-sonnet-4-5", # Anthropic: fallback
|
|
]
|
|
|
|
# Try preferred models in order
|
|
for preferredName in _preferredModelNames:
|
|
for model in availableModels:
|
|
if preferredName in model.name.lower() and model.functionCall and model.isAvailable:
|
|
logger.info(f"SPEECH_TEAMS: Selected preferred model '{model.name}' ({model.displayName})")
|
|
return model
|
|
|
|
# Fallback: pick fastest available model with DATA_ANALYSE capability
|
|
_dataAnalyseModels = []
|
|
for model in availableModels:
|
|
if not model.functionCall or not model.isAvailable:
|
|
continue
|
|
for opRating in model.operationTypes:
|
|
if opRating.operationType == OperationTypeEnum.DATA_ANALYSE:
|
|
_dataAnalyseModels.append((model, opRating.rating))
|
|
break
|
|
|
|
if _dataAnalyseModels:
|
|
# Sort by speed rating (descending) then cost (ascending)
|
|
_dataAnalyseModels.sort(key=lambda x: (-x[0].speedRating, x[0].costPer1kTokensInput))
|
|
bestModel = _dataAnalyseModels[0][0]
|
|
logger.info(f"SPEECH_TEAMS: Selected fallback model '{bestModel.name}' (speed={bestModel.speedRating})")
|
|
return bestModel
|
|
|
|
# Last resort: first available model
|
|
for model in availableModels:
|
|
if model.functionCall and model.isAvailable:
|
|
logger.warning(f"SPEECH_TEAMS: Using last-resort model '{model.name}'")
|
|
return model
|
|
|
|
return None
|
|
|
|
def _buildSpeechTeamsSystemPrompt(self, userSystemPrompt: str, botName: str) -> str:
|
|
"""
|
|
Build the specialized system prompt for SPEECH_TEAMS meeting analysis.
|
|
Combines a fixed base prompt with user-configurable instructions.
|
|
"""
|
|
basePrompt = f"""Du bist "{botName}", ein AI-Teilnehmer in einem Microsoft Teams Meeting.
|
|
Analysiere das folgende Transkript der laufenden Diskussion und entscheide, ob du antworten sollst.
|
|
|
|
ANTWORTE NUR wenn:
|
|
1. Du direkt angesprochen wirst (z.B. "{botName}, was denkst du?" oder "Hey {botName}")
|
|
2. Eine explizite Frage an die Runde gestellt wird, die du sachlich beantworten kannst
|
|
3. Du einen wertvollen, nicht-offensichtlichen Beitrag zur Diskussion hast
|
|
|
|
ANTWORTE NICHT wenn:
|
|
- Die Teilnehmer normal miteinander sprechen ohne dich einzubeziehen
|
|
- Die Diskussion keinen Input von dir benoetigt
|
|
- Du nur wiederholen wuerdest, was bereits gesagt wurde
|
|
- Es sich um Smalltalk oder Begruessung handelt
|
|
|
|
ANTWORT-STIL:
|
|
- Halte dich kurz und praezise (max 2-3 Saetze fuer Sprache)
|
|
- Sei professionell aber natuerlich
|
|
- Beziehe dich konkret auf das Gesagte"""
|
|
|
|
# Append user-configured instructions if provided
|
|
if userSystemPrompt and userSystemPrompt.strip():
|
|
basePrompt += f"\n\nZUSAETZLICHE ANWEISUNGEN:\n{userSystemPrompt.strip()}"
|
|
|
|
basePrompt += f"""
|
|
|
|
WICHTIG: Antworte IMMER als valides JSON in exakt diesem Format:
|
|
{{
|
|
"shouldRespond": true/false,
|
|
"responseText": "Deine Antwort hier" oder null,
|
|
"reasoning": "Kurze Begruendung deiner Entscheidung",
|
|
"detectedIntent": "addressed" | "question" | "proactive" | "none"
|
|
}}
|
|
|
|
detectedIntent-Werte:
|
|
- "addressed": {botName} wurde direkt angesprochen
|
|
- "question": Eine allgemeine Frage wurde gestellt
|
|
- "proactive": Du hast einen wertvollen proaktiven Beitrag
|
|
- "none": Kein Handlungsbedarf"""
|
|
|
|
return basePrompt
|
|
|
|
def _preflightBillingCheck(self) -> None:
|
|
"""
|
|
Pre-flight billing validation - like a 0 CHF credit card authorization check.
|
|
|
|
Validates that ALL required billing context is present and that a billing
|
|
transaction CAN be recorded. This dry-run check catches missing context
|
|
BEFORE an expensive AI call starts.
|
|
|
|
FAIL-SAFE: This method RAISES if billing context is incomplete.
|
|
An AI call without billing context MUST NOT proceed.
|
|
|
|
Raises:
|
|
BillingContextError: If billing context is incomplete or invalid
|
|
"""
|
|
if not self.services:
|
|
raise BillingContextError("No service context available - cannot bill AI call")
|
|
|
|
user = getattr(self.services, 'user', None)
|
|
if not user:
|
|
raise BillingContextError("No user context - cannot bill AI call")
|
|
|
|
mandateId = getattr(self.services, 'mandateId', None)
|
|
if not mandateId:
|
|
raise BillingContextError(
|
|
f"No mandateId in service context for user {user.id} - cannot bill AI call. "
|
|
"Every AI call MUST have a mandate context for billing."
|
|
)
|
|
|
|
# Validate billing service can be created
|
|
featureInstanceId = getattr(self.services, 'featureInstanceId', None)
|
|
featureCode = getattr(self.services, 'featureCode', None)
|
|
|
|
try:
|
|
billingService = getBillingService(user, mandateId, featureInstanceId, featureCode)
|
|
except Exception as e:
|
|
raise BillingContextError(
|
|
f"Cannot create billing service for user {user.id}, mandate {mandateId}: {e}"
|
|
)
|
|
|
|
# Dry-run: verify billing service can check balance (DB accessible)
|
|
try:
|
|
billingService.checkBalance(0.0)
|
|
except Exception as e:
|
|
raise BillingContextError(
|
|
f"Billing system not accessible for mandate {mandateId}: {e}"
|
|
)
|
|
|
|
logger.debug(
|
|
f"Pre-flight billing check PASSED: user={user.id}, mandate={mandateId}, "
|
|
f"feature={featureCode or 'none'}, instance={featureInstanceId or 'none'}"
|
|
)
|
|
|
|
async def _checkBillingBeforeAiCall(self) -> None:
|
|
"""
|
|
Check billing status before making an AI call.
|
|
|
|
FAIL-SAFE: Context validation is done in _preflightBillingCheck() which is
|
|
called first. This method handles balance and provider permission checks.
|
|
|
|
Verifies:
|
|
1. User has sufficient balance (for prepay models)
|
|
2. Provider is allowed for the user (via RBAC)
|
|
|
|
Raises:
|
|
InsufficientBalanceException: If balance is insufficient
|
|
ProviderNotAllowedException: If provider is not allowed
|
|
BillingContextError: If billing check fails unexpectedly
|
|
"""
|
|
# Context is already validated by _preflightBillingCheck()
|
|
user = self.services.user
|
|
mandateId = self.services.mandateId
|
|
featureInstanceId = getattr(self.services, 'featureInstanceId', None)
|
|
featureCode = getattr(self.services, 'featureCode', None)
|
|
|
|
try:
|
|
# Get billing service
|
|
billingService = getBillingService(user, mandateId, featureInstanceId, featureCode)
|
|
|
|
# Check balance (estimate typical AI call cost)
|
|
estimatedCost = 0.01 # ~1 cent CHF minimum
|
|
balanceCheck = billingService.checkBalance(estimatedCost)
|
|
|
|
if not balanceCheck.allowed:
|
|
logger.warning(
|
|
f"Billing check failed for user {user.id}: "
|
|
f"Balance {balanceCheck.currentBalance:.2f} CHF, "
|
|
f"Reason: {balanceCheck.reason}"
|
|
)
|
|
raise InsufficientBalanceException(
|
|
currentBalance=balanceCheck.currentBalance or 0.0,
|
|
requiredAmount=estimatedCost,
|
|
message=f"Ungenugendes Guthaben. Aktuell: CHF {balanceCheck.currentBalance:.2f}"
|
|
)
|
|
|
|
logger.debug(f"Billing check passed: Balance {balanceCheck.currentBalance:.2f} CHF")
|
|
|
|
# Check if at least one provider is allowed (RBAC check)
|
|
rbacAllowedProviders = billingService.getallowedProviders()
|
|
if not rbacAllowedProviders:
|
|
logger.warning(f"No AI providers allowed for user {user.id} in mandate {mandateId}")
|
|
raise ProviderNotAllowedException(
|
|
provider="any",
|
|
message="Keine AI-Provider fuer Ihre Rolle freigegeben. Kontaktieren Sie Ihren Administrator."
|
|
)
|
|
|
|
# Check automation-level allowedProviders restriction
|
|
automationAllowedProviders = getattr(self.services, 'allowedProviders', None)
|
|
if automationAllowedProviders:
|
|
effectiveProviders = [p for p in automationAllowedProviders if p in rbacAllowedProviders]
|
|
if not effectiveProviders:
|
|
logger.warning(f"No providers available after automation restriction. "
|
|
f"Automation allows: {automationAllowedProviders}, "
|
|
f"RBAC allows: {rbacAllowedProviders}")
|
|
raise ProviderNotAllowedException(
|
|
provider="any",
|
|
message="Die konfigurierten AI-Provider dieser Automation sind fuer Ihre Rolle nicht freigegeben."
|
|
)
|
|
logger.debug(f"Automation provider check passed: {effectiveProviders}")
|
|
|
|
# Check if preferred providers (from UI multiselect) are allowed
|
|
preferredProviders = getattr(self.services, 'preferredProviders', None)
|
|
if preferredProviders:
|
|
for provider in preferredProviders:
|
|
if provider not in rbacAllowedProviders:
|
|
logger.warning(f"Preferred provider {provider} not allowed for user {user.id}")
|
|
raise ProviderNotAllowedException(
|
|
provider=provider,
|
|
message=f"Der gewaehlte Provider '{provider}' ist fuer Ihre Rolle nicht freigegeben."
|
|
)
|
|
logger.debug(f"All preferred providers are allowed: {preferredProviders}")
|
|
|
|
logger.debug(f"Provider check passed: {len(rbacAllowedProviders)} providers allowed")
|
|
|
|
except InsufficientBalanceException:
|
|
raise
|
|
except ProviderNotAllowedException:
|
|
raise
|
|
except BillingContextError:
|
|
raise
|
|
except Exception as e:
|
|
# FAIL-SAFE: Don't silently swallow errors - log at ERROR level
|
|
logger.error(f"BILLING FAIL-SAFE: Billing check failed with unexpected error: {e}")
|
|
raise BillingContextError(f"Billing check failed: {e}")
|
|
|
|
def _createBillingCallback(self):
|
|
"""
|
|
Create a billing callback for interfaceAiObjects._callWithModel().
|
|
|
|
Returns a function that records one billing transaction per individual model call.
|
|
Each transaction contains the exact provider name AND model name.
|
|
|
|
For a 200 MB document processed with N parallel AI calls (possibly different models),
|
|
this creates N separate billing transactions - one per model call.
|
|
"""
|
|
user = self.services.user
|
|
mandateId = self.services.mandateId
|
|
featureInstanceId = getattr(self.services, 'featureInstanceId', None)
|
|
featureCode = getattr(self.services, 'featureCode', None)
|
|
|
|
# Get workflow ID if available
|
|
workflowId = None
|
|
workflow = getattr(self.services, 'workflow', None)
|
|
if workflow and hasattr(workflow, 'id'):
|
|
workflowId = workflow.id
|
|
|
|
billingService = getBillingService(user, mandateId, featureInstanceId, featureCode)
|
|
|
|
def _billingCallback(response) -> None:
|
|
"""Record billing for a single AI model call."""
|
|
if not response or getattr(response, 'errorCount', 0) > 0:
|
|
return
|
|
|
|
priceCHF = getattr(response, 'priceCHF', 0.0)
|
|
if not priceCHF or priceCHF <= 0:
|
|
return
|
|
|
|
provider = getattr(response, 'provider', None) or 'unknown'
|
|
modelName = getattr(response, 'modelName', None) or 'unknown'
|
|
|
|
try:
|
|
billingService.recordUsage(
|
|
priceCHF=priceCHF,
|
|
workflowId=workflowId,
|
|
aicoreProvider=provider,
|
|
aicoreModel=modelName,
|
|
description=f"AI: {modelName}"
|
|
)
|
|
logger.debug(
|
|
f"Billed model call: {priceCHF:.4f} CHF, "
|
|
f"provider={provider}, model={modelName}, mandate={mandateId}"
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f"BILLING: Failed to record transaction! "
|
|
f"Cost={priceCHF:.4f} CHF, user={user.id}, mandate={mandateId}, "
|
|
f"provider={provider}, model={modelName}, error={e}"
|
|
)
|
|
|
|
return _billingCallback
|
|
|
|
def _calculateEffectiveProviders(self) -> Optional[List[str]]:
|
|
"""
|
|
Calculate effective allowed providers: RBAC ∩ Workflow.
|
|
|
|
RBAC is master - only RBAC-permitted providers can ever be used.
|
|
If workflow specifies allowedProviders, intersect with RBAC.
|
|
If no workflow providers, use all RBAC-permitted providers.
|
|
|
|
Returns:
|
|
List of effective allowed providers, or None if no filtering needed
|
|
"""
|
|
try:
|
|
user = getattr(self.services, 'user', None)
|
|
mandateId = getattr(self.services, 'mandateId', None)
|
|
|
|
if not user or not mandateId:
|
|
return None
|
|
|
|
# Get RBAC-permitted providers (master list)
|
|
# Note: getBillingService is imported at module level from mainServiceBilling
|
|
featureInstanceId = getattr(self.services, 'featureInstanceId', None)
|
|
featureCode = getattr(self.services, 'featureCode', None)
|
|
billingService = getBillingService(user, mandateId, featureInstanceId, featureCode)
|
|
rbacProviders = billingService.getallowedProviders()
|
|
|
|
if not rbacProviders:
|
|
logger.warning("No RBAC-permitted providers found")
|
|
return None
|
|
|
|
# Get workflow-specified providers (optional filter)
|
|
workflowProviders = getattr(self.services, 'allowedProviders', None)
|
|
|
|
if workflowProviders:
|
|
# Intersect: only providers that are both RBAC-permitted AND workflow-allowed
|
|
effectiveProviders = [p for p in workflowProviders if p in rbacProviders]
|
|
logger.debug(f"Provider filter: RBAC={rbacProviders}, Workflow={workflowProviders}, Effective={effectiveProviders}")
|
|
else:
|
|
# No workflow filter - use all RBAC-permitted providers
|
|
effectiveProviders = rbacProviders
|
|
logger.debug(f"Provider filter: RBAC={rbacProviders} (no workflow filter)")
|
|
|
|
return effectiveProviders if effectiveProviders else None
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Error calculating effective providers: {e}")
|
|
return None
|
|
|
|
def _storeAiCallStats(self, response, request: AiCallRequest) -> None:
|
|
"""Store workflow stats after an AI call.
|
|
|
|
This method stores the AI call statistics (cost, processing time, bytes)
|
|
to the workflow stats collection for tracking and billing purposes.
|
|
|
|
Args:
|
|
response: AiCallResponse with cost/timing data
|
|
request: Original AiCallRequest for context
|
|
"""
|
|
try:
|
|
# Skip if no workflow context
|
|
workflow = getattr(self.services, 'workflow', None)
|
|
if not workflow or not hasattr(workflow, 'id') or not workflow.id:
|
|
logger.debug("No workflow context - skipping stats storage")
|
|
return
|
|
|
|
# Skip if response is an error
|
|
if not response or getattr(response, 'errorCount', 0) > 0:
|
|
logger.debug("Error response - skipping stats storage")
|
|
return
|
|
|
|
# Determine process name from operation type
|
|
opType = getattr(request.options, 'operationType', 'unknown') if request.options else 'unknown'
|
|
process = f"ai.call.{opType}"
|
|
|
|
# Store the stat
|
|
self.services.chat.storeWorkflowStat(workflow, response, process)
|
|
logger.debug(f"Stored AI call stat: {process}, cost={getattr(response, 'priceCHF', 0):.4f} CHF")
|
|
|
|
except Exception as e:
|
|
# Log but don't fail - stats storage is not critical
|
|
logger.debug(f"Could not store AI call stat: {str(e)}")
|
|
|
|
async def ensureAiObjectsInitialized(self):
|
|
"""Ensure aiObjects is initialized and submodules are ready."""
|
|
if self.aiObjects is None:
|
|
logger.info("Lazy initializing AiObjects...")
|
|
self.aiObjects = await AiObjects.create()
|
|
logger.info("AiObjects initialization completed")
|
|
# Initialize submodules after aiObjects is ready
|
|
self._initializeSubmodules()
|
|
|
|
@classmethod
|
|
async def create(cls, serviceCenter=None) -> "AiService":
|
|
"""Create AiService instance with all connectors and submodules initialized."""
|
|
logger.info("AiService.create() called")
|
|
instance = cls(serviceCenter)
|
|
logger.info("AiService created, about to call AiObjects.create()...")
|
|
instance.aiObjects = await AiObjects.create()
|
|
logger.info("AiObjects.create() completed")
|
|
# Initialize all submodules after aiObjects is ready
|
|
instance._initializeSubmodules()
|
|
logger.info("AiService submodules initialized")
|
|
return instance
|
|
|
|
# Helper methods
|
|
|
|
def _buildPromptWithPlaceholders(self, prompt: str, placeholders: Optional[Dict[str, str]]) -> str:
|
|
"""
|
|
Build full prompt by replacing placeholders with their content.
|
|
Uses the new {{KEY:placeholder}} format.
|
|
|
|
Args:
|
|
prompt: The base prompt template
|
|
placeholders: Dictionary of placeholder key-value pairs
|
|
|
|
Returns:
|
|
Prompt with placeholders replaced
|
|
"""
|
|
if not placeholders:
|
|
return prompt
|
|
|
|
full_prompt = prompt
|
|
for placeholder, content in placeholders.items():
|
|
# Skip if content is None or empty
|
|
if content is None:
|
|
continue
|
|
# Replace {{KEY:placeholder}}
|
|
full_prompt = full_prompt.replace(f"{{{{KEY:{placeholder}}}}}", str(content))
|
|
|
|
return full_prompt
|
|
|
|
async def _analyzePromptAndCreateOptions(self, prompt: str) -> AiCallOptions:
|
|
"""Analyze prompt to determine appropriate AiCallOptions parameters."""
|
|
try:
|
|
# Get dynamic enum values from Pydantic models
|
|
operationTypes = [e.value for e in OperationTypeEnum]
|
|
priorities = [e.value for e in PriorityEnum]
|
|
processingModes = [e.value for e in ProcessingModeEnum]
|
|
|
|
# Create analysis prompt for AI to determine operation type and parameters
|
|
analysisPrompt = f"""
|
|
You are an AI operation analyzer. Analyze the following prompt and determine the most appropriate operation type and parameters.
|
|
|
|
PROMPT TO ANALYZE:
|
|
{self.services.utils.sanitizePromptContent(prompt, 'userinput')}
|
|
|
|
Based on the prompt content, determine:
|
|
1. operationType: Choose the most appropriate from: {', '.join(operationTypes)}
|
|
2. priority: Choose from: {', '.join(priorities)}
|
|
3. processingMode: Choose from: {', '.join(processingModes)}
|
|
4. compressPrompt: true/false (true for story-like prompts, false for structured prompts with JSON/schemas)
|
|
5. compressContext: true/false (true to summarize context, false to process fully)
|
|
|
|
Respond with ONLY a JSON object in this exact format:
|
|
{{
|
|
"operationType": "dataAnalyse",
|
|
"priority": "balanced",
|
|
"processingMode": "basic",
|
|
"compressPrompt": true,
|
|
"compressContext": true
|
|
}}
|
|
"""
|
|
|
|
# Use AI to analyze the prompt
|
|
request = AiCallRequest(
|
|
prompt=analysisPrompt,
|
|
options=AiCallOptions(
|
|
operationType=OperationTypeEnum.DATA_ANALYSE,
|
|
priority=PriorityEnum.SPEED,
|
|
processingMode=ProcessingModeEnum.BASIC,
|
|
compressPrompt=True,
|
|
compressContext=False
|
|
)
|
|
)
|
|
|
|
response = await self.callAi(request)
|
|
|
|
# Parse AI response using structured parsing with AiCallOptions model
|
|
try:
|
|
# Use parseJsonWithModel to parse response into AiCallOptions (handles enum conversion automatically)
|
|
analysis = parseJsonWithModel(response.content, AiCallOptions)
|
|
return analysis
|
|
except Exception as e:
|
|
logger.warning(f"Failed to parse AI analysis response: {e}")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Prompt analysis failed: {e}")
|
|
|
|
# Fallback to default options
|
|
return AiCallOptions(
|
|
operationType=OperationTypeEnum.DATA_ANALYSE,
|
|
priority=PriorityEnum.BALANCED,
|
|
processingMode=ProcessingModeEnum.BASIC
|
|
)
|
|
|
|
async def callAiWithLooping(
|
|
self,
|
|
prompt: str,
|
|
options: AiCallOptions,
|
|
debugPrefix: str = "ai_call",
|
|
promptBuilder: Optional[callable] = None,
|
|
promptArgs: Optional[Dict[str, Any]] = None,
|
|
operationId: Optional[str] = None,
|
|
userPrompt: Optional[str] = None,
|
|
contentParts: Optional[List[ContentPart]] = None, # ARCHITECTURE: Support ContentParts for large content
|
|
useCaseId: Optional[str] = None # REQUIRED: Explicit use case ID for generic looping system
|
|
) -> str:
|
|
"""Public method: Delegate to AiCallLooper for AI calls with looping support."""
|
|
return await self.aiCallLooper.callAiWithLooping(
|
|
prompt, options, debugPrefix, promptBuilder, promptArgs, operationId, userPrompt, contentParts, useCaseId
|
|
)
|
|
|
|
# JSON merging logic moved to subJsonResponseHandling.py
|
|
|
|
def _extractSectionsFromResponse(
|
|
self,
|
|
result: str,
|
|
iteration: int,
|
|
debugPrefix: str,
|
|
allSections: List[Dict[str, Any]] = None,
|
|
accumulationState: Optional[JsonAccumulationState] = None
|
|
) -> Tuple[List[Dict[str, Any]], bool, Optional[Dict[str, Any]], Optional[JsonAccumulationState]]:
|
|
"""Delegate to ResponseParser."""
|
|
return self.responseParser.extractSectionsFromResponse(
|
|
result, iteration, debugPrefix, allSections, accumulationState
|
|
)
|
|
|
|
def _shouldContinueGeneration(
|
|
self,
|
|
allSections: List[Dict[str, Any]],
|
|
iteration: int,
|
|
wasJsonComplete: bool,
|
|
rawResponse: str = None
|
|
) -> bool:
|
|
"""Delegate to ResponseParser."""
|
|
return self.responseParser.shouldContinueGeneration(
|
|
allSections, iteration, wasJsonComplete, rawResponse
|
|
)
|
|
|
|
def _extractDocumentMetadata(
|
|
self,
|
|
parsedResult: Dict[str, Any]
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Delegate to ResponseParser."""
|
|
return self.responseParser.extractDocumentMetadata(parsedResult)
|
|
|
|
def _buildFinalResultFromSections(
|
|
self,
|
|
allSections: List[Dict[str, Any]],
|
|
documentMetadata: Optional[Dict[str, Any]] = None
|
|
) -> str:
|
|
"""Delegate to ResponseParser."""
|
|
return self.responseParser.buildFinalResultFromSections(allSections, documentMetadata)
|
|
|
|
# Public API Methods
|
|
|
|
# Planning AI Call
|
|
async def callAiPlanning(
|
|
self,
|
|
prompt: str,
|
|
placeholders: Optional[List[PromptPlaceholder]] = None,
|
|
debugType: Optional[str] = None
|
|
) -> str:
|
|
"""
|
|
Planning AI call for task planning, action planning, action selection, etc.
|
|
Always uses static parameters optimized for planning tasks.
|
|
|
|
Args:
|
|
prompt: The planning prompt
|
|
placeholders: Optional list of placeholder replacements
|
|
debugType: Optional debug file type identifier (e.g., 'taskplan', 'dynamic', 'intentanalysis')
|
|
If not provided, defaults to 'plan'
|
|
|
|
Returns:
|
|
Planning JSON response
|
|
"""
|
|
await self.ensureAiObjectsInitialized()
|
|
|
|
# Planning calls always use static parameters
|
|
options = AiCallOptions(
|
|
operationType=OperationTypeEnum.PLAN,
|
|
priority=PriorityEnum.QUALITY,
|
|
processingMode=ProcessingModeEnum.DETAILED,
|
|
compressPrompt=False,
|
|
compressContext=False
|
|
)
|
|
|
|
# Build full prompt with placeholders
|
|
if placeholders:
|
|
placeholdersDict = {p.label: p.content for p in placeholders}
|
|
fullPrompt = self._buildPromptWithPlaceholders(prompt, placeholdersDict)
|
|
else:
|
|
fullPrompt = prompt
|
|
|
|
# Root-cause fix: planning must return raw single-shot JSON, not section-based output
|
|
request = AiCallRequest(
|
|
prompt=fullPrompt,
|
|
context="",
|
|
options=options
|
|
)
|
|
|
|
# Debug: persist prompt/response for analysis with context-specific naming
|
|
debugPrefix = debugType if debugType else "plan"
|
|
self.services.utils.writeDebugFile(fullPrompt, f"{debugPrefix}_prompt")
|
|
response = await self.callAi(request) # Use callAi to ensure stats are stored
|
|
result = response.content or ""
|
|
self.services.utils.writeDebugFile(result, f"{debugPrefix}_response")
|
|
return result
|
|
|
|
# Helper methods for callAiContent refactoring
|
|
|
|
async def _handleImageGeneration(
|
|
self,
|
|
prompt: str,
|
|
options: AiCallOptions,
|
|
title: Optional[str],
|
|
parentOperationId: Optional[str]
|
|
) -> AiResponse:
|
|
"""Handle IMAGE_GENERATE operation type using image generation path."""
|
|
from modules.services.serviceGeneration.paths.imagePath import ImageGenerationPath
|
|
|
|
imagePath = ImageGenerationPath(self.services)
|
|
|
|
# Extract format from options
|
|
format = options.resultFormat or "png"
|
|
|
|
return await imagePath.generateImages(
|
|
userPrompt=prompt,
|
|
format=format,
|
|
title=title,
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
async def _handleWebOperation(
|
|
self,
|
|
prompt: str,
|
|
options: AiCallOptions,
|
|
opType: OperationTypeEnum,
|
|
aiOperationId: str
|
|
) -> AiResponse:
|
|
"""Handle WEB_SEARCH_DATA and WEB_CRAWL operation types."""
|
|
self.services.chat.progressLogUpdate(aiOperationId, 0.4, f"Calling AI for {opType.name}")
|
|
|
|
request = AiCallRequest(
|
|
prompt=prompt, # Raw JSON prompt - connector will parse it
|
|
context="",
|
|
options=options
|
|
)
|
|
|
|
response = await self.callAi(request)
|
|
|
|
if not response.content:
|
|
errorMsg = f"No content returned from {opType.name}: {response.content}"
|
|
logger.error(f"Error in {opType.name}: {errorMsg}")
|
|
self.services.chat.progressLogFinish(aiOperationId, False)
|
|
raise ValueError(errorMsg)
|
|
|
|
metadata = AiResponseMetadata(
|
|
operationType=opType.value
|
|
)
|
|
|
|
# Note: Stats are now stored centrally in callAi() - no need to duplicate here
|
|
|
|
self.services.chat.progressLogUpdate(aiOperationId, 0.9, f"{opType.name} completed")
|
|
self.services.chat.progressLogFinish(aiOperationId, True)
|
|
|
|
# Preserve metadata from response if available (e.g., results_with_content from Tavily)
|
|
# Check if response has metadata attribute (AiCallResponse from callAi)
|
|
if hasattr(response, 'metadata') and response.metadata:
|
|
# If metadata is a dict, store it in additionalData
|
|
if isinstance(response.metadata, dict):
|
|
if not metadata.additionalData:
|
|
metadata.additionalData = {}
|
|
metadata.additionalData.update(response.metadata)
|
|
# If metadata is an object with attributes, extract them
|
|
elif hasattr(response.metadata, '__dict__'):
|
|
if not metadata.additionalData:
|
|
metadata.additionalData = {}
|
|
for key, value in response.metadata.__dict__.items():
|
|
if not key.startswith('_'):
|
|
metadata.additionalData[key] = value
|
|
|
|
return AiResponse(
|
|
content=response.content,
|
|
metadata=metadata
|
|
)
|
|
|
|
def _getIntentForDocument(
|
|
self,
|
|
docId: str,
|
|
intents: Optional[List[DocumentIntent]]
|
|
) -> Optional[DocumentIntent]:
|
|
"""Find DocumentIntent for given documentId."""
|
|
if not intents:
|
|
return None
|
|
for intent in intents:
|
|
if intent.documentId == docId:
|
|
return intent
|
|
return None
|
|
|
|
async def clarifyDocumentIntents(
|
|
self,
|
|
documents: List[ChatDocument],
|
|
userPrompt: str,
|
|
actionParameters: Dict[str, Any],
|
|
parentOperationId: str
|
|
) -> List[DocumentIntent]:
|
|
"""Public method: Delegate to DocumentIntentAnalyzer."""
|
|
return await self.intentAnalyzer.clarifyDocumentIntents(
|
|
documents, userPrompt, actionParameters, parentOperationId
|
|
)
|
|
|
|
async def extractAndPrepareContent(
|
|
self,
|
|
documents: List[ChatDocument],
|
|
documentIntents: List[DocumentIntent],
|
|
parentOperationId: str
|
|
) -> List[ContentPart]:
|
|
"""Public method: Delegate to ContentExtractor."""
|
|
return await self.contentExtractor.extractAndPrepareContent(
|
|
documents, documentIntents, parentOperationId, self._getIntentForDocument
|
|
)
|
|
|
|
async def generateStructure(
|
|
self,
|
|
userPrompt: str,
|
|
contentParts: List[ContentPart],
|
|
outputFormat: Optional[str] = None,
|
|
parentOperationId: str = None
|
|
) -> Dict[str, Any]:
|
|
"""Public method: Delegate to StructureGenerator."""
|
|
return await self.structureGenerator.generateStructure(
|
|
userPrompt, contentParts, outputFormat, parentOperationId
|
|
)
|
|
|
|
async def fillStructure(
|
|
self,
|
|
structure: Dict[str, Any],
|
|
contentParts: List[ContentPart],
|
|
userPrompt: str,
|
|
parentOperationId: str
|
|
) -> Dict[str, Any]:
|
|
"""Public method: Delegate to StructureFiller."""
|
|
return await self.structureFiller.fillStructure(
|
|
structure, contentParts, userPrompt, parentOperationId
|
|
)
|
|
|
|
async def renderResult(
|
|
self,
|
|
filledStructure: Dict[str, Any],
|
|
outputFormat: str,
|
|
language: str,
|
|
title: str,
|
|
userPrompt: str,
|
|
parentOperationId: str
|
|
) -> List[RenderedDocument]:
|
|
"""
|
|
Phase 5E: Rendert gefüllte Struktur zum Ziel-Format.
|
|
Jedes Dokument wird einzeln gerendert, jeder Renderer kann 1..n Dokumente zurückgeben.
|
|
|
|
Render filled structure to documents.
|
|
Per-document format and language are extracted from structure (validated in State 3).
|
|
The outputFormat and language parameters are only used as global fallbacks.
|
|
Multiple documents can have different formats and languages.
|
|
|
|
Args:
|
|
filledStructure: Gefüllte Struktur mit elements
|
|
outputFormat: Ziel-Format (pdf, docx, html, etc.) - Global fallback
|
|
language: Language (global fallback) - Per-document language extracted from structure
|
|
title: Dokument-Titel
|
|
userPrompt: User-Anfrage
|
|
parentOperationId: Parent Operation-ID für ChatLog-Hierarchie
|
|
|
|
Returns:
|
|
List of RenderedDocument objects.
|
|
Jedes RenderedDocument repräsentiert ein gerendertes Dokument (Hauptdokument oder unterstützende Datei)
|
|
"""
|
|
# Language comes from structure (per-document), validated in State 3
|
|
# This parameter is only used as global fallback if structure validation fails
|
|
# Use validated currentUserLanguage as fallback (always valid)
|
|
if not language:
|
|
language = self._getUserLanguage() if hasattr(self, '_getUserLanguage') else (self.services.currentUserLanguage if hasattr(self.services, 'currentUserLanguage') else 'en')
|
|
# Erstelle Operation-ID für Rendering
|
|
renderOperationId = f"{parentOperationId}_rendering"
|
|
|
|
# Starte ChatLog mit Parent-Referenz
|
|
self.services.chat.progressLogStart(
|
|
renderOperationId,
|
|
"Content Rendering",
|
|
"Rendering",
|
|
f"Rendering to {outputFormat} format",
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
try:
|
|
from modules.services.serviceGeneration.mainServiceGeneration import GenerationService
|
|
|
|
generationService = GenerationService(self.services)
|
|
|
|
# renderReport verarbeitet jetzt jedes Dokument einzeln
|
|
# und gibt Liste von (documentData, mimeType, filename) zurück
|
|
renderedDocuments = await generationService.renderReport(
|
|
filledStructure,
|
|
outputFormat,
|
|
language, # Pass language (global fallback, per-document extracted in renderReport)
|
|
title,
|
|
userPrompt,
|
|
self,
|
|
parentOperationId=renderOperationId # Parent-Referenz für ChatLog-Hierarchie
|
|
)
|
|
|
|
# ChatLog abschließen
|
|
self.services.chat.progressLogFinish(renderOperationId, True)
|
|
|
|
return renderedDocuments
|
|
|
|
except Exception as e:
|
|
self.services.chat.progressLogFinish(renderOperationId, False)
|
|
logger.error(f"Error in _renderResult: {str(e)}")
|
|
raise
|
|
|
|
def _shouldSkipContentPart(
|
|
self,
|
|
part: ContentPart
|
|
) -> bool:
|
|
"""Check if ContentPart should be skipped (already structured JSON)."""
|
|
if part.typeGroup == "structure" and part.mimeType == "application/json":
|
|
if part.metadata.get("skipExtraction", False):
|
|
logger.debug(f"Skipping already-structured JSON ContentPart {part.id} (skipExtraction=True)")
|
|
return True
|
|
try:
|
|
if isinstance(part.data, str):
|
|
jsonData = json.loads(part.data)
|
|
if isinstance(jsonData, dict) and ("documents" in jsonData or "sections" in jsonData):
|
|
logger.debug(f"Skipping already-structured JSON ContentPart {part.id} (contains documents/sections)")
|
|
return True
|
|
except Exception:
|
|
pass # Not JSON, continue processing
|
|
return False
|
|
|
|
async def callAiContent(
|
|
self,
|
|
prompt: str,
|
|
options: AiCallOptions,
|
|
contentParts: Optional[List[ContentPart]] = None,
|
|
documentList: Optional[Any] = None, # DocumentReferenceList
|
|
documentIntents: Optional[List[DocumentIntent]] = None,
|
|
outputFormat: Optional[str] = None,
|
|
title: Optional[str] = None,
|
|
parentOperationId: Optional[str] = None,
|
|
generationIntent: Optional[str] = None # NEW: Explicit intent from action (skips detection)
|
|
) -> AiResponse:
|
|
"""
|
|
Unified AI content generation with explicit intent requirement.
|
|
|
|
All AI-Actions (ai.process, ai.generateDocument, etc.) route through here.
|
|
They differ only in parameters, not in logic.
|
|
|
|
Args:
|
|
prompt: The main prompt for the AI call
|
|
options: AI call configuration options (REQUIRED - operationType must be set)
|
|
contentParts: Optional list of already-extracted content parts (preferred)
|
|
documentList: Optional DocumentReferenceList (wird zu ChatDocuments konvertiert)
|
|
documentIntents: Optional list of DocumentIntent objects (wird erstellt wenn nicht vorhanden)
|
|
outputFormat: Optional output format for document generation (e.g., 'pdf', 'docx', 'xlsx')
|
|
title: Optional title for generated documents
|
|
parentOperationId: Optional parent operation ID for hierarchical logging
|
|
generationIntent: REQUIRED explicit intent ("document" | "code" | "image") from action.
|
|
NO auto-detection - actions must explicitly specify intent.
|
|
|
|
Returns:
|
|
AiResponse with content, metadata, and optional documents
|
|
"""
|
|
await self.ensureAiObjectsInitialized()
|
|
|
|
# Erstelle Operation-ID
|
|
workflowId = self.services.workflow.id if self.services.workflow else f"no-workflow-{int(time.time())}"
|
|
aiOperationId = f"ai_content_{workflowId}_{int(time.time())}"
|
|
|
|
# Starte Progress-Tracking mit Parent-Referenz
|
|
formatDisplay = outputFormat if outputFormat else "auto-determined"
|
|
self.services.chat.progressLogStart(
|
|
aiOperationId,
|
|
"AI content processing",
|
|
"Content Processing",
|
|
f"Format: {formatDisplay}",
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
try:
|
|
# outputFormat is optional - if None, formats determined from prompt by AI
|
|
# No default fallback here - let AI service handle it
|
|
|
|
opType = getattr(options, "operationType", None)
|
|
if not opType:
|
|
options.operationType = OperationTypeEnum.DATA_GENERATE
|
|
opType = OperationTypeEnum.DATA_GENERATE
|
|
|
|
# Route zu Operation-spezifischen Handlern
|
|
if opType == OperationTypeEnum.IMAGE_GENERATE:
|
|
# Image generation - route to image path
|
|
return await self._handleImageGeneration(prompt, options, title, parentOperationId)
|
|
|
|
if opType == OperationTypeEnum.WEB_SEARCH_DATA or opType == OperationTypeEnum.WEB_CRAWL:
|
|
return await self._handleWebOperation(prompt, options, opType, aiOperationId)
|
|
|
|
# Data generation - REQUIRES explicit generationIntent
|
|
if opType == OperationTypeEnum.DATA_GENERATE:
|
|
if not generationIntent:
|
|
errorMsg = (
|
|
"generationIntent is required for DATA_GENERATE operation. "
|
|
"Actions must explicitly specify 'document' or 'code' intent. "
|
|
"No auto-detection - use qualified actions (ai.generateDocument, ai.generateCode)."
|
|
)
|
|
logger.error(errorMsg)
|
|
self.services.chat.progressLogFinish(aiOperationId, False)
|
|
raise ValueError(errorMsg)
|
|
|
|
# Route based on explicit intent (no auto-detection, no fallback)
|
|
if generationIntent == "code":
|
|
# Route to code generation path
|
|
return await self._handleCodeGeneration(
|
|
prompt=prompt,
|
|
options=options,
|
|
contentParts=contentParts,
|
|
outputFormat=outputFormat,
|
|
title=title,
|
|
parentOperationId=parentOperationId
|
|
)
|
|
else:
|
|
# Route to document generation path (existing behavior)
|
|
return await self._handleDocumentGeneration(
|
|
prompt=prompt,
|
|
options=options,
|
|
documentList=documentList,
|
|
documentIntents=documentIntents,
|
|
contentParts=contentParts,
|
|
outputFormat=outputFormat,
|
|
title=title,
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
# DATA_EXTRACT: Extract content from documents and process with AI (no structure generation)
|
|
if opType == OperationTypeEnum.DATA_EXTRACT:
|
|
return await self._handleDataExtraction(
|
|
prompt=prompt,
|
|
options=options,
|
|
documentList=documentList,
|
|
documentIntents=documentIntents,
|
|
contentParts=contentParts,
|
|
outputFormat=outputFormat,
|
|
title=title,
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
# Other operation types (DATA_ANALYSE, etc.) - not supported
|
|
errorMsg = f"Unsupported operation type: {opType}. Supported types: IMAGE_GENERATE, DATA_GENERATE, DATA_EXTRACT"
|
|
logger.error(errorMsg)
|
|
self.services.chat.progressLogFinish(aiOperationId, False)
|
|
raise ValueError(errorMsg)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in callAiContent: {str(e)}")
|
|
self.services.chat.progressLogFinish(aiOperationId, False)
|
|
raise
|
|
|
|
async def _handleDataExtraction(
|
|
self,
|
|
prompt: str,
|
|
options: AiCallOptions,
|
|
documentList: Optional[Any],
|
|
documentIntents: Optional[List[DocumentIntent]],
|
|
contentParts: Optional[List[ContentPart]],
|
|
outputFormat: str,
|
|
title: str,
|
|
parentOperationId: Optional[str]
|
|
) -> AiResponse:
|
|
"""
|
|
Handle DATA_EXTRACT: Extract content from documents (no AI), then process with AI.
|
|
This is the original flow: extract all documents first, then process contentParts with AI.
|
|
"""
|
|
import time
|
|
|
|
# Create operation ID
|
|
workflowId = self.services.workflow.id if self.services.workflow else f"no-workflow-{int(time.time())}"
|
|
extractOperationId = f"data_extract_{workflowId}_{int(time.time())}"
|
|
|
|
# Start progress tracking
|
|
self.services.chat.progressLogStart(
|
|
extractOperationId,
|
|
"Data Extraction",
|
|
"Extraction",
|
|
f"Format: {outputFormat}",
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
try:
|
|
# Step 1: Get documents from documentList
|
|
documents = []
|
|
if documentList:
|
|
documents = self.services.chat.getChatDocumentsFromDocumentList(documentList)
|
|
|
|
# Filter: Remove original documents if already covered by pre-extracted JSONs
|
|
# (to prevent duplicate ContentParts - pre-extracted JSONs contain already extracted ContentParts)
|
|
if documents:
|
|
# Step 1: Identify all original document IDs covered by pre-extracted JSONs
|
|
originalDocIdsCoveredByPreExtracted = set()
|
|
for doc in documents:
|
|
preExtracted = self.intentAnalyzer.resolvePreExtractedDocument(doc)
|
|
if preExtracted:
|
|
originalDocId = preExtracted["originalDocument"]["id"]
|
|
originalDocIdsCoveredByPreExtracted.add(originalDocId)
|
|
logger.debug(f"Found pre-extracted JSON {doc.id} covering original document {originalDocId}")
|
|
|
|
# Step 2: Filter documents - remove originals covered by pre-extracted JSONs
|
|
filteredDocuments = []
|
|
for doc in documents:
|
|
preExtracted = self.intentAnalyzer.resolvePreExtractedDocument(doc)
|
|
if preExtracted:
|
|
filteredDocuments.append(doc) # Keep pre-extracted JSON
|
|
elif doc.id in originalDocIdsCoveredByPreExtracted:
|
|
logger.info(f"Skipping original document {doc.id} ({doc.fileName}) - already covered by pre-extracted JSON")
|
|
else:
|
|
filteredDocuments.append(doc) # Keep regular document
|
|
|
|
documents = filteredDocuments # Use filtered list
|
|
|
|
# Step 2: Clarify document intents (if not provided) - REQUIRED for all documents
|
|
if not documentIntents and documents:
|
|
documentIntents = await self.clarifyDocumentIntents(
|
|
documents,
|
|
prompt,
|
|
{"outputFormat": outputFormat},
|
|
extractOperationId
|
|
)
|
|
|
|
# Step 3: Extract and prepare content (NO AI - pure extraction) - REQUIRED for all documents
|
|
if documents:
|
|
preparedContentParts = await self.extractAndPrepareContent(
|
|
documents,
|
|
documentIntents or [],
|
|
extractOperationId
|
|
)
|
|
|
|
# Merge with provided contentParts (if any)
|
|
if contentParts:
|
|
for part in contentParts:
|
|
if part.metadata.get("skipExtraction", False):
|
|
part.metadata.setdefault("contentFormat", "extracted")
|
|
part.metadata.setdefault("isPreExtracted", True)
|
|
preparedContentParts.extend(contentParts)
|
|
|
|
contentParts = preparedContentParts
|
|
|
|
# Step 4: Process extracted contentParts with AI (simple text processing, no structure generation)
|
|
if not contentParts:
|
|
raise ValueError("No content extracted from documents")
|
|
|
|
# Use simple AI call to process extracted content
|
|
# Prepare content for AI processing
|
|
contentText = "\n\n".join([
|
|
f"[Document: {part.metadata.get('documentName', 'Unknown')}]\n{part.data}"
|
|
for part in contentParts
|
|
if part.data
|
|
])
|
|
|
|
# Check content size and use chunking if needed
|
|
# Conservative estimate: 2 bytes per token, 80% of model limit for safety
|
|
contentSizeBytes = len(contentText.encode('utf-8'))
|
|
promptSizeBytes = len(prompt.encode('utf-8'))
|
|
totalSizeBytes = contentSizeBytes + promptSizeBytes
|
|
estimatedTokens = totalSizeBytes / 2 # Conservative: 2 bytes per token
|
|
|
|
# Get max model context (use Claude's 200k as reference, 80% = 160k tokens)
|
|
maxSafeTokens = 160000
|
|
|
|
if estimatedTokens > maxSafeTokens:
|
|
# Content too large - use chunking via ExtractionService
|
|
logger.warning(f"Content too large for single AI call: ~{estimatedTokens:.0f} tokens (limit: {maxSafeTokens}). Using chunked processing.")
|
|
|
|
# Use ExtractionService for chunked processing
|
|
extractionService = self.services.extraction
|
|
aiResponse = await extractionService.processContentPartsWithPrompt(
|
|
contentParts=contentParts,
|
|
prompt=prompt,
|
|
aiObjects=self.aiObjects,
|
|
options=options,
|
|
operationId=extractOperationId,
|
|
parentOperationId=parentOperationId
|
|
)
|
|
else:
|
|
# Content fits - use single AI call
|
|
aiRequest = AiCallRequest(
|
|
prompt=f"{prompt}\n\nExtracted Content:\n{contentText}",
|
|
context="",
|
|
options=options
|
|
)
|
|
|
|
aiResponse = await self.callAi(aiRequest)
|
|
|
|
# Create response document
|
|
resultDocument = DocumentData(
|
|
documentName=f"{title or 'extracted_data'}.{outputFormat}",
|
|
documentData=aiResponse.content.encode('utf-8') if isinstance(aiResponse.content, str) else aiResponse.content,
|
|
mimeType=f"text/{outputFormat}" if outputFormat in ["txt", "json", "csv"] else "application/octet-stream"
|
|
)
|
|
|
|
metadata = AiResponseMetadata(
|
|
title=title or "Extracted Data",
|
|
operationType=OperationTypeEnum.DATA_EXTRACT.value
|
|
)
|
|
|
|
self.services.chat.progressLogFinish(extractOperationId, True)
|
|
|
|
return AiResponse(
|
|
content=aiResponse.content if isinstance(aiResponse.content, str) else aiResponse.content.decode('utf-8', errors='replace'),
|
|
metadata=metadata,
|
|
documents=[resultDocument]
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in data extraction: {str(e)}")
|
|
self.services.chat.progressLogFinish(extractOperationId, False)
|
|
raise
|
|
|
|
async def _handleCodeGeneration(
|
|
self,
|
|
prompt: str,
|
|
options: AiCallOptions,
|
|
contentParts: Optional[List[ContentPart]],
|
|
outputFormat: str,
|
|
title: str,
|
|
parentOperationId: Optional[str]
|
|
) -> AiResponse:
|
|
"""Handle code generation using code generation path."""
|
|
from modules.services.serviceGeneration.paths.codePath import CodeGenerationPath
|
|
|
|
codePath = CodeGenerationPath(self.services)
|
|
return await codePath.generateCode(
|
|
userPrompt=prompt,
|
|
outputFormat=outputFormat,
|
|
contentParts=contentParts,
|
|
title=title or "Generated Code",
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
async def _handleDocumentGeneration(
|
|
self,
|
|
prompt: str,
|
|
options: AiCallOptions,
|
|
documentList: Optional[Any],
|
|
documentIntents: Optional[List[DocumentIntent]],
|
|
contentParts: Optional[List[ContentPart]],
|
|
outputFormat: str,
|
|
title: str,
|
|
parentOperationId: Optional[str]
|
|
) -> AiResponse:
|
|
"""Handle document generation using document generation path."""
|
|
from modules.services.serviceGeneration.paths.documentPath import DocumentGenerationPath
|
|
|
|
# Set compression options for document generation
|
|
options.compressPrompt = False
|
|
options.compressContext = False
|
|
|
|
documentPath = DocumentGenerationPath(self.services)
|
|
return await documentPath.generateDocument(
|
|
userPrompt=prompt,
|
|
documentList=documentList,
|
|
documentIntents=documentIntents,
|
|
contentParts=contentParts,
|
|
outputFormat=outputFormat,
|
|
title=title or "Generated Document",
|
|
parentOperationId=parentOperationId
|
|
)
|
|
|
|
|
|
def _determineDocumentName(
|
|
self,
|
|
filledStructure: Dict[str, Any],
|
|
outputFormat: str,
|
|
title: Optional[str]
|
|
) -> str:
|
|
"""Bestimme Dokument-Namen aus Struktur oder Titel."""
|
|
# Versuche aus Struktur zu extrahieren
|
|
if isinstance(filledStructure, dict) and "documents" in filledStructure:
|
|
docs = filledStructure["documents"]
|
|
if isinstance(docs, list) and len(docs) > 0:
|
|
firstDoc = docs[0]
|
|
if isinstance(firstDoc, dict) and firstDoc.get("filename"):
|
|
return firstDoc["filename"]
|
|
|
|
# Fallback zu Titel
|
|
if title:
|
|
sanitized = re.sub(r"[^a-zA-Z0-9._-]", "_", title)
|
|
sanitized = re.sub(r"_+", "_", sanitized).strip("_")
|
|
if sanitized:
|
|
if not sanitized.lower().endswith(f".{outputFormat}"):
|
|
return f"{sanitized}.{outputFormat}"
|
|
return sanitized
|
|
|
|
return f"generated.{outputFormat}"
|
|
|