JashanLabs/LogsRAG
GitHub: JashanLabs/LogsRAG
LogsRAG 是一套面向事件响应场景的实时 RAG 系统，通过 Kafka 流式摄入 GitHub、Jira 及日志事件并写入 pgvector，支持值班工程师基于混合检索进行上下文感知的故障排查。
Stars: 0 | Forks: 0
# LogsRAG
一个用于事件响应（incident response）的实时 RAG pipeline。通过 Kafka 将操作事件（GitHub commits、Jira tickets、系统日志）流式传输到 pgvector 知识库，然后使用混合检索（hybrid retrieval）和 LLM 生成来回答值班（on-call）查询。
标准的 RAG pipeline 会按计划批处理文档。这意味着在正在发生的服务中断期间，导致该事件的 commits 尚未被索引。OpsStream-RAG 消除了这一差距 —— 事件在发生后的几秒钟内即被向量化并可供查询。
## 架构
六个服务，全部容器化。只需一条 `docker compose up --build -d` 命令即可运行整个技术栈。
```
GitHub / Jira Webhooks On-Call Engineer
│ │
▼ ▼
┌────────────────┐ ┌────────────────┐
│ API Gateway │ │ API Gateway │
│ POST /webhooks │ │ POST /ask │
│ (Pydantic) │ │ (RAG generate) │
└───────┬────────┘ └───────┬────────┘
│ │
▼ │
┌────────────────┐ │
│ Apache Kafka │ │
│ (3 partitions)│ │
└───────┬────────┘ │
│ │
┌──────┼──────┐ │
▼ ▼ ▼ │
Consumer Consumer Consumer │
(batch → embed → upsert) │
│ │ │ │
└──────┼──────┘ │
▼ │
┌────────────────┐ │
│ Embedding API │ │
│ Jina v2 local │ │
│ (768-dim) │ │
└───────┬────────┘ │
▼ │
┌────────────────┐ ┌────────┴───────┐
│ PostgreSQL 16 │◄─────────────────────┤ Retrieval API │
│ pgvector HNSW │ │ hybrid search │
│ + GIN FTS │ │ + time decay │
└────────────────┘ └────────────────┘
```
| 服务 | 容器 | 端口 | 功能描述 |
|---|---|---|---|
| API Gateway | `opsstream_rag_api` | 8897 | 接收 webhooks，生产消息到 Kafka。负责协调 RAG 答案的生成。 |
| Kafka | `opsstream_broker` | 9092 | KRaft 模式 broker。包含 3 个分区的 topic，用于并行消费。 |
| Consumer | ×3 个副本 | — | 轮询 Kafka，批量处理事件，调用 embedding API，并 upsert 到 Postgres。 |
| Embedding API | `opsstream_embedder` | 8000 | 通过 PyTorch 在本地运行 `jina-embeddings-v2-base-code`。模型已内置到 Docker 镜像中。 |
| Retrieval API | `opsstream_retrieval` | 8085 | 混合 vector + 关键词搜索，并带有时间衰减重排序（temporal decay reranking）。 |
| PostgreSQL | `opsstream_db` | 5432 | 配备用于向量的 HNSW 索引 + 用于全文搜索的 GIN 索引的 pgvector。 |
## 数据摄入（Ingestion）工作原理
1. Webhook 发送至 `POST /webhooks/github` 或 `POST /webhooks/jira`
2. Pydantic 验证 payload —— 格式错误的请求会被拒绝并返回 HTTP 422，绝不会触及 Kafka
3. 根据事件 ID 生成确定性的 UUID v5（相同的输入 → 每次都会生成相同的 UUID）
4. 事件被生产到 Kafka topic `system-events` 中，并采用基于 key 的分区（partitioning）
5. API 立即返回 —— webhook 响应耗时 <5ms
在 consumer 端：
6. 3 个 consumer 副本以 `enable.auto.commit: False` 模式轮询 Kafka
7. 事件会被缓冲，直到累积达到 10 条或超过 2 秒
8. 批次被发送到本地 embedding API（Jina v2，768 维向量）
9. 向量通过 `ON CONFLICT (id) DO NOTHING` 被 upsert 到 PostgreSQL
10. **仅在**数据库事务成功之后，Kafka 的 offset 才会被提交
如果 consumer 在第 8 步到第 10 步之间崩溃，offset 将保持未提交状态，Kafka 会在重启时重放该批次。确定性的 UUID + 冲突保护机制可确保不会有重复数据进入数据库。
## 检索（Retrieval）工作原理
检索 API 针对 PostgreSQL 运行两个并行的、由索引支持的查询，然后合并结果：
**语义搜索（Semantic search）** —— 针对 `embedding` 列的 HNSW 索引，按余弦距离（cosine distance）排序。捕获语义层面的相似性（比如 _"database crashed"_ 会匹配 _"PostgreSQL connection refused"_）。
**关键词搜索（Keyword search）** —— 针对 `to_tsvector('english', content)` 的 GIN 索引，按 `ts_rank_cd` 排序。捕获向量搜索容易遗漏的精确匹配：如 `Refs #33173` 这样的 ticket ID、`test_ForeignKey_using_to_field` 这样的测试名称，以及版本号。
这两个结果集会被合并并评分：
```
hybrid_score = 0.7 × semantic_score + 0.3 × keyword_score
```
接着会应用时间衰减因子来降低旧事件的权重：
```
final_score = hybrid_score × exp(−0.05 × age_in_days)
```
这会带来大约 14 天的半衰期（half-life）—— 近期事件会自然浮现，而无需使用那些可能会丢弃潜在有用的旧上下文的硬性日期截断。衰减系数存在于应用层，因此调整它不需要修改 schema 或重建索引。
在执行搜索之前，查询本身会经过一次 LLM 调用来进行优化，该调用会从自然语言输入中提取错误代码、服务名称和关键术语。这是一个尽力而为（best-effort）的步骤 —— 如果失败（如受到速率限制、超时），则会退回到原始查询作为后备方案。
## 故障处理
| 故障情况 | 后果 |
|---|---|
| PostgreSQL 宕机 | Consumers 以指数退避方式重试（5秒 → 10秒 → 20秒）。Kafka offsets 保持未提交状态，因此事件会保留在磁盘上，并在数据库恢复后重放。 |
| Embedding API 崩溃 | Consumer 会记录错误并故意崩溃。Docker 的 `restart: unless-stopped` 策略会重新启动它。未提交的 offsets 会触发重放。 |
| 消息结构错误 | 如果 embedding API 返回 400/422（根本上属于坏数据），consumer 会将该批次路由到死信队列（Dead Letter Queue）topic，并提交 offset 以避免阻塞队列。 |
| Consumer 副本死亡 | Kafka 会检测到心跳丢失，触发分区重平衡（rebalance），并将孤立的分区重新分配给存活的副本。 |
| 查询时触及 LLM 速率限制 | API 会以指数退避方式重试（最多 5 次）。检索 API 的查询优化步骤会平滑降级为使用原始查询。 |
## 快速开始
**前提要求：** Docker Desktop (Compose V2+)、Python 3.11+、一个 [OpenRouter](https://openrouter.ai/) API key（免费额度即可）。
```
git clone https://github.com/YOUR_USERNAME/OpsStream-RAG.git
cd OpsStream-RAG
```
创建 `.env`：
```
OPENROUTER_API_KEY=sk-or-v1-...
LLM_MODEL=google/gemma-4-31b-it:free
GITHUB_TOKEN=ghp_... # optional, raises GitHub API rate limit
```
启动技术栈：
```
docker compose up --build -d
```
首次构建需要几分钟（下载 PyTorch，并将 Jina 模型内置到镜像中）。在此之后，启动将是瞬间完成的。
### 加载数据
从 [Django](https://github.com/django/django) 仓库摄入 1,000 条 commits：
```
python -m venv venv
venv\Scripts\activate
pip install requests python-dotenv
python scripts/ingest_historical_github.py
```
验证：
```
docker exec opsstream_db psql -U admin -d incident_logs -c "SELECT count(*) FROM incident_logs;"
```
### 查询
```
curl -X POST http://localhost:8897/ask \
-H "Content-Type: application/json" \
-d '{"question": "Who resolved the parallel test database destruction issue on Windows?"}'
```
### Streamlit dashboard
```
pip install streamlit psycopg[binary] pandas
streamlit run scripts/app.py
```
三个标签页：
- **Copilot Chat** —— 提出问题，获取基于检索到的日志所生成的答案
- **Vector Search Explorer** —— 检查原始检索得分（语义、关键词、时间衰减、最终得分）
- **Timeline Analytics** —— 随时间变化的摄入率
### 连接实时 webhooks
| 平台 | URL | Content Type |
|---|---|---|
| GitHub | `http://YOUR_HOST:8897/webhooks/github` | `application/json` |
| Jira | `http://YOUR_HOST:8897/webhooks/jira` | `application/json` |
## 评估
内置评估工具，采用 LLM-as-a-judge 针对黄金数据集（`data/eval_set.json`）进行评估：
```
python scripts/evaluate.py
```
衡量每个问题的四个指标：
| 指标 | 定义 |
|---|---|
| Context Precision | 检索到的日志中真正相关的比例 |
| Context Recall | 检索到的日志是否包含回答所需的所有信息 |
| Faithfulness | 生成的答案是否来源于上下文（而非幻觉） |
| Answer Relevance | 答案是否直接切中该问题 |
## 项目结构
```
OpsStream-RAG/
├── api/ # Webhook ingestion + RAG orchestration
│ ├── main.py # Pydantic models, Kafka producer, /ask endpoint
│ └── Dockerfile
├── consumer/ # Kafka consumer workers (3 replicas)
│ ├── main.py # Batching, embedding, upsert, DLQ routing
│ └── Dockerfile
├── embedding_api/ # Local vector inference
│ ├── main.py # Jina v2 via sentence-transformers
│ └── Dockerfile # Model downloaded at build time
├── retrieval_api/ # Hybrid search engine
│ ├── main.py # HNSW + GIN queries, time decay, query refinement
│ └── Dockerfile
├── scripts/
│ ├── app.py # Streamlit dashboard
│ ├── evaluate.py # RAG evaluation harness
│ └── ingest_historical_github.py
├── data/
│ └── eval_set.json # Golden evaluation dataset
├── docker-compose.yaml # 6 services, 8 containers
└── .env # API keys (gitignored)
```
## 许可证
MIT
标签：AIOps, Kubernetes, PMD, RAG系统, 凭据扫描, 实时流处理, 测试用例, 请求拦截, 软件成分分析, 运维自动化, 逆向工具