pieter/eveAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- Introduction of TRACIE_KO_INTERVIEW_DEFINITION_SPECIALIST

Browse Source 
- Re-introduction of EveAIAsset
- Make translation services resistent for situation with and without current_event defined.
- Ensure first question is asked in eveai_chat_client
- Start of version 1.4.0 of TRAICIE_SELECTION_SPECIALIST
This commit is contained in:
Josako
2025-07-02 16:58:43 +02:00
parent fbc9f44ac8
commit 51d029d960
34 changed files with 1292 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
common/services/utils/answer_check_services.py Normal file
View File

@@ -0,0 +1,65 @@
from flask import current_app, session
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from common.utils.business_event import BusinessEvent
from common.utils.business_event_context import current_event
from common.utils.model_utils import get_template
from eveai_chat_workers.outputs.globals.q_a_output.q_a_output_v1_0 import QAOutput
class AnswerCheckServices:
@staticmethod
def check_affirmative_answer(question: str, answer: str, language_iso: str) -> bool:
return AnswerCheckServices._check_answer(question, answer, language_iso, "check_affirmative_answer",
"Check Affirmative Answer")
@staticmethod
def check_additional_information(question: str, answer: str, language_iso: str) -> bool:
return AnswerCheckServices._check_answer(question, answer, language_iso, "check_additional_information",
"Check Additional Information")
@staticmethod
def _check_answer(question: str, answer: str, language_iso: str, template_name: str, span_name: str) -> bool:
if language_iso.strip() == '':
raise ValueError("Language cannot be empty")
language = current_app.config.get('SUPPORTED_LANGUAGE_ISO639_1_LOOKUP').get(language_iso)
if language is None:
raise ValueError(f"Unsupported language: {language_iso}")
if question.strip() == '':
raise ValueError("Question cannot be empty")
if answer.strip() == '':
raise ValueError("Answer cannot be empty")
tenant_id = session.get('tenant').get('id')
if not current_event:
with BusinessEvent('Answer Check Service', tenant_id):
with current_event.create_span(span_name):
return AnswerCheckServices._check_answer_logic(question, answer, language, template_name)
else:
with current_event.create_span('Check Affirmative Answer'):
return AnswerCheckServices._check_answer_logic(question, answer, language, template_name)
@staticmethod
def _check_answer_logic(question: str, answer: str, language: str, template_name: str) -> bool:
prompt_params = {
'question': question,
'answer': answer,
'language': language,
}
template, llm = get_template(template_name)
check_answer_prompt = ChatPromptTemplate.from_template(template)
setup = RunnablePassthrough()
output_schema = QAOutput
structured_llm = llm.with_structured_output(output_schema)
chain = (setup | check_answer_prompt | structured_llm )
raw_answer = chain.invoke(prompt_params)
current_app.logger.debug(f"Raw answer: {raw_answer}")
return raw_answer.answer

152
common/services/utils/translation_services.py
View File

@@ -1,5 +1,8 @@
import json
from typing import Dict, Any, Optional
from flask import session
from common.extensions import cache_manager
from common.utils.business_event import BusinessEvent
from common.utils.business_event_context import current_event
@@ -7,11 +10,13 @@ from common.utils.business_event_context import current_event
class TranslationServices:
@staticmethod
def translate_config(config_data: Dict[str, Any], field_config: str, target_language: str, source_language: Optional[str] = None, context: Optional[str] = None) -> Dict[str, Any]:
def translate_config(tenant_id: int, config_data: Dict[str, Any], field_config: str, target_language: str,
source_language: Optional[str] = None, context: Optional[str] = None) -> Dict[str, Any]:
"""
Vertaalt een configuratie op basis van een veld-configuratie.
Args:
tenant_id: Identificatie van de tenant waarvoor we de vertaling doen.
config_data: Een dictionary of JSON (die dan wordt geconverteerd naar een dictionary) met configuratiegegevens
field_config: De naam van een veld-configuratie (bijv. 'fields')
target_language: De taal waarnaar vertaald moet worden
@@ -21,6 +26,26 @@ class TranslationServices:
Returns:
Een dictionary met de vertaalde configuratie
"""
config_type = config_data.get('type', 'Unknown')
config_version = config_data.get('version', 'Unknown')
span_name = f"{config_type}-{config_version}-{field_config}"
if current_event:
with current_event.create_span(span_name):
translated_config = TranslationServices._translate_config(tenant_id, config_data, field_config,
target_language, source_language, context)
return translated_config
else:
with BusinessEvent('Config Translation Service', tenant_id):
with current_event.create_span(span_name):
translated_config = TranslationServices._translate_config(tenant_id, config_data, field_config,
target_language, source_language, context)
return translated_config
@staticmethod
def _translate_config(tenant_id: int, config_data: Dict[str, Any], field_config: str, target_language: str,
source_language: Optional[str] = None, context: Optional[str] = None) -> Dict[str, Any]:
# Zorg ervoor dat we een dictionary hebben
if isinstance(config_data, str):
config_data = json.loads(config_data)
@@ -31,78 +56,79 @@ class TranslationServices:
# Haal type en versie op voor de Business Event span
config_type = config_data.get('type', 'Unknown')
config_version = config_data.get('version', 'Unknown')
span_name = f"{config_type}-{config_version}-{field_config}"
# Start een Business Event context
with BusinessEvent('Config Translation Service', 0):
with current_event.create_span(span_name):
# Controleer of de gevraagde veld-configuratie bestaat
if field_config in config_data:
fields = config_data[field_config]
if field_config in config_data:
fields = config_data[field_config]
# Haal description uit metadata voor context als geen context is opgegeven
description_context = ""
if not context and 'metadata' in config_data and 'description' in config_data['metadata']:
description_context = config_data['metadata']['description']
# Haal description uit metadata voor context als geen context is opgegeven
description_context = ""
if not context and 'metadata' in config_data and 'description' in config_data['metadata']:
description_context = config_data['metadata']['description']
# Loop door elk veld in de configuratie
for field_name, field_data in fields.items():
# Vertaal name als het bestaat en niet leeg is
if 'name' in field_data and field_data['name']:
# Gebruik context indien opgegeven, anders description_context
field_context = context if context else description_context
translated_name = cache_manager.translation_cache.get_translation(
text=field_data['name'],
target_lang=target_language,
source_lang=source_language,
context=field_context
)
if translated_name:
translated_config[field_config][field_name]['name'] = translated_name.translated_text
# Loop door elk veld in de configuratie
for field_name, field_data in fields.items():
# Vertaal name als het bestaat en niet leeg is
if 'name' in field_data and field_data['name']:
# Gebruik context indien opgegeven, anders description_context
field_context = context if context else description_context
translated_name = cache_manager.translation_cache.get_translation(
text=field_data['name'],
target_lang=target_language,
source_lang=source_language,
context=field_context
)
if translated_name:
translated_config[field_config][field_name]['name'] = translated_name.translated_text
if 'title' in field_data and field_data['title']:
# Gebruik context indien opgegeven, anders description_context
field_context = context if context else description_context
translated_title = cache_manager.translation_cache.get_translation(
text=field_data['title'],
target_lang=target_language,
source_lang=source_language,
context=field_context
)
if translated_title:
translated_config[field_config][field_name]['title'] = translated_title.translated_text
if 'title' in field_data and field_data['title']:
# Gebruik context indien opgegeven, anders description_context
field_context = context if context else description_context
translated_title = cache_manager.translation_cache.get_translation(
text=field_data['title'],
target_lang=target_language,
source_lang=source_language,
context=field_context
)
if translated_title:
translated_config[field_config][field_name]['title'] = translated_title.translated_text
# Vertaal description als het bestaat en niet leeg is
if 'description' in field_data and field_data['description']:
# Gebruik context indien opgegeven, anders description_context
field_context = context if context else description_context
translated_desc = cache_manager.translation_cache.get_translation(
text=field_data['description'],
target_lang=target_language,
source_lang=source_language,
context=field_context
)
if translated_desc:
translated_config[field_config][field_name]['description'] = translated_desc.translated_text
# Vertaal description als het bestaat en niet leeg is
if 'description' in field_data and field_data['description']:
# Gebruik context indien opgegeven, anders description_context
field_context = context if context else description_context
translated_desc = cache_manager.translation_cache.get_translation(
text=field_data['description'],
target_lang=target_language,
source_lang=source_language,
context=field_context
)
if translated_desc:
translated_config[field_config][field_name]['description'] = translated_desc.translated_text
# Vertaal context als het bestaat en niet leeg is
if 'context' in field_data and field_data['context']:
translated_ctx = cache_manager.translation_cache.get_translation(
text=field_data['context'],
target_lang=target_language,
source_lang=source_language,
context=context
)
if translated_ctx:
translated_config[field_config][field_name]['context'] = translated_ctx.translated_text
# Vertaal context als het bestaat en niet leeg is
if 'context' in field_data and field_data['context']:
translated_ctx = cache_manager.translation_cache.get_translation(
text=field_data['context'],
target_lang=target_language,
source_lang=source_language,
context=context
)
if translated_ctx:
translated_config[field_config][field_name]['context'] = translated_ctx.translated_text
return translated_config
return translated_config
@staticmethod
def translate(text: str, target_language: str, source_language: Optional[str] = None,
def translate(tenant_id: int, text: str, target_language: str, source_language: Optional[str] = None,
context: Optional[str] = None)-> str:
with BusinessEvent('Translation Service', 0):
if current_event:
with current_event.create_span('Translation'):
translation_cache = cache_manager.translation_cache.get_translation(text, target_language,
source_language, context)
return translation_cache.translated_text
return translation_cache.translated_text
else:
with BusinessEvent('Translation Service', tenant_id):
with current_event.create_span('Translation'):
translation_cache = cache_manager.translation_cache.get_translation(text, target_language,
source_language, context)
return translation_cache.translated_text