AI安全知识库
一个开源、精心策划、结构化和可搜索的人工智能安全语料库，为安全感知代理、MCP工具和对齐研究工作流程提供支持。
此存储库提供什么
- 摄入管道 将对齐论坛帖子、arXiv论文、AI事件数据库条目、治理报告和内部文档规范化为一个具有来源、版本控制和源记录的统一模式。
- 存储层 基于SQL数据库（SQLite用于开发，Postgres+pgvector用于生产）构建，具有块级嵌入和元数据过滤器。
- 源注册表+目录 因此，每个文档都可以追溯到每个源记录，包括摄取模式、状态、文档计数和规范URL。自动生成
sources_catalog.md保持一个动态的“链接”页面与注册表同步。 - 检索API (
safety_kb.retrieval)暴露异步帮助程序，例如search,get_document,get_chunks_for_document,search_by_topic，以及list_topics这些函数返回MCP工具可以直接包装的Pydantic模型。 - 公用事业 用于分块、文本清理、嵌入管理、日志记录和摄取编排。
- 测试和迁移 在KB扩展到数万个文档和数百万个块时保持其稳定。
项目布局
ai-safety-knowledge-base/
├── env.example # Copy to .env to override defaults
├── pyproject.toml # PEP 621 metadata + dependencies
├── README.md
├── migrations/
│ └── 001_init.sql # Reference SQL definition for Postgres/pgvector
├── src/
│ └── safety_kb/
│ ├── __init__.py
│ ├── config.py # Settings + helpers
│ ├── models.py # Pydantic domain models
│ ├── schemas.py # SQLAlchemy ORM tables
│ ├── storage.py # Async persistence helpers
│ ├── retrieval.py # Public KB API
│ ├── indexing.py # Ingestion + indexing orchestration
│ ├── sources/ # Source-specific fetchers
│ ├── catalog.py # Source catalog generator (produces sources_catalog.md)
│ └── utils/ # Text cleaning, chunking, embeddings, logging, checksums
└── tests/
├── test_indexing.py
├── test_ingestion_sources.py
└── test_retrieval.py入门指南
- 安装依赖项
pip install -e .- 配置设置
复制 env.example 到 .env （或设置环境变量）并更新值，例如 SAFETY_KB_DATABASE_URL 以及嵌入提供者配置。
- 创建数据库
对于本地开发人员：
python3 -m safety_kb.storage --init-db对于生产Postgres，请申请 migrations/001_init.sql 并确保 pgvector 已启用。
- 运行测试
pytest- 通过目录或本地文件添加新源
- Web链接：编辑 sources_catalog.md 并添加具有源名称、种类（例如。， website)，模式，以及最后一列中的Markdown超链接。
- 地方文献：drop .txt, .md, .html，或 .pdf 文件进入 sources/files/.
- 然后运行：
python3 -m safety_kb.catalog_sync --catalog sources_catalog.md --sources-dir sources/files此命令：
1. 解析目录行，扰乱这些源，以及（ website 种类）获取/摄取它们的内容。 1. 在以下位置发现新文件 sources/files/，摄取它们，并使用工作超链接将行添加到目录中。 1. 重新生成 sources_catalog.md 来自实时注册表，因此页面始终反映实际摄入的内容。
提交摄入的文件（如果需要）和刷新的目录。
与MCP集成
这 safety_kb.retrieval 模块特意设计得很薄，以便MCP工具定义保持简单：
from safety_kb.retrieval import search, get_document
@server.tool(name="kb.search")
async def kb_search(request):
results = await search(
query=request.input["query"],
k=request.input.get("k", 5),
filters=request.input.get("filters", {}),
)
return {"results": [r.model_dump() for r in results]}路线图
- 添加生产级摄入连接器（校准论坛、arXiv、AIID等）。
- 扩展可观察性（度量、结构化日志、跟踪）。
- 支持缓存嵌入和使用workers（Celery/Arq）排队摄取。
- 引入更先进的重新排序（交叉编码器）和混合BM25+ANN管道。