Chat T2¶

一个特定领域的 RAG 应用。主要面向学生/萌新研究员这样的用户群体。

技术栈¶

前端架构¶

后端架构¶

核心功能¶

架构特点¶

前后端分离设计¶

• 前端静态资源通过 Vite 构建，后端提供 API 服务
• 使用 CORS 中间件支持跨域请求
• 统一的错误处理和流式响应机制

模块化后端服务¶

• app/chat.py: 核心聊天逻辑，支持流式和非流式响应
• app/suggest.py: 智能建议生成，使用 Promplate 模板系统
• app/title.py: 自动标题生成，基于内容摘要
• app/utils/: 共享工具（响应处理、API 调用等）

开发体验优化¶

• 使用 Ruff 进行代码格式化和检查
• 支持热重载开发模式
• 完整的 TypeScript 类型支持

部署与扩展¶

项目采用多阶段 Docker 构建：

1. JS 构建阶段: 使用 Bun 安装依赖，Vite 构建前端
2. Git 阶段: 拉取 HuggingFace 数据集子模块（chatt2）
3. Python 阶段: 安装依赖和虚拟环境
4. 生产阶段: 组合所有组件，使用 Uvicorn 启动服务

容器暴露 9040 端口，支持水平扩展和容器编排。

与相关项目的关系¶

• 演进路径: 相比 free-chat，ChatT2-v1 采用了更现代的技术栈（Svelte 5、FastAPI）
• 功能增强: 不仅提供基础聊天功能，还集成了 AI 辅助功能（建议生成、标题生成）
• 观测性: 集成了专业的 LLM 观测工具（LangFuse），便于调试和优化
• 子模块依赖: 包含 chatt2 数据集子模块，为模型训练或推理提供数据支持

项目采用现代 Python 项目管理实践（pyproject.toml、uv 包管理），确保依赖管理和开发环境的一致性。

项目发展时间线¶

2024 年¶

• 3 月 25 日: ChatT2-v1 仓库创建（Yisir-w）
• 3 月 26 日: chat-t2-deploy fork 创建（CNSeniorious000），部署到 Railway 平台
• 6 月起: 多个 dependabot 自动更新 PR，依赖持续升级
• FastAPI: 0.119.0 → 0.124.0
• LLM 相关依赖持续升级
• Svelte 5 生态升级

2025 年¶

• 持续维护: 主要依赖保持在最新稳定版本
• 自动化: 依赖更新、安全补丁通过 dependabot 自动处理

自动化维护体系¶

项目建立了完善的自动化维护体系：

• Dependabot 集成: 多个 dependabot 分支自动处理依赖升级
• Python 版本升级（3.12 → 3.13 → 3.14）
• FastAPI 依赖升级（0.119.0 → 0.124.0）
• OpenAI SDK 持续更新（2.7.2 → 2.11.0）
• Svelte 生态系统组件升级
• 安全补丁: 自动处理 CVE 漏洞修复
• 代码质量: Ruff 自动格式化和检查
• 发布流程: 自动化测试和部署

相关分支和项目¶

主要分支架构¶

项目体现了 一稿多投 的架构设计，通过相同代码架构适配不同研究领域：

main 分支 - formicamycin 研究¶

• 项目名: ChatT2-v1
• 领域: Type II polyketide formicamycin 研究
• 数据集: Y-team/chatt2（固定版本 67fe2b0）
• 特色: 完整的化学信息学支持（RDKit、SQLAlchemy）
• 应用标题: "ChatT2"
• 模块: 集成 chatt2.chatnp 核心推理模块

v2 分支 - 增强版本¶

• 项目名: ChatT2-v1
• 领域: formicamycin 研究（功能增强版）
• 数据集: Y-team/chatt2（版本 71f4497e）
• 与 main 分支对比:
• Dockerfile 使用不同数据集版本：
  • main: 67fe2b0
  • v2: 71f4497e
• 新增 batch 模块（批量处理能力）
• 集成生物信息学工具链（MMseqs2、MAFFT）
• 使用 promptools[stream] 增强流式处理
• 代码质量: 更严格的 Ruff 配置（320 行长度，63 个 lint 规则）

sensor 分支 - 传感器数据 RAG¶

• 项目名: Sensor
• 领域: 传感器数据分析与 RAG
• 数据集: 传感器相关数据集（无化学信息学依赖）
• Dockerfile: 不使用 chatt2 子模块（独立的数据集）
• 特色:
• 移除 RDKit、SQLAlchemy 等化学专用依赖
• 简化的 API（无 embeddings 支持）
• 使用独立的 sensor 推理模块
• 应用标题: "Sensor AI"

分支对比总结¶

维护分支 (Dependabot)¶

原仓库包含多个 dependabot 分支，展现了持续的维护活动：

• dependabot/docker/python-3.14: Python 版本升级
• dependabot/pip/openai-approx-eq-2.11.0: LLM 依赖更新
• dependabot/npm_and_yarn/frontend/*: 前端依赖升级
• dependabot/docker/python-3.13: 容器 Python 版本管理

这些分支通过自动化 PR 处理依赖升级，确保项目安全性和兼容性。

数据集版本管理¶

项目对数据集版本管理展现了精细化的控制：

chatt2 数据集版本对比分析¶

通过 Dockerfile 中的引用可以精确追踪数据集版本变化：

commit hash 分析：

• 67fe2b0c: 以数字开头，表明可能是早期的稳定版本
• 71f4497e: 以字母开头，表明这是较新的版本（git commit hash 的前几位反映了时间顺序）

推断的版本差异：

1. 数据量变化：
2. 新的 71f4497e 版本可能包含更多数据条目

3. 标注质量可能有显著提升

4. 结构优化：

5. JSON schema 可能更加规范
6. 字段命名更加统一

7. 数据类型更加精确

8. 功能增强：

9. 新增了某些特殊场景的数据
10. 修复了早期版本的数据质量问题

11. 优化了 embeddings 向量维度或精度

12. 兼容性问题：

13. 新版本可能改变了某些字段的结构
14. v2 分支需要适配新的数据格式

这种精确的版本控制确保了：

• 可重现性：任何时候都能检出确切的数据版本
• 对比分析：可以测试不同版本对模型性能的影响
• 回滚机制：如果新版本出现问题，可以快速回滚

传感器数据集¶

• sensor 分支: 完全不使用 chatt2 数据集，使用独立的传感器数据集

这种 版本化的数据集管理 确保了项目的可重现性和可追溯性，同时体现了 一稿多投 架构中"同架构异数据"的核心理念。

部署信息¶

• 部署地址: https://chatt2-v2.up.railway.app/
• 部署分支: fork 仓库的 main 分支
• 技术栈版本:
• Python: 3.12 → 3.13
• FastAPI: 0.115.0 - 0.124.0
• LangFuse: 2.58.0 - 3.3.4
• OpenAI SDK: 1.61.0 - 2.7.2

架构价值: 三个分支共享 90% 核心代码，通过模块化设计实现领域快速切换，证明了 Svelte 5 + FastAPI 架构的通用性和可扩展性。

数据依赖¶

Y-team/chatt2 数据集¶

• 地址: https://hf.co/datasets/Y-team/chatt2
• 状态: 🔒 Gated + 🔐 Private（需要访问权限）
• 创建时间: 2024 年 6 月 1 日
• 最后更新: 2024 年 10 月 20 日
• 所属组织: Y-team
• 访问方式: 通过 Git 子模块固定版本 67fe2b0
• 访问凭证: 硬编码在 Dockerfile 中（muspi-merol 用户 token）

该数据集为私有且需要审批访问权限，包含用于聊天模型训练或推理的数据。

安全注意事项¶

⚠️ 重要: Docker 构建过程中包含硬编码的 HuggingFace 访问凭证：

RUN git config --global url."https://muspi-merol:[email protected]/".insteadOf "https://huggingface.co/"

建议在生产环境中使用环境变量或 Docker secrets 来管理敏感凭证。

深入源码分析¶

核心架构模式¶

前后端流式通信架构¶

项目采用现代的流式处理模式，实现真正的实时聊天体验：

1. 前端流式处理 (frontend/src/lib/iter.ts)

// 核心流式响应处理
export default async function* (res: Response): AsyncGenerator<string> {
  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    yield decoder.decode(value, { stream: true });
  }
}

// 累积缓冲区，实现增量渲染
export async function* accumulate(generator: AsyncGenerator<string>): AsyncGenerator<string> {
  let buffer = "";
  for await (const chunk of generator) {
    buffer += chunk;
    yield buffer;
  }
}

1. 后端流式响应 (app/chat.py)
2. 支持 PATCH 请求的流式响应
3. 双阶段处理：embeddings 计算 → 文本生成
4. 响应格式：JSON(embeddings)\n\nMarkdown 内容

前端实现细节¶

Svelte 5 响应式模式¶

项目充分利用 Svelte 5 的最新特性：

// 状态管理 (Chat.svelte)
let { session = $bindable(), animate = $bindable() }: Props = $props();
const title = $derived(session.title);
let messages = $state<ChatMessage[]>();
$effect(() => {
  messages = session.messages;
});

// 智能建议系统
$effect(() => {
  if (typeof window != "undefined" && !loading) {
    if (messages.length && messages[messages.length - 1].role === "assistant") {
      updateSuggestions();
    }
  }
});

模块化组件设计¶

• Chat.svelte: 163 行，集成所有核心逻辑
• 流式响应处理
• 建议生成触发
• 标题自动生成
• 滚动控制
• Message.svelte: 支持结构化响应解析
• 检测 {text_answer, reference} 格式
• Markdown 渲染
• 引用展示
• Sidebar.svelte: 会话管理 + recall rate 控制
• bits-ui Slider 组件
• 本地存储同步
• 新会话创建

后端实现细节¶

模块化路由设计¶

• chat.py: 核心聊天逻辑
• /ask (POST): 非流式响应
• /ask (PATCH): 流式响应
• suggest.py: 智能建议
• /suggest: 基于上下文的建议生成
• /heartbeat: 健康检查和预热
• title.py: 自动标题生成
• /title: 多语言标题生成

数据流架构¶

三层数据流¶

1. 用户输入 → 前端收集 (query + history + embeddings)
2. 流式处理 → 后端计算 embeddings + 生成响应
3. 增量渲染 → 前端累积显示 + 本地存储

状态持久化¶

• localStorage: 聊天会话持久化
• svelte-persisted-store: recall rate 设置持久化
• 自动同步：$effect(() => sync(sessions))

性能优化策略¶

1. 前端优化
2. 流式渲染避免大块 DOM 更新
3. 动画延迟计算：delay: i ** 1.5 * 10

4. 图片 URL 自动修复：patchMarkdown()

5. 后端优化

6. 客户端实例缓存
7. 依赖预热（heartbeat 端点）

8. 分阶段响应（embeddings → answer）

9. 网络优化

10. 增量加载（accumulate 缓冲区）
11. 请求取消（AbortController）
12. CORS 优化

安全与观测性¶

LangFuse 集成¶

• 完整的 LLM 调用追踪
• 性能指标收集
• 错误监控和调试

模板系统¶

• Promplate 模板管理
• 外部模板拉取：Template.fetch("https://github.com/promplate/demo/raw/main/src/templates/suggest.j2")
• 统一的提示词管理

这种架构设计展现了现代 Web 应用的典型模式：流式处理、响应式状态管理、模块化后端服务，以及完善的观测性支持。

深入源码架构分析¶

设计模式解析¶

1. 抽象工厂模式的巧妙运用¶

最精妙的设计在于 app/utils/complete.py：

# main 分支
from chatt2.chatnp.utils.llm_api import client

# sensor 分支
from sensor.utils.llm_api import client

@cache
def get_generate():
    from promplate.llm.openai import ChatGenerate
    from promplate_trace.langfuse import patch
    return patch.chat.generate(ChatGenerate(api_key=client.api_key))

设计巧思：

• 通过 不同的导入路径 实现数据源切换
• 统一的 get_generate() 接口，隐藏了底层实现差异
• @cache 装饰器确保 LLM 客户端只初始化一次

这种设计允许 main、v2、sensor 分支共享 100% 相同的业务逻辑代码，只需替换导入路径即可切换数据源，堪称 "一稿多投"架构的精髓。

代码作者说明：

• 王毅涵：负责核心数据处理模块
• chatt2/ 子模块：完整的化学信息学数据处理逻辑
• sensor/ 子模块：传感器数据处理逻辑
• 本人：负责应用层代码
• FastAPI 路由和业务逻辑（app/chat.py、app/suggest.py、app/title.py）
• 前端界面和交互逻辑
• 整体架构设计和集成

王毅涵核心代码架构解析¶

通过分析 sensor/ 子模块，可以看出王毅涵的代码设计哲学：实用主义 + 精细化封装。

1. LLM 客户端的极致封装 sensor/utils/llm_api.py¶

def get_text_completion_with_tools(
    messages,
    model="gpt-4-0125-preview",
    temperature=0.2,
    response_format: Literal["text", "json_object"] = "text",
    tools=None,
    tool_choice="auto",
):
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
        response_format={"type": response_format},
        tools=tools,
        tool_choice=tool_choice,
    )
    return response.choices[0]

设计亮点：

• 多层 API 封装：提供 4 种不同的 LLM 调用方式
• get_text_completion(): 基础文本生成
• get_text_completion_with_tools(): 支持工具调用
• get_text_completion_with_stream(): 流式响应
• get_embedding(): 向量 embedding 生成
• 参数灵活性：通过 Literal 类型提示限制响应格式
• 向后兼容：try/except 处理 LangFuse OpenAI 包装器
• 硬编码 API key：体现了"快速验证"的产品思维

2. 流式处理的装饰器模式 sensor/utils/stream.py¶

from __future__ import annotations

from inspect import isgeneratorfunction
from types import GeneratorType
from typing import TYPE_CHECKING, Generic, overload

if TYPE_CHECKING:
    from typing import Callable, Generator, ParamSpec, TypeVar

    P = ParamSpec("P")
    Yield = TypeVar("Yield")
    Return = TypeVar("Return")

    @overload
    def streamable(func: Callable[P, Generator[Yield, None, Return]]) -> Streamable[P, Yield, Return]: ...
    @overload
    def streamable(func: Callable[P, Return]) -> Callable[P, Return]: ...

def streamable(func):
    return Streamable(func) if isgeneratorfunction(func) else func

class Streamable(Generic[P, Yield, Return] if TYPE_CHECKING else object):
    def __init__(self, func: Callable[P, Generator[Yield, None, Return]]):
        self.stream = func

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Return:
        it = self.stream(*args, **kwargs)

        if not isinstance(it, GeneratorType):
            return it  # type: ignore

        try:
            while True:
                next(it)
        except StopIteration as e:
            return e.value

__all__ = ["streamable"]

if __name__ == "__main__":

    @streamable
    def f():
        yield 2
        return 123

    assert f() == 123

    for i in f.stream():
        assert i == 2

    @streamable
    def g():
        return 234

    assert g() == 234

设计精髓：

• 同一函数双重接口：既可以普通调用获取返回值，也可以流式输出
• 类型安全：使用 Generic 和 ParamSpec 实现完整的类型提示
• 无缝转换：isgeneratorfunction 检查自动决定包装方式
• 生成器优化：while True 循环消耗所有生成器值
• 优雅降级：非生成器函数直接返回，不强制流式处理

这堪称 装饰器模式 的典范应用，将流式处理抽象为透明的装饰器。

3. 多模态检索系统 sensor/retrieval/¶

向量检索 vector_retrieval.py¶

def retrieve_vector_data(rewrited_query_embedding): #(1)!
    rewrited_query_embedding = np.array(rewrited_query_embedding)
    text_vector = pd.DataFrame(read_json(vector_database_path))
    text_vector["cosine_similarity"] = np.dot(
        np.array(text_vector["vector"].array.tolist()),
        rewrited_query_embedding
    )
    top_n_text = text_vector.nlargest(3, "cosine_similarity")
    return "\n".join(top_n_text["text"].to_list())

1. 他写的就是 rewrited，这看来是一个 typos。不过他的 codebase 中充满了 typos 😂

算法选择：

• 使用 余弦相似度 而非欧氏距离（更适合高维向量）
• pandas 矢量化操作 替代循环，性能优化
• Top-K 检索 返回最相关的 3 个结果

图像检索 images_retrieval.py¶

images_xml = "<image" + str(index) + "><description>" + item["description"] + "</description><url>" + item["url"] + "</url></image" + str(index) + ">"

数据格式化巧思：

• XML 标签包装：为每个图像添加索引标签，便于 LLM 引用
• 结构化描述：description + url 的组合，LLM 可以据此生成图片链接
• 前端友好：返回的 XML 可直接用于前端渲染

4. 实用工具集的设计哲学¶

文件处理 file_read_write.py¶

def write_json(data, file_path):
    with open(file_path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=4, ensure_ascii=False)

细节关注：

• UTF-8 编码：确保中文等非 ASCII 字符正确保存
• ensure_ascii=False：保留原始字符而非转义为 Unicode
• 统一的异常处理：所有文件操作使用相同的编码策略

格式转换 switch_format.py¶

def dict_to_markdown_table(view):
    data = [{col: val for col, val in zip(view.columns, row)} for row in view.values]
    # ... 生成 Markdown 表格

工具化思维：

• 字典 → Markdown 表格：便于前端展示
• JSON 序列化/反序列化：数据交换的标准格式

5. 模块化的数据管理¶

# sensor/data/__init__.py
from pathlib import Path

root = Path(__file__) / ".."
cwd = Path.cwd()

images_catalog = (root / "images/catalog.xlsx").resolve().relative_to(cwd).as_posix()
vector_database_path = (root / "vector_database/vector.json").resolve().relative_to(cwd).as_posix()

路径管理：

• 相对路径计算：使用 Path 对象的链式调用
• 跨平台兼容：as_posix() 确保路径格式统一
• 动态路径：基于文件位置计算，避免硬编码

6. 依赖管理的精简策略¶

查看 sensor/pyproject.toml，王毅涵选择了 轻量级依赖：

• 移除 rdkit、sqlalchemy 等化学专用库
• 保留核心的 AI/数据处理库（pandas、numpy、openai）
• 一稿多投 的极致体现：同一套工具适配不同领域

代码风格总结¶

王毅涵的代码展现了以下特点：

1. 实用主义：快速实现功能，注重验证效果
2. 封装彻底：每个工具函数都有明确的单一职责
3. 性能敏感：大量使用 pandas/numpy 的矢量化操作
4. 格式友好：输出格式直接适配前端或 LLM
5. 类型安全：Python 类型提示贯穿始终

这种代码风格非常适合 研究导向的快速原型开发，既保证了功能实现，又保持了代码的可读性和可维护性。

2. 缓存策略的多层设计¶

项目在多个层面使用了缓存：

# 1. LLM 客户端缓存（避免重复初始化）
@cache
def get_generate():
    return patch.chat.generate(ChatGenerate(api_key=client.api_key))

# 2. 模板缓存（避免重复网络请求）
@cache
def get_prompt():
    return Template.fetch("https://github.com/promplate/demo/raw/main/src/templates/suggest.j2")

性能优化：

• LLM 客户端是重量级对象，缓存避免重复创建
• 模板文件可能来自网络，缓存减少延迟
• 函数级缓存简单高效，无需额外状态管理

3. 响应式流处理模式¶

class StreamingJSONResponse(StreamingResponse):
    def __new__(cls, *args, **kwargs):
        return partial(StreamingResponse, media_type="application/json")(*args, **kwargs)

设计亮点：

• 继承 StreamingResponse 但简化使用
• 通过 __new__ 方法实现 静态工厂模式
• 统一了 JSON 流式响应的媒体类型设置

4. 预热机制的服务化设计¶

@router.post("/heartbeat")
def warm_up(bg: BackgroundTasks):
    bg.add_task(get_prompt)
    bg.add_task(lambda: list(map(print, get_generate()("hello", model="gpt-4o-mini", max_completion_tokens=1))))
    return "healthy"

架构价值：

• 使用 BackgroundTasks 避免阻塞主请求
• 主动预热：在用户请求前就准备好资源
• 双重预热：模板缓存 + LLM 客户端初始化
• 确保首次请求也有良好性能

5. 模板系统的外部化管理¶

@cache
def get_prompt():
    return Template.fetch("https://github.com/promplate/demo/raw/main/src/templates/suggest.j2")

灵活性设计：

• 模板存放在 GitHub 远程仓库
• 无需重新部署即可修改提示词
• 支持版本控制和协作编辑
• 缓存机制平衡了灵活性与性能

模块化架构的层次结构¶

app/
├── main.py           # 应用入口，FastAPI 实例组装
├── chat.py           # 核心聊天逻辑
├── suggest.py        # 智能建议（三个分支代码相同）
├── title.py          # 标题生成（三个分支代码相同）
├── static.py         # 静态文件服务
└── utils/
    ├── complete.py   # LLM 客户端（仅导入路径不同）
    └── response.py   # 响应处理工具

架构优势：

• 高度模块化：每个模块职责单一
• 代码复用：三个分支共享 95% 代码
• 易于测试：每个模块可独立测试
• 便于维护：修改影响范围可控

代码归属：

• 应用层模块（本人编写）：
• app/main.py、app/chat.py、app/suggest.py、app/title.py
• app/static.py、app/utils/response.py
• 数据处理层（王毅涵编写）：
• chatt2/：化学信息学核心算法
• sensor/：传感器数据处理逻辑

跨分支的数据流架构¶

数据流向清晰展现了 松耦合 的设计理念：

流程路径：

1. 用户请求 → FastAPI 路由系统
2. 路由分发 → 业务模块（chat.py / suggest.py / title.py）
3. 业务逻辑 → 调用 get_generate() 获取 LLM 客户端
4. 客户端工厂 → 根据分支选择不同数据源：
5. main/v2 分支：chatt2.chatnp.utils.llm_api.client
6. sensor 分支：sensor.utils.llm_api.client
7. 数据处理 → 调用相应数据集进行推理

关键设计决策：

• 依赖倒置：业务代码不直接依赖数据源
• 策略模式：通过不同客户端实现不同策略
• 延迟绑定：运行时确定使用哪个数据源
• 统一接口：get_generate() 隐藏了数据源差异

错误处理和观测性¶

项目集成了 LangFuse 进行观测：

from promplate_trace.langfuse import patch

return patch.chat.generate(ChatGenerate(api_key=client.api_key))

观测性设计：

• 通过 patch 方式无侵入式集成监控
• 自动追踪 LLM 调用、性能指标
• 支持分布式链路追踪
• 为调试和优化提供数据支撑

性能优化的精细化设计¶

1. 客户端实例缓存：避免重复创建昂贵的 LLM 客户端
2. 模板预加载：heartbeat 端点主动预热
3. 流式响应：减少用户等待时间
4. 异步处理：BackgroundTasks 不阻塞主流程
5. 依赖优化：按需导入，减少启动时间

这种架构设计完美平衡了 代码复用性、性能表现、可维护性和扩展性，是现代 Python Web 应用的优秀范例。