def knowledge_search_workflow(
request: WSGIRequest,
user_input: str,
message_history: MessageHistory | None = None,
max_results: int = 10,
use_llm_preprocessing: bool = True,
) -> BaseMessageIntel | None:
from django_spire.knowledge.entry.version.block.models import EntryVersionBlock
entries = list(
Entry.services.search
.search(
query=user_input,
use_llm_preprocessing=use_llm_preprocessing,
)
.user_has_access(user=request.user)
.select_related('current_version', 'current_version__author', 'collection')
.prefetch_related(
Prefetch(
'current_version__blocks',
queryset=EntryVersionBlock.objects.active().order_by('order')
)
)[:max_results]
)
if not entries:
return NO_KNOWLEDGE_MESSAGE_INTEL
answer_intel_future = KnowledgeAnswerBot(llm_temperature=0.5).process_to_future(
user_input=user_input,
entries=entries,
message_history=message_history,
)
entries_intel_future = KnowledgeEntriesBot(llm_temperature=0.5).process_to_future(
user_input=user_input,
entries=entries
)
return KnowledgeMessageIntel(
answer_intel=answer_intel_future.result,
entries_intel=entries_intel_future.result,
)