82 lines
4.7 KiB
Python
82 lines
4.7 KiB
Python
"""ORM mapping between DeerFlow ownership and WeKnora knowledge bases."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import Boolean, Integer, String, Text, UniqueConstraint, false, true
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from deerflow.persistence.base import Base
|
|
from deerflow.persistence.types import BeijingDateTime, PortableJSON
|
|
|
|
|
|
class LlmWikiKnowledgeBaseRow(Base):
|
|
__tablename__ = "llmwiki_knowledge_bases"
|
|
|
|
id: Mapped[str] = mapped_column(String(36), primary_key=True)
|
|
weknora_id: Mapped[str] = mapped_column(String(128), unique=True, index=True)
|
|
owner_user_id: Mapped[str] = mapped_column(String(64), index=True)
|
|
name: Mapped[str] = mapped_column(String(255), default="")
|
|
description: Mapped[str] = mapped_column(String(2048), default="")
|
|
# ``None`` means "use the product defaults derived from the knowledge-base
|
|
# name". An explicit empty list is meaningful: the owner disabled starter
|
|
# questions for this library.
|
|
test_questions: Mapped[list | None] = mapped_column(PortableJSON(), nullable=True)
|
|
retrieval_settings: Mapped[dict | None] = mapped_column(PortableJSON(), nullable=True)
|
|
wiki_parent_directories: Mapped[list[str] | None] = mapped_column(PortableJSON(), nullable=True)
|
|
wiki_directory_paths: Mapped[list[list[str]] | None] = mapped_column(PortableJSON(), nullable=True)
|
|
wiki_focus_instructions: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
wiki_content_instructions_raw: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
wiki_extraction_instructions_raw: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
wiki_generation_managed: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
|
|
wiki_config_schema_version: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
wiki_config_revision: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
wiki_config_apply_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
|
|
wiki_config_apply_error: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
kb_type: Mapped[str] = mapped_column(String(32), default="document")
|
|
publication_status: Mapped[str] = mapped_column(String(16), default="private", index=True)
|
|
published_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
|
reviewed_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
|
published_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
|
|
wiki_index_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default=true(), index=True)
|
|
external_search_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=false(), index=True)
|
|
external_search_updated_at: Mapped[datetime | None] = mapped_column(BeijingDateTime(), nullable=True)
|
|
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
|
|
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
|
|
|
|
def to_dict(self) -> dict:
|
|
return {column.name: getattr(self, column.name) for column in self.__table__.columns}
|
|
|
|
|
|
class LlmWikiConversationDepositRow(Base):
|
|
"""Mapping from one DeerFlow answer turn to its WeKnora manual document.
|
|
|
|
The actual content lives in WeKnora. DeerFlow keeps only the stable ids it
|
|
needs to rewrite/delete the document when the user regenerates or removes a
|
|
conversation turn.
|
|
"""
|
|
|
|
__tablename__ = "llmwiki_conversation_deposits"
|
|
__table_args__ = (
|
|
UniqueConstraint(
|
|
"thread_id",
|
|
"assistant_message_id",
|
|
name="uq_llmwiki_conversation_deposit_answer",
|
|
),
|
|
)
|
|
|
|
id: Mapped[str] = mapped_column(String(36), primary_key=True)
|
|
thread_id: Mapped[str] = mapped_column(String(128), index=True)
|
|
human_message_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
|
|
assistant_message_id: Mapped[str] = mapped_column(String(128), index=True)
|
|
knowledge_base_mapping_id: Mapped[str] = mapped_column(String(36), index=True)
|
|
weknora_knowledge_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
|
|
title: Mapped[str] = mapped_column(String(255), default="")
|
|
created_by_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
|
|
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
|
|
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
|
|
|
|
def to_dict(self) -> dict:
|
|
return {column.name: getattr(self, column.name) for column in self.__table__.columns}
|