def chunking_by_token_size(
    tokenizer: Tokenizer,
    content: str,
    split_by_character: str | None = None,
    split_by_character_only: bool = False,
    overlap_token_size: int = 128,
    max_token_size: int = 1024,
) -> list[dict[str, Any]]:
    """
    Token-based chunking with configurable overlap
    
    Args:
        tokenizer: Tiktoken tokenizer instance
        content: Raw text content
        split_by_character: Optional character delimiter (e.g., '\n\n')
        overlap_token_size: Overlap between chunks
        max_token_size: Maximum tokens per chunk
    """
    tokens = tokenizer.encode(content)
    results: list[dict[str, Any]] = []
    
    if split_by_character:
        # Character-based splitting logic with token control
        raw_chunks = content.split(split_by_character)
        # ... processing logic
    else:
        # Pure token-based splitting with overlap
        for index, start in enumerate(
            range(0, len(tokens), max_token_size - overlap_token_size)
        ):
            chunk_content = tokenizer.decode(tokens[start : start + max_token_size])
            results.append({
                "tokens": min(max_token_size, len(tokens) - start),
                "content": chunk_content.strip(),
                "chunk_order_index": index,
            })
    
    return results

def _apply_chunk_template(
    self, content_type: str, original_item: Dict[str, Any], description: str
) -> str:
    """
    Apply specialized templates for different content types
    
    Ensures optimal chunking for images, tables, equations, code
    """
    templates = {
        "text": self.text_template,
        "image": self.image_template,        # 图像描述模板
        "table": self.table_template,        # 表格结构化模板
        "equation": self.equation_template,  # 公式语义化模板
        "code": self.code_template           # 代码块模板
    }
    
    template = templates.get(content_type, self.default_template)
    return template.format(
        content=original_item.get('content', ''), 
        description=description
    )

# 缓存机制优化性能
async def _get_from_parse_cache(self, cache_key: str, file_path: Path) -> Optional[tuple]:
    """Retrieve parsed content from cache to avoid redundant processing"""
    cache_file = self.cache_dir / f"{cache_key}.json"
    if cache_file.exists():
        try:
            with open(cache_file, 'r') as f:
                return json.load(f)
        except Exception as e:
            self.logger.warning(f"Cache read failed: {e}")
    return None

class Chunker:
    def __init__(
        self,
        tokenizer: BaseTokenizer,
        enable_multipass: bool = False,        # 多通道索引
        enable_large_chunks: bool = False,     # 大 chunk 支持
        enable_contextual_rag: bool = False,   # 上下文 RAG
        blurb_size: int = BLURB_SIZE,
        chunk_token_limit: int = 1024,
        chunk_overlap: int = 128,
        mini_chunk_size: int = 256,            # Mini-chunk 大小
    ) -> None:
        # Token 计数器
        def token_counter(text: str) -> int:
            return len(tokenizer.encode(text))
        
        # 主分块器：基于句子的语义边界
        self.chunk_splitter = SentenceChunker(
            tokenizer_or_token_counter=token_counter,
            chunk_size=chunk_token_limit,
            chunk_overlap=chunk_overlap,
            return_type="texts",
        )
        
        # Mini-chunk 分块器（用于多通道索引）
        self.mini_chunk_splitter = (
            SentenceChunker(
                tokenizer_or_token_counter=token_counter,
                chunk_size=mini_chunk_size,
                chunk_overlap=0,
                return_type="texts",
            )
            if enable_multipass
            else None
        )

import chonkie

# Token-based chunking
if chunk_strategy == "token":
    chunker = chonkie.TokenChunker(
        tokenizer=tokenizer_name_or_path, 
        chunk_size=chunk_size
    )

# Sentence-based chunking
elif chunk_strategy == "sentence":
    chunker = chonkie.SentenceChunker(
        tokenizer_or_token_counter=tokenizer_name_or_path, 
        chunk_size=chunk_size
    )

# Recursive chunking for structured content
elif chunk_strategy == "recursive":
    chunker = chonkie.RecursiveChunker(
        tokenizer=tokenizer_name_or_path,
        chunk_size=chunk_size,
        separators=["\n\n", "\n", " ", ""]
    )

# 根据 embedding 模型自动配置 chunk 大小
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL")
embedding_model_instance = AutoEmbeddings.get_embeddings(EMBEDDING_MODEL)

chunker_instance = RecursiveChunker(
    chunk_size=getattr(embedding_model_instance, "max_seq_length", 512)
)

code_chunker_instance = CodeChunker(
    chunk_size=getattr(embedding_model_instance, "max_seq_length", 512)
)

def preprocess_text(text: str) -> str:
    """文本预处理最佳实践"""
    # 1. 移除多余空白
    text = re.sub(r'\s+', ' ', text)
    
    # 2. 统一换行符
    text = text.replace('\r\n', '\n').replace('\r', '\n')
    
    # 3. 移除特殊控制字符
    text = ''.join(char for char in text if ord(char) >= 32 or char in '\n\t')
    
    # 4. 处理中英文混排空格
    text = re.sub(r'([一-龥])\s+([a-zA-Z])', r'\1 \2', text)
    text = re.sub(r'([a-zA-Z])\s+([一-龥])', r'\1 \2', text)
    
    return text.strip()

def chunk_with_structure_awareness(content: str, max_tokens: int = 1024):
    """保持 Markdown 结构的分块"""
    sections = []
    current_section = []
    current_tokens = 0
    
    for line in content.split('\n'):
        # 检测标题
        if line.startswith('#'):
            # 如果当前 section 不为空且即将超出，先保存
            if current_tokens > max_tokens * 0.8:
                sections.append('\n'.join(current_section))
                current_section = []
                current_tokens = 0
        
        line_tokens = len(tokenizer.encode(line))
        current_section.append(line)
        current_tokens += line_tokens
        
        # 超过限制则切分
        if current_tokens > max_tokens:
            sections.append('\n'.join(current_section))
            current_section = []
            current_tokens = 0
    
    # 保存最后一个 section
    if current_section:
        sections.append('\n'.join(current_section))
    
    return sections

def evaluate_chunking_quality(chunks: List[str]) -> Dict[str, float]:
    """分块质量评估指标"""
    metrics = {}
    
    # 1. Size 分布均匀度（标准差/平均值）
    sizes = [len(tokenizer.encode(c)) for c in chunks]
    metrics['size_cv'] = np.std(sizes) / np.mean(sizes)
    
    # 2. 语义连贯性（相邻 chunk embedding 相似度）
    embeddings = [get_embedding(c) for c in chunks]
    similarities = [
        cosine_similarity(embeddings[i], embeddings[i+1])
        for i in range(len(embeddings)-1)
    ]
    metrics['avg_coherence'] = np.mean(similarities)
    
    # 3. 覆盖率（chunk 边界是否覆盖关键信息）
    # 提取关键句并检查是否被完整包含在某个 chunk 中
    key_sentences = extract_key_sentences(original_text)
    coverage = sum(
        any(sent in chunk for chunk in chunks)
        for sent in key_sentences
    ) / len(key_sentences)
    metrics['key_coverage'] = coverage
    
    return metrics

def get_adaptive_chunk_size(doc_type: str, doc_length: int) -> int:
    """根据文档类型和长度自适应调整 chunk 大小"""
    base_size = 512
    
    if doc_type == "code":
        return 1024  # 代码需要更大上下文
    elif doc_type == "dialogue":
        return 256   # 对话可以较小
    elif doc_type == "academic":
        # 长文档用更大 chunk 减少总数
        return min(2048, base_size + (doc_length // 10000) * 128)
    
    return base_size

# 为混合检索优化：生成 chunk 变体
def create_chunk_variants(chunk: str) -> Dict[str, str]:
    """为混合检索创建 chunk 变体"""
    return {
        "original": chunk,
        "summary": summarize(chunk, max_length=128),  # 用于粗排
        "keywords": extract_keywords(chunk),          # 用于 BM25
        "questions": generate_questions(chunk)        # 用于 HyDE
    }