ai-ticket-platform/backend/apps/knowledge/services.py

import hashlib
import math

from apps.knowledge.models import KnowledgeBase, KnowledgeChunk, KnowledgeDocument, KnowledgeItem
from apps.solutions.models import Solution

DEFAULT_CHUNK_SIZE = 600


def split_text(content: str, chunk_size: int = DEFAULT_CHUNK_SIZE) -> list[str]:
    normalized = "\n".join(line.strip() for line in content.splitlines() if line.strip())
    if not normalized:
        return []
    return [
        normalized[index : index + chunk_size]
        for index in range(0, len(normalized), chunk_size)
    ]


def embed_text(content: str, dimensions: int = 16) -> list[float]:
    digest = hashlib.sha256(content.encode("utf-8")).digest()
    values = [digest[index] / 255 for index in range(dimensions)]
    length = math.sqrt(sum(value * value for value in values)) or 1
    return [round(value / length, 6) for value in values]


def keyword_score(query: str, content: str) -> float:
    normalized_query = query.lower()
    terms = [term for term in normalized_query.split() if term]
    terms.extend(character for character in normalized_query if not character.isspace())
    lowered = content.lower()
    return sum(1 for term in terms if term in lowered)


def vector_score(left: list[float], right: list[float]) -> float:
    return sum(a * b for a, b in zip(left, right, strict=False))


def rebuild_item_chunks(item: KnowledgeItem) -> list[KnowledgeChunk]:
    item.chunks.all().delete()
    content = "\n".join(
        [
            item.title,
            item.problem,
            item.applicable_scope,
            item.error_code,
            item.root_cause,
            item.troubleshooting_steps,
            item.solution,
            item.risk_note,
        ]
    )
    chunks = []
    for index, chunk_text in enumerate(split_text(content)):
        chunks.append(
            KnowledgeChunk.objects.create(
                knowledge_base=item.knowledge_base,
                item=item,
                chunk_index=index,
                content=chunk_text,
                embedding=embed_text(chunk_text),
                metadata={
                    "source_type": "knowledge_item",
                    "title": item.title,
                    "item_id": item.id,
                },
            )
        )
    return chunks


def ingest_document_text(
    knowledge_base_id: int,
    file_name: str,
    file_type: str,
    content: str,
    operator=None,
) -> KnowledgeDocument:
    document = KnowledgeDocument.objects.create(
        knowledge_base_id=knowledge_base_id,
        file_name=file_name,
        file_type=file_type,
        content_text=content,
        parse_status="success",
        created_by=operator,
    )
    for index, chunk_text in enumerate(split_text(content)):
        KnowledgeChunk.objects.create(
            knowledge_base=document.knowledge_base,
            document=document,
            chunk_index=index,
            content=chunk_text,
            embedding=embed_text(chunk_text),
            metadata={
                "source_type": "document",
                "title": document.file_name,
                "document_id": document.id,
            },
        )
    return document


def index_published_solution(solution: Solution) -> list[KnowledgeChunk]:
    base, _ = KnowledgeBase.objects.get_or_create(
        name="解决方案知识库",
        defaults={"description": "从已发布解决方案生成的检索知识", "scope": "solutions"},
    )
    KnowledgeChunk.objects.filter(metadata__solution_id=solution.id).delete()
    content = "\n".join(
        [
            solution.title,
            solution.problem_signature,
            solution.related_system,
            solution.environment_scope,
            solution.root_cause,
            solution.solution,
            solution.risk_note,
        ]
    )
    chunks = []
    for index, chunk_text in enumerate(split_text(content)):
        chunks.append(
            KnowledgeChunk.objects.create(
                knowledge_base=base,
                chunk_index=index,
                content=chunk_text,
                embedding=embed_text(chunk_text),
                metadata={
                    "source_type": "solution",
                    "title": solution.title,
                    "solution_id": solution.id,
                },
            )
        )
    return chunks


def retrieve_knowledge(
    query: str,
    knowledge_base_id: int | None = None,
    limit: int = 5,
) -> list[dict]:
    query_embedding = embed_text(query)
    queryset = KnowledgeChunk.objects.select_related("knowledge_base", "item", "document")
    if knowledge_base_id:
        queryset = queryset.filter(knowledge_base_id=knowledge_base_id)
    scored = []
    for chunk in queryset[:1000]:
        score = keyword_score(query, chunk.content) * 10 + vector_score(
            query_embedding,
            chunk.embedding,
        )
        if score > 0:
            scored.append((score, chunk))
    scored.sort(key=lambda pair: pair[0], reverse=True)
    return [
        {
            "id": chunk.id,
            "knowledge_base_id": chunk.knowledge_base_id,
            "title": chunk.metadata.get("title", chunk.knowledge_base.name),
            "source_type": chunk.metadata.get("source_type", "chunk"),
            "content": chunk.content,
            "score": round(score, 4),
            "metadata": chunk.metadata,
        }
        for score, chunk in scored[:limit]
    ]