Function create_knowledge_record

backend/database/knowledge_db.py:25–89

Create a knowledge base record Args: query: Dictionary containing all knowledge base data, must include: - index_name: Knowledge base name - knowledge_describe: Knowledge base description - knowledge_status: Knowledge base status - us

(query: Dict[str, Any])

def create_knowledge_record(query: Dict[str, Any]) -> Dict[str, Any]:
    """
    Create a knowledge base record

    Args:
        query: Dictionary containing all knowledge base data, must include:
            - index_name: Knowledge base name
            - knowledge_describe: Knowledge base description
            - knowledge_status: Knowledge base status
            - user_id: Optional user ID for created_by and updated_by fields
            - tenant_id: Optional tenant ID for created_by and updated_by fields
            - embedding_model_name: embedding model name for the knowledge base
            - preserve_source_file: whether to preserve uploaded source documents (optional)

    Returns:
        Dict[str, Any]: Dictionary with at least 'knowledge_id' and 'index_name'
    """
    try:
        with get_db_session() as session:
            # Determine user-facing knowledge base name
            knowledge_name = query.get(
                "knowledge_name") or query.get("index_name")

            # Prepare data dictionary
            group_ids = query.get("group_ids")
            data: Dict[str, Any] = {
                "knowledge_describe": query.get("knowledge_describe", ""),
                "created_by": query.get("user_id"),
                "updated_by": query.get("user_id"),
                "knowledge_sources": query.get("knowledge_sources", "elasticsearch"),
                "tenant_id": query.get("tenant_id"),
                "embedding_model_name": query.get("embedding_model_name"),
                "embedding_model_id": query.get("embedding_model_id"),
                "knowledge_name": knowledge_name,
                "group_ids": convert_list_to_string(group_ids) if isinstance(group_ids, list) else group_ids,
                "ingroup_permission": query.get("ingroup_permission"),
                "preserve_source_file": query.get("preserve_source_file", True),
            }

            # For backward compatibility: if caller explicitly provides index_name,
            # respect it and do not regenerate; otherwise generate after flush.
            explicit_index_name = query.get("index_name")
            if explicit_index_name:
                data["index_name"] = explicit_index_name

            # Create new record
            new_record = KnowledgeRecord(**data)
            session.add(new_record)
            session.flush()

            # Generate internal index_name for new records when not explicitly provided
            if not explicit_index_name:
                generated_index_name = _generate_index_name(
                    new_record.knowledge_id)
                new_record.index_name = generated_index_name
                session.flush()

            session.commit()