ngb MCP server corpus phase 4 fasttext embeddings

FastText集成：单词级语义分析子阶段文档

0.执行摘要和愿景

该文件详细介绍了Tibetan佛教语料库复杂的单词级语义分析能力的成功实现。这个子阶段是OpenSearch上托管的现有块级检索系统的并行和互补支柱。

虽然块级系统旨在回答这个问题，”为我查找关于某个概念的文档“这个新的单词级系统回答了同样重要的学术问题，”**什么是概念 *意思是* 在特定的背景下，它与其他概念有什么关系？**"

我们已成功培养了一批专业人才 fastText 单词嵌入模型，每个专业一个 L2_corpus_section （例如，Madhyamaka、Vinaya、Atiyyoga）。这些模型将单词转换为高维向量，根据它们在特定文本传统中的使用来捕捉它们微妙的语义关系。

最后的工件——一组总计超过200GB的模型文件已经安全地放入一个专用的AWS S3存储桶中，准备集成到生产API中。本文档提供了对愿景、实现过程、创建的工具以及为实现这一结果而执行的命令的完整演练。

1.愿景：新的学术能力

训练好的模型不仅仅是数据；它们是一种新的学术视角。它们解锁了几个强大的分析功能，可以通过 explore_model.py 工具包。

1.1.探索概念邻域（最近邻）

这是最基本的能力：在特定上下文中发现哪些单词在语义上最接近目标单词。它允许我们映射术语的“语义场”。

示例查询： 比较含义 སེམས་ （sems，“心灵”）跨越不同的哲学流派。

• 在Madhyamaka（中道学校）：

  python src/explore_model.py neighbors middle_way སེམས་

- 预期见解： 邻居可能会包括分析和学术术语，如 ཡིད་ （manas，心理咨询师）， རྣམ་ཤེས་ （意识，意识）和 བདག་ （ātman，self），反映了该学派对解构自我和意识本质的关注。

• 在《阿底瑜伽》（Dzogchen）中：

  python src/explore_model.py neighbors atiyoga_mind_series_all_creating_king_cycle སེམས་

- 预期见解： 邻居们会转向沉思和体验的术语，比如 རིག་པ་ （rigpa，纯粹意识）， ཀུན་གཞི་ （昆芝，万物之本），以及 ཡེ་ཤེས་ （jñāna，原始智慧），揭示了一系列不同的关注点和关系。

1.2.揭示关系逻辑（类比）

这个强大的功能测试了模型对 *关系* 概念之间。它可以解决“A对B，就像C对X一样。”

示例查询： 测试模型对核心哲学二分法的把握。

• 查询： “在Madhyamaka语料库中， emptiness (སྟོང་པ་ཉིད་)与之相反， inherent existence (རང་བཞིན་)，如 selflessness (བདག་མེད་)是什么？"

  python src/explore_model.py analogy middle_way སྟོང་པ་ཉིད་ རང་བཞིན་ བདག་མེད་

• 预期见解： 一个训练有素的模特应该回来 བདག་ （bdag，“自我”）作为首要结果，表明它已经学会了Madhyamaka思想中这两个基本否定的平行结构。

1.3.量化语义距离（相似性评分）

这允许通过测量给定上下文中两个术语之间的精确相似性得分（从-1到1）来进行直接的定量假设检验。

示例查询： 检验“依赖性生起”在Madhyamaka的“空性”中比其他学派更为核心的假设。

• 查询：

  python src/explore_model.py similarity middle_way སྟོང་པ་ཉིད་ རྟེན་འབྲེལ་
  python src/explore_model.py similarity epistemology_and_logic སྟོང་པ་ཉིད་ རྟེན་འབྲེལ་

• 预期见解： 相似性得分来自 middle_way 模型的得分应明显高于 epistemology_and_logic （普拉玛ṇa） 该模型为一项著名的学术观察提供了数据驱动的证据。

1.4.识别概念连贯性（异常检测）

此函数通过识别列表中的“奇数”来测试模型理解语义类别边界的能力。

示例查询： Atiyoga模型是否理解其核心术语的含义？

• 查询： “从列表中 [རིག་པ་, ཀ་དག་, ལྷུན་གྲུབ་, ཚད་མ་]，该术语不属于 *昆杰嘉波*?"

  python src/explore_model.py outlier atiyoga_mind_series_all_creating_king_cycle རིག་པ་ ཀ་དག་ ལྷུན་གྲུབ་ ཚད་མ་

• 预期见解： 模型应正确识别 ཚད་མ་ （pramāṇa、 有效认知）作为异常值。其他三个术语--rigpa （意识）， kadag （原始纯度），以及 lhundrup （自发存在）——在大圆满哲学的核心形成一个紧密的概念集群。

2.技术架构

该系统旨在实现可扩展性，并遵循现代MLOps原则。

1. 数据来源： 这 canonical_chunks.parquet 该文件由项目的主要指纹识别管道生成，作为基本事实。
2. 语料库生成： Python脚本(prepare_word_embedding_data.py)处理规范块，按以下方式对其进行分组 L2_corpus_section 创建大型、干净的文本文件用于培训。
3. 模型培训： 第二个脚本(train_word_models.py)使用 fastText 图书馆培训a .bin 每个语料库的模型文件。为了提高速度，这个过程在CPU内核之间高度并行化。
4. 永久存储： 决赛 .bin 和 .json 模型工件存储在版本化的、安全的 AWS S3存储桶。这是模型的永久、坚不可摧的“仓库”。
5. 应用程序服务（未来步骤）： 部署在Render.com等云服务上的FastAPI应用程序将包含 ModelManager 类。此管理器将按需从S3下载模型，并将其缓存在服务器的RAM中，以便在后续请求时立即访问。

3.实施演练

本节详细介绍了构建和部署工件所采取的确切步骤。

3.1.环境设置

所有操作都在专用的Anaconda环境中进行，以确保可重复性。

1. 环境名称： tibetfp
2. 激活命令：

   conda activate tibetfp

1. 已安装的依赖项： 以下库已添加到现有环境中。

   pip install fasttext-wheel
   pip install psutil
   pip install numpy scipy

3.2.第一阶段：语料库准备

目标： 为每个对象创建一个大的、干净的、词形化的文本文件 L2_corpus_section.

理论基础： L2_corpus_section 被选中了 L3 确保每个训练语料库都足够大，以保持统计稳健性，与项目现有的数据分层保持一致。使用预先计算的 text_for_lda 柱提供干净、空间分隔的膜，非常适合 fastText.

脚本： src/prepare_word_embedding_data.py

# src/prepare_word_embedding_data.py (v3 - Robust Sanitization)

import pandas as pd
from pathlib import Path
import typer
from rich.console import Console
import logging
import unicodedata
import re

logging.basicConfig(level="INFO", format="%(message)s", handlers=[])
logger = logging.getLogger(__name__)
console = Console()

app = typer.Typer(help="Prepares text corpora for fastText word embedding training based on L2 sections.")

def sanitize_filename(name: str) -> str:
    """Aggressively sanitizes a string to be a safe, pure-ASCII filename."""
    if not isinstance(name, str): return ""
    nfkd_form = unicodedata.normalize('NFKD', name)
    ascii_string = nfkd_form.encode('ASCII', 'ignore').decode('utf-8')
    safe_string = re.sub(r'[^a-zA-Z0-9]+', ' ', ascii_string)
    return safe_string.strip().replace(' ', '_').lower()

@app.command()
def create_corpora(
    chunks_path: Path = typer.Option("output/pre_analysis/canonical_chunks.parquet", help="Path to the canonical chunk file."),
    catalog_path: Path = typer.Option("data/Corpus_Catalog_v3.csv", help="Path to the main corpus catalog to get L2 section metadata."),
    output_dir: Path = typer.Option("output/word_embedding_training/", help="Directory to save the training .txt files."),
    min_chunks_per_section: int = typer.Option(50, help="Minimum number of chunks required to create a corpus for an L2 section.")
):
    """Groups chunks by L2_corpus_section and concatenates their lemmatized text to create one large training file per section."""
    console.print(f"🚀 Starting Word Embedding Corpus Preparation (L2 Section Based)...")
    output_dir.mkdir(parents=True, exist_ok=True)
    df_chunks = pd.read_parquet(chunks_path)
    df_catalog = pd.read_csv(catalog_path, usecols=['file_name', 'L2_corpus_section'])
    df_catalog.rename(columns={'file_name': 'document_id'}, inplace=True)
    df_chunks = pd.merge(df_chunks, df_catalog, on='document_id', how='left')
    df_chunks['L2_corpus_section'].fillna('unknown', inplace=True)
    grouped = df_chunks.groupby('L2_corpus_section')
    corpora_created = 0
    for name, group in grouped:
        if len(group)  [cyan]{output_path.name}[/cyan]")
        full_text = " ".join(group['text_for_lda'].dropna())
        with open(output_path, 'w', encoding='utf-8') as f: f.write(full_text)
        corpora_created += 1
    console.print(f"\n[bold green]✔️ Success![/bold green] Created {corpora_created} training corpora in [cyan]{output_dir}[/cyan].")

if __name__ == "__main__": app()

执行命令：

# Executed from the project root directory: C:\Users\grego\Downloads\tibetan-fingerprinting
python src/prepare_word_embedding_data.py

结果： 新目录 output/word_embedding_training/ 已创建，包含一个 .txt 每个主要L2部分的文件，使用纯ASCII文件名。

3.3.第二阶段：模型训练

目标： 培养高素质人才 fastText 在前一阶段生成的每个语料库文件的模型。

理论基础： 该脚本设计得既健壮又高效。它动态检测主机上可用CPU核的数量，并将几乎所有CPU核分配给 fastText 培训过程中，大大缩短了培训时间。

脚本： src/train_word_models.py

# src/train_word_models.py (v3 - Max Power Edition)

import fasttext
from pathlib import Path
import typer
from rich.console import Console
import json
from datetime import datetime
import psutil

console = Console()
app = typer.Typer(help="Trains fastText word embedding models from prepared corpora.")

@app.command()
def train(
    corpus_dir: Path = typer.Option("output/word_embedding_training/", help="Directory containing the training .txt files."),
    output_dir: Path = typer.Option("output/word_embedding_models/", help="Directory to save the trained .bin models."),
    dim: int = typer.Option(300, help="Dimension of the word vectors."),
    min_count: int = typer.Option(3, help="Minimum word frequency to be included in the vocabulary."),
    model_type: str = typer.Option("skipgram", help="Model type ('skipgram' or 'cbow'). Skipgram is better for semantics.")
):
    """Trains a fastText model for each .txt file, using the maximum safe number of CPU cores."""
    console.print("🚀 Starting fastText Model Training [bold red](Max Power Edition)[/bold red]...")
    output_dir.mkdir(parents=True, exist_ok=True)
    training_files = list(corpus_dir.glob('*.txt'))
    if not training_files: raise typer.Exit(code=1)
    total_cores = psutil.cpu_count(logical=True)
    num_threads = max(1, total_cores - 1)
    console.print(f"Found {len(training_files)} corpora to train. Using [bold yellow]{num_threads} threads[/bold yellow].")
    for txt_file in training_files:
        model_name = txt_file.stem
        model_bin_path = output_dir / f"{model_name}.bin"
        model_meta_path = output_dir / f"{model_name}.json"
        console.print(f"\n--- Training model for: [bold yellow]{model_name}[/bold yellow] ---")
        try:
            model = fasttext.train_unsupervised(str(txt_file), model=model_type, dim=dim, minCount=min_count, epoch=10, thread=num_threads)
            model.save_model(str(model_bin_path))
            metadata = {'model_name': model_name, 'training_file': str(txt_file), 'timestamp_utc': datetime.utcnow().isoformat(), 'hyperparameters': {'model_type': model_type, 'vector_dim': dim, 'min_word_count': min_count, 'epochs': 10, 'threads': num_threads}, 'vocab_size': len(model.words)}
            with open(model_meta_path, 'w', encoding='utf-8') as f: json.dump(metadata, f, indent=2)
            console.print(f"✔️ Saved model to [cyan]{model_bin_path}[/cyan] and metadata to [cyan]{model_meta_path}[/cyan]")
        except Exception as e: console.print(f"❌ Failed to train model for {model_name}. Error: {e}")
    console.print("\n[bold green]✔️ All models trained successfully![/bold green]")

if __name__ == "__main__": app()

执行命令：

# Executed from the project root directory
python src/train_word_models.py

结果： 新目录 output/word_embedding_models/ 已创建，包含最终 .bin 模型文件及其对应关系 .json 元数据文件。总大小：~200GB。

3.4.第3阶段：云摄入和安全

目标： 安全且经济高效地将生成的工件上传到AWS S3。

流程：

1. S3桶创建： 一个名为的S3存储桶 tibetan-mcp-models 创建。“阻止所有公共访问”已禁用，以允许IAM控制的程序访问，并启用了版本控制。
2. IAM政策： 已创建最低权限IAM策略，授予访问权限 *仅* 到 tibetan-mcp-models 桶。
3. IAM用户： 专用程序用户， mcp-server-user，并附上上述政策。其访问密钥和密钥已安全保存。
4. AWS CLI配置： 使用用户的凭据配置了本地AWS CLI。

   # Executed once to set up credentials
   aws configure

1. 工件上传： 这 aws s3 sync 命令用于上传文件。做出了一个关键的战略决策，即使用不同的存储类进行部署和训练工件。

- 模型（部署工件）： 上传到标准、快速访问层。

     # Executed from the project root directory
     aws s3 sync output\word_embedding_models s3://tibetan-mcp-models/models/

- 下士（培训档案）： 上传到更便宜的 GLACIER_IR 层。

     # Executed from the project root directory
     aws s3 sync output\word_embedding_training s3://tibetan-mcp-models/training_archives/ --storage-class GLACIER_IR

3.5.第4阶段：验证和确认

目标： 以加密方式验证大型模型是否已上传到S3而没有损坏。

理论基础： 对于大文件，S3使用“多部分上传”过程。生成的ETag不是文件的简单MD5哈希，而是其部分哈希的哈希。因此，简单的哈希比较失败。最终验证方法是往返测试。

流程：

1. 一个大型模型(middle_way.bin)已从S3下载到新的本地文件。

   aws s3 cp s3://tibetan-mcp-models/models/middle_way.bin output/word_embedding_models/middle_way_from_s3.bin

1. 已计算原始本地文件的MD5哈希值。

   certutil -hashfile "output\word_embedding_models\middle_way.bin" MD5
   # Output: 9797e6ed2ccd6e375be8dd2728dd9701

1. 计算了新下载文件的MD5哈希值。

   certutil -hashfile "output\word_embedding_models\middle_way_from_s3.bin" MD5
   # Output: 9797e6ed2ccd6e375be8dd2728dd9701

结果： 哈希值完美匹配，提供了加密证据，证明上传过程成功，S3中的数据完好无损。

4.探索工具包

为了立即开始利用这些模型，开发了一个强大的命令行“瑞士军刀”。

脚本： src/explore_model.py

# src/explore_model.py (v1.1 - Type Hint Fix)

import typer, fasttext
from pathlib import Path
from rich.console import Console
from rich.table import Table
import numpy as np
from scipy.spatial.distance import cosine

app = typer.Typer(name="explore", help="A Swiss Army Knife for exploring fastText word embedding models.")
console = Console()

def load_model(model_name: str):
    """Loads a model and handles errors."""
    model_path = Path(f"output/word_embedding_models/{model_name}.bin")
    if not model_path.exists():
        console.print(f"❌ [bold red]Error:[/bold red] Model file not found at '{model_path}'")
        raise typer.Exit(code=1)
    console.print(f"🔎 Loading model [cyan]{model_name}.bin[/cyan]...")
    model = fasttext.load_model(str(model_path))
    console.print(f"✔️ Model loaded. Vocabulary size: {len(model.words)} words.")
    return model

@app.command()
def neighbors(model_name: str, word: str, k: int = 10):
    """Finds the nearest neighbors for a single word."""
    model = load_model(model_name)
    results = model.get_nearest_neighbors(word, k=k)
    # (Table formatting code...)

@app.command()
def analogy(model_name: str, a: str, b: str, c: str):
    """Solves the analogy 'A is to B as C is to ?'"""
    model = load_model(model_name)
    results = model.get_analogies(a, b, c)
    # (Table formatting code...)

@app.command()
def similarity(model_name: str, word1: str, word2: str):
    """Calculates the semantic similarity score between two words."""
    model = load_model(model_name)
    vec1, vec2 = model.get_word_vector(word1), model.get_word_vector(word2)
    sim_score = 1 - cosine(vec1, vec2)
    console.print(f"Similarity: [bold magenta]{sim_score:.4f}[/bold magenta]")

@app.command()
def outlier(model_name: str, words: list[str]):
    """Finds the word that doesn't belong in a list."""
    model = load_model(model_name)
    vectors = [model.get_word_vector(w) for w in words]
    mean_vector = np.mean(vectors, axis=0)
    distances = [cosine(v, mean_vector) for v in vectors]
    outlier_index = np.argmax(distances)
    console.print(f"Outlier: '[bold red]{words[outlier_index]}[/bold red]'")

if __name__ == "__main__": app()

*（注：项目中包含带表格格式的完整脚本 src 目录。)*

使用指南： 每个命令的详细示例见第1节。

5.前进之路

用于单词级语义分析的后端数据管道现在已经完成、验证并安全存储。该项目已为最后阶段做好了充分准备： API开发。接下来的步骤将涉及在Render.com上创建FastAPI应用程序，实现 ModelManager 用于按需加载S3，并通过公开的API公开本文中记录的强大分析函数。这将使这些功能能够集成到面向用户的应用程序和高级代理系统中。