{
  "vector_id": "doc_123_chunk_5",
  "content": "机器学习是人工智能的分支...",
  "embedding": [0.12, -0.34, ...],  // 向量
  
  // 元数据
  "metadata": {
    // 基础元数据
    "source": "research_paper.pdf",
    "page": 5,
    "chunk_id": 5,
    "title": "Introduction to ML",
    "author": "Zhang San",
    
    // 业务元数据
    "category": "AI/Machine Learning",
    "language": "zh",
    "created_at": "2024-01-15",
    "updated_at": "2024-02-20",
    
    // 权限元数据
    "owner": "team_ai",
    "visibility": "internal",
    "access_level": 2,
    
    // 版本元数据
    "version": "v1.2",
    "is_latest": true,
    
    // 质量元数据
    "confidence_score": 0.92,
    "word_count": 237
  }
}

from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class BasicMetadata:
    """
    基础元数据（最小可行方案）
    
    适用场景：
    - 个人项目
    - 原型验证
    - 无复杂业务需求
    """
    # 必需字段
    chunk_id: str
    document_id: str
    content: str
    
    # 来源信息
    source_file: str
    page: Optional[int] = None
    
    # 时间信息
    created_at: datetime = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()
    
    def to_dict(self) -> dict:
        """转为字典（存储到向量数据库）"""
        return {
            "chunk_id": self.chunk_id,
            "document_id": self.document_id,
            "content": self.content,
            "source_file": self.source_file,
            "page": self.page,
            "created_at": self.created_at.isoformat()
        }
    
    @classmethod
    def from_dict(cls, data: dict) -> "BasicMetadata":
        """从字典创建"""
        data["created_at"] = datetime.fromisoformat(data["created_at"])
        return cls(**data)

# 使用示例
metadata = BasicMetadata(
    chunk_id="doc_1_chunk_5",
    document_id="doc_1",
    content="机器学习是...",
    source_file="ml_intro.pdf",
    page=5
)

# 存储到向量数据库
vector_store.add(
    vector=embedding,
    metadata=metadata.to_dict()
)

# 检索时过滤
results = vector_store.search(
    query_vector=query_embedding,
    filter={"source_file": "ml_intro.pdf"}  # 只搜索特定文件
)

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional, List, Dict
import json

class AccessLevel(Enum):
    """访问级别"""
    PUBLIC = 0
    INTERNAL = 1
    CONFIDENTIAL = 2
    RESTRICTED = 3

class DocumentStatus(Enum):
    """文档状态"""
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"
    DELETED = "deleted"

@dataclass
class EnterpriseMetadata:
    """
    企业级元数据
    
    特性：
    - 权限控制（RBAC）
    - 版本管理
    - 审计日志
    - 多维度标签
    """
    # 基础标识
    chunk_id: str
    document_id: str
    content: str
    
    # 来源信息
    source_file: str
    source_type: str  # pdf / docx / webpage / email
    page: Optional[int] = None
    section: Optional[str] = None
    
    # 内容元数据
    title: Optional[str] = None
    author: Optional[str] = None
    language: str = "en"
    word_count: int = 0
    
    # 业务元数据
    category: Optional[str] = None
    tags: List[str] = field(default_factory=list)
    department: Optional[str] = None
    project: Optional[str] = None
    
    # 权限元数据
    owner: str = None
    access_level: AccessLevel = AccessLevel.INTERNAL
    allowed_groups: List[str] = field(default_factory=list)
    allowed_users: List[str] = field(default_factory=list)
    
    # 版本元数据
    version: str = "v1.0"
    is_latest: bool = True
    parent_version: Optional[str] = None
    status: DocumentStatus = DocumentStatus.PUBLISHED
    
    # 时间元数据
    created_at: datetime = None
    updated_at: datetime = None
    indexed_at: datetime = None
    expires_at: Optional[datetime] = None
    
    # 质量元数据
    confidence_score: float = 1.0
    relevance_score: Optional[float] = None
    
    # 审计元数据
    created_by: str = None
    updated_by: str = None
    index_method: str = "auto"
    
    # 扩展元数据（业务自定义）
    custom_fields: Dict[str, any] = field(default_factory=dict)
    
    def __post_init__(self):
        """初始化时间字段"""
        now = datetime.now()
        if self.created_at is None:
            self.created_at = now
        if self.updated_at is None:
            self.updated_at = now
        if self.indexed_at is None:
            self.indexed_at = now
    
    def has_access(self, user_id: str, user_groups: List[str]) -> bool:
        """
        检查用户是否有访问权限
        
        Args:
            user_id: 用户ID
            user_groups: 用户所属组列表
        
        Returns:
            True if user has access
        """
        # 1. 检查是否是 owner
        if self.owner == user_id:
            return True
        
        # 2. 检查用户是否在允许列表
        if user_id in self.allowed_users:
            return True
        
        # 3. 检查用户组是否在允许列表
        if any(group in self.allowed_groups for group in user_groups):
            return True
        
        # 4. PUBLIC 文档所有人可访问
        if self.access_level == AccessLevel.PUBLIC:
            return True
        
        return False
    
    def is_expired(self) -> bool:
        """检查文档是否过期"""
        if self.expires_at is None:
            return False
        return datetime.now() > self.expires_at
    
    def to_dict(self) -> dict:
        """转为字典（存储）"""
        return {
            # 基础
            "chunk_id": self.chunk_id,
            "document_id": self.document_id,
            "content": self.content,
            
            # 来源
            "source_file": self.source_file,
            "source_type": self.source_type,
            "page": self.page,
            "section": self.section,
            
            # 内容
            "title": self.title,
            "author": self.author,
            "language": self.language,
            "word_count": self.word_count,
            
            # 业务
            "category": self.category,
            "tags": self.tags,
            "department": self.department,
            "project": self.project,
            
            # 权限
            "owner": self.owner,
            "access_level": self.access_level.value,
            "allowed_groups": self.allowed_groups,
            "allowed_users": self.allowed_users,
            
            # 版本
            "version": self.version,
            "is_latest": self.is_latest,
            "parent_version": self.parent_version,
            "status": self.status.value,
            
            # 时间
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
            "indexed_at": self.indexed_at.isoformat(),
            "expires_at": self.expires_at.isoformat() if self.expires_at else None,
            
            # 质量
            "confidence_score": self.confidence_score,
            "relevance_score": self.relevance_score,
            
            # 审计
            "created_by": self.created_by,
            "updated_by": self.updated_by,
            "index_method": self.index_method,
            
            # 自定义
            "custom_fields": self.custom_fields
        }
    
    def to_filter_dict(self) -> dict:
        """
        转为过滤条件字典（仅包含可过滤字段）
        """
        return {
            "document_id": self.document_id,
            "source_type": self.source_type,
            "category": self.category,
            "tags": self.tags,
            "language": self.language,
            "access_level": self.access_level.value,
            "status": self.status.value,
            "is_latest": self.is_latest
        }

# 使用示例

# 创建企业级元数据
metadata = EnterpriseMetadata(
    chunk_id="doc_123_chunk_5",
    document_id="doc_123",
    content="机器学习是...",
    source_file="ml_research.pdf",
    source_type="pdf",
    page=5,
    title="Machine Learning Introduction",
    author="Zhang San",
    language="zh",
    category="AI/Machine Learning",
    tags=["ML", "AI", "Tutorial"],
    department="Research",
    owner="user_456",
    access_level=AccessLevel.INTERNAL,
    allowed_groups=["team_ai", "team_research"],
    version="v1.0",
    created_by="user_456"
)

# 权限检查
user_has_access = metadata.has_access(
    user_id="user_789",
    user_groups=["team_ai"]
)

# 存储
vector_store.add(
    vector=embedding,
    metadata=metadata.to_dict()
)

# 带权限过滤的检索
results = vector_store.search(
    query_vector=query_embedding,
    filter={
        "status": "published",
        "is_latest": True,
        "allowed_groups": {"$in": user_groups}
    }
)

from dataclasses import dataclass
from typing import Optional, Dict, List
from enum import Enum

class ContentType(Enum):
    """内容类型"""
    TEXT = "text"
    IMAGE = "image"
    VIDEO = "video"
    AUDIO = "audio"
    TABLE = "table"
    CODE = "code"

@dataclass
class MultimodalMetadata:
    """
    多模态元数据
    
    支持：
    - 文本、图像、视频、音频
    - 不同类型内容的特定元数据
    """
    # 基础
    chunk_id: str
    document_id: str
    content_type: ContentType
    
    # 通用元数据
    source_file: str
    created_at: str
    
    # 文本特定元数据
    text_content: Optional[str] = None
    word_count: Optional[int] = None
    language: Optional[str] = None
    
    # 图像特定元数据
    image_path: Optional[str] = None
    image_caption: Optional[str] = None
    image_width: Optional[int] = None
    image_height: Optional[int] = None
    image_type: Optional[str] = None  # photo / chart / diagram / screenshot
    
    # 视频特定元数据
    video_path: Optional[str] = None
    video_duration: Optional[float] = None  # 秒
    video_fps: Optional[float] = None
    keyframe_timestamp: Optional[float] = None  # 关键帧时间戳
    video_transcript: Optional[str] = None
    
    # 音频特定元数据
    audio_path: Optional[str] = None
    audio_duration: Optional[float] = None
    audio_transcript: Optional[str] = None
    audio_language: Optional[str] = None
    
    # 表格特定元数据
    table_data: Optional[List[List[str]]] = None
    table_headers: Optional[List[str]] = None
    table_rows: Optional[int] = None
    table_cols: Optional[int] = None
    
    # 代码特定元数据
    code_content: Optional[str] = None
    code_language: Optional[str] = None
    code_line_start: Optional[int] = None
    code_line_end: Optional[int] = None
    
    def to_dict(self) -> dict:
        """转为字典"""
        base = {
            "chunk_id": self.chunk_id,
            "document_id": self.document_id,
            "content_type": self.content_type.value,
            "source_file": self.source_file,
            "created_at": self.created_at
        }
        
        # 根据类型添加特定字段
        if self.content_type == ContentType.TEXT:
            base.update({
                "text_content": self.text_content,
                "word_count": self.word_count,
                "language": self.language
            })
        
        elif self.content_type == ContentType.IMAGE:
            base.update({
                "image_path": self.image_path,
                "image_caption": self.image_caption,
                "image_width": self.image_width,
                "image_height": self.image_height,
                "image_type": self.image_type
            })
        
        elif self.content_type == ContentType.VIDEO:
            base.update({
                "video_path": self.video_path,
                "video_duration": self.video_duration,
                "video_fps": self.video_fps,
                "keyframe_timestamp": self.keyframe_timestamp,
                "video_transcript": self.video_transcript
            })
        
        elif self.content_type == ContentType.AUDIO:
            base.update({
                "audio_path": self.audio_path,
                "audio_duration": self.audio_duration,
                "audio_transcript": self.audio_transcript,
                "audio_language": self.audio_language
            })
        
        elif self.content_type == ContentType.TABLE:
            base.update({
                "table_data": self.table_data,
                "table_headers": self.table_headers,
                "table_rows": self.table_rows,
                "table_cols": self.table_cols
            })
        
        elif self.content_type == ContentType.CODE:
            base.update({
                "code_content": self.code_content,
                "code_language": self.code_language,
                "code_line_start": self.code_line_start,
                "code_line_end": self.code_line_end
            })
        
        return base

# 使用示例

# 文本 chunk
text_metadata = MultimodalMetadata(
    chunk_id="text_1",
    document_id="doc_1",
    content_type=ContentType.TEXT,
    source_file="paper.pdf",
    created_at="2024-01-15",
    text_content="机器学习是...",
    word_count=237,
    language="zh"
)

# 图像 chunk
image_metadata = MultimodalMetadata(
    chunk_id="image_1",
    document_id="doc_1",
    content_type=ContentType.IMAGE,
    source_file="paper.pdf",
    created_at="2024-01-15",
    image_path="/images/figure_1.png",
    image_caption="CNN architecture diagram",
    image_width=1920,
    image_height=1080,
    image_type="diagram"
)

# 视频 chunk（关键帧）
video_metadata = MultimodalMetadata(
    chunk_id="video_1_frame_10",
    document_id="video_1",
    content_type=ContentType.VIDEO,
    source_file="lecture.mp4",
    created_at="2024-01-15",
    video_path="/videos/lecture.mp4",
    video_duration=3600.0,
    video_fps=30.0,
    keyframe_timestamp=300.0,  # 5分钟处的关键帧
    video_transcript="In this section, we discuss..."
)

class DocumentVersionManager:
    """
    文档版本管理器
    
    功能：
    - 追踪文档历史版本
    - 总是使用最新版本
    - 支持回滚
    """
    
    def __init__(self, vector_store):
        self.vector_store = vector_store
    
    def add_document_version(
        self,
        document_id: str,
        content: str,
        version: str,
        updated_by: str
    ):
        """
        添加新版本
        
        策略：
        1. 标记旧版本为 is_latest=False
        2. 添加新版本 is_latest=True
        3. 记录父版本
        """
        # 1. 查找当前最新版本
        current_version = self.vector_store.search(
            filter={
                "document_id": document_id,
                "is_latest": True
            }
        )
        
        # 2. 标记旧版本
        if current_version:
            for old_doc in current_version:
                self.vector_store.update(
                    id=old_doc["id"],
                    metadata={"is_latest": False}
                )
        
        # 3. 添加新版本
        new_metadata = {
            "document_id": document_id,
            "version": version,
            "is_latest": True,
            "parent_version": current_version[0]["version"] if current_version else None,
            "updated_by": updated_by,
            "updated_at": datetime.now().isoformat()
        }
        
        # 生成 embedding 并添加
        embedding = self.generate_embedding(content)
        self.vector_store.add(
            vector=embedding,
            metadata=new_metadata
        )
    
    def get_latest_version(self, document_id: str):
        """获取最新版本"""
        results = self.vector_store.search(
            filter={
                "document_id": document_id,
                "is_latest": True
            }
        )
        return results[0] if results else None
    
    def get_version_history(self, document_id: str) -> List[dict]:
        """获取版本历史"""
        results = self.vector_store.search(
            filter={"document_id": document_id}
        )
        # 按版本排序
        return sorted(results, key=lambda x: x["metadata"]["updated_at"], reverse=True)
    
    def rollback_to_version(self, document_id: str, target_version: str):
        """回滚到指定版本"""
        # 1. 标记所有版本为非最新
        all_versions = self.get_version_history(document_id)
        for version in all_versions:
            self.vector_store.update(
                id=version["id"],
                metadata={"is_latest": False}
            )
        
        # 2. 标记目标版本为最新
        target = [v for v in all_versions if v["metadata"]["version"] == target_version][0]
        self.vector_store.update(
            id=target["id"],
            metadata={"is_latest": True}
        )

class SmartTaggingSystem:
    """
    智能标签系统
    
    功能：
    - 自动生成标签
    - 标签层级结构
    - 标签推荐
    """
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.tag_hierarchy = self._init_tag_hierarchy()
    
    def _init_tag_hierarchy(self) -> Dict:
        """初始化标签层级"""
        return {
            "Technology": {
                "AI": ["Machine Learning", "Deep Learning", "NLP", "Computer Vision"],
                "Web": ["Frontend", "Backend", "DevOps"],
                "Mobile": ["iOS", "Android", "React Native"]
            },
            "Business": {
                "Finance": ["Accounting", "Investment", "Tax"],
                "Marketing": ["SEO", "Social Media", "Content Marketing"]
            },
            "Science": {
                "Physics": ["Quantum", "Classical"],
                "Biology": ["Genetics", "Ecology"]
            }
        }
    
    async def auto_tag(self, content: str, max_tags: int = 5) -> List[str]:
        """
        自动生成标签
        
        策略：
        1. 用 LLM 提取关键主题
        2. 匹配到标签层级
        3. 返回最相关的标签
        """
        prompt = f"""Extract up to {max_tags} relevant tags from the following text.

Text: {content[:500]}

Available tags:
{self._format_tag_hierarchy()}

Output only the tags, one per line.
"""
        
        response = await self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        
        tags = response.choices[0].message.content.strip().split("\n")
        return [tag.strip() for tag in tags if tag.strip()]
    
    def _format_tag_hierarchy(self) -> str:
        """格式化标签层级（用于提示词）"""
        lines = []
        for category, subcategories in self.tag_hierarchy.items():
            lines.append(f"- {category}")
            for subcat, tags in subcategories.items():
                lines.append(f"  - {subcat}: {', '.join(tags)}")
        return "\n".join(lines)
    
    def recommend_tags(self, existing_tags: List[str]) -> List[str]:
        """
        推荐相关标签
        
        策略：找到同一类别下的其他标签
        """
        recommendations = []
        
        for category, subcategories in self.tag_hierarchy.items():
            for subcat, tags in subcategories.items():
                # 如果已有标签在这个子类别中
                if any(tag in tags for tag in existing_tags):
                    # 推荐同子类别的其他标签
                    recommendations.extend([t for t in tags if t not in existing_tags])
        
        return recommendations[:5]

# 使用示例
tagging_system = SmartTaggingSystem(llm_client)

content = "这篇文章介绍了深度学习在计算机视觉中的应用..."
tags = await tagging_system.auto_tag(content)
# Output: ["AI", "Deep Learning", "Computer Vision"]

# 推荐相关标签
recommended = tagging_system.recommend_tags(tags)
# Output: ["Machine Learning", "NLP"]

from pydantic import BaseModel, validator, Field
from datetime import datetime
from typing import Optional, List

class MetadataSchema(BaseModel):
    """
    元数据验证模式（使用 Pydantic）
    
    优势：
    - 类型检查
    - 数据验证
    - 自动文档生成
    """
    chunk_id: str = Field(..., min_length=1, max_length=100)
    document_id: str = Field(..., min_length=1, max_length=100)
    content: str = Field(..., min_length=1)
    
    source_file: str
    page: Optional[int] = Field(None, ge=1)
    
    category: Optional[str] = Field(None, max_length=50)
    tags: List[str] = Field(default_factory=list, max_items=10)
    
    language: str = Field("en", regex="^[a-z]{2}$")  # ISO 639-1
    word_count: int = Field(0, ge=0)
    
    created_at: datetime
    confidence_score: float = Field(1.0, ge=0.0, le=1.0)
    
    @validator("tags")
    def validate_tags(cls, v):
        """验证标签（去重、转小写）"""
        return list(set(tag.lower() for tag in v))
    
    @validator("source_file")
    def validate_source_file(cls, v):
        """验证文件路径"""
        if not v.endswith(('.pdf', '.docx', '.txt', '.md')):
            raise ValueError("Unsupported file type")
        return v
    
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

# 使用示例

# 有效元数据
valid_metadata = MetadataSchema(
    chunk_id="doc_1_chunk_1",
    document_id="doc_1",
    content="Some text...",
    source_file="document.pdf",
    page=5,
    tags=["AI", "ML", "AI"],  # 会自动去重
    language="zh",
    word_count=100,
    created_at=datetime.now()
)

# 无效元数据（会抛出验证错误）
try:
    invalid_metadata = MetadataSchema(
        chunk_id="",  # 太短
        document_id="doc_1",
        content="Text",
        source_file="file.unknown",  # 不支持的格式
        page=-1,  # 负数
        created_at=datetime.now()
    )
except ValueError as e:
    print(f"Validation error: {e}")

# Qdrant 风格的复杂查询

# 1. AND 查询
filter = {
    "must": [
        {"key": "category", "match": {"value": "AI"}},
        {"key": "language", "match": {"value": "zh"}},
        {"key": "is_latest", "match": {"value": True}}
    ]
}

# 2. OR 查询
filter = {
    "should": [
        {"key": "tags", "match": {"any": ["ML", "DL"]}},
        {"key": "category", "match": {"value": "AI"}}
    ]
}

# 3. 范围查询
filter = {
    "must": [
        {"key": "created_at", "range": {
            "gte": "2024-01-01",
            "lte": "2024-12-31"
        }},
        {"key": "confidence_score", "range": {
            "gte": 0.8
        }}
    ]
}

# 4. 嵌套查询
filter = {
    "must": [
        {"key": "status", "match": {"value": "published"}},
        {
            "should": [
                {"key": "access_level", "match": {"value": 0}},  # PUBLIC
                {"key": "allowed_groups", "match": {"any": user_groups}}
            ]
        }
    ]
}

# 5. NOT 查询
filter = {
    "must": [
        {"key": "is_latest", "match": {"value": True}}
    ],
    "must_not": [
        {"key": "status", "match": {"value": "archived"}},
        {"key": "tags", "match": {"any": ["deprecated", "obsolete"]}}
    ]
}