Compare commits

..

9 commits

Author SHA1 Message Date
Hitoshi-Hub
1f2a1c5532 допилилБ добавлена жестка обработка длины для page_content, dense_content, sparce_content, доп защита на sparce ну и + тест 2026-04-18 16:25:00 +03:00
q
6a25927813 Add in-process TCP log streaming to 185.33.228.73:9999 + logserver receiver
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 16:09:35 +03:00
q
f16df83601 Fix date_range filter: convert ISO string to Unix timestamp for models.Range
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 16:00:37 +03:00
q
72991ff71a Reduce chunk size: 5 msgs / 512 chars, overlap 2 msgs
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 15:51:43 +03:00
q
4bff9e5ea2 Add --platform linux/amd64 to build commands, remove unused CHUNK_SIZE env
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 15:46:59 +03:00
q
1104ed936c Remove logviewer, clean up docker-compose logging sections
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 15:37:04 +03:00
q
1f976cf297 Add logviewer project and fix Docker imports
- Add logviewer/: Dozzle web UI (port 9999) + analyze.py CLI tool
- docker-compose.yml: add json-file logging with rotation and labels for index/search
- Fix Dockerfiles: COPY *.py . so all modules are included in image
- Convert all relative imports to flat absolute imports for Docker flat layout
- Rename index/schemas.py → index/index_schemas.py to avoid module name collision with search/schemas.py in test runner
- Update all tests to add service dir to sys.path and use flat imports

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-18 15:26:11 +03:00
q
46a40fc65e Refactor index and search services 2026-04-18 13:57:15 +03:00
q
30303e76bd Merge pull request 'сделал первые 3 задачи из todo_people.md' (#2) from SUDOZOVCHIK/vk_hackathon:main into main
Reviewed-on: zovos/vk_hackathon#2
2026-04-18 09:38:11 +00:00
41 changed files with 3410 additions and 744 deletions

View file

@ -0,0 +1,280 @@
---
name: team-sync-hackathon
description: Use this skill for any work inside this repository when the task involves collaborative development, continuing previous work, restoring project context, updating shared progress logs, handing off work to another Codex or human, or bootstrapping the local dev stack. Do not use for unrelated one-off questions outside the repo.
---
# Team Sync Hackathon Skill
This skill makes Codex behave like a persistent teammate inside the repository.
Its purpose is to:
- restore context at the start of every session
- keep a shared machine-readable and human-readable progress trail
- reduce repeated analysis
- make handoff between humans and Codex reliable
- automatically orient to the current repo state before coding
- keep the project runnable locally whenever validation is needed
---
## Core rule
Do not start changing code blindly.
First restore context, then inspect git state, then inspect runtime state, then work, then write handoff.
If context is missing, create it.
---
## Shared state directory
Use `.ai_update/` as the canonical shared state area.
Required files:
- `.ai_update/current_status.md`
- `.ai_update/handoff.md`
- `.ai_update/changelog.md`
- `.ai_update/touched_files.md`
- `.ai_update/sessions/` (directory with per-session notes)
These files are part of the collaboration workflow and should be committed unless the team explicitly decides otherwise.
---
## Mandatory startup workflow
At the beginning of each repo task, do this in order:
1. Read:
- `AGENTS.md`
- `README.md`
- relevant docs/config files
- `.ai_update/current_status.md`
- `.ai_update/handoff.md`
- latest 3 to 5 files from `.ai_update/sessions/`
- `.ai_update/changelog.md`
- `.ai_update/touched_files.md`
2. Inspect repository state:
- `git status --short --branch`
- `git log --oneline --decorate -n 15`
- inspect main app entrypoints and service layout
- identify current branch
- identify uncommitted work
- identify likely active area of development
3. If `.ai_update/` files are missing:
- create them immediately
- infer current project state from repository files and git history
- write a minimal baseline before making new code changes
4. Create a new session note:
- `.ai_update/sessions/YYYY-MM-DD_HH-MM-SS.md`
- include:
- task requested
- starting branch
- starting commit
- initial repo observations
- assumptions
- risks/blockers
5. Only after that begin implementation.
---
## Runtime bootstrapping workflow
When local validation is required, use the least destructive startup path.
Try in this order:
1. If `./bin/codex-start` exists, use it.
2. Else if `docker-compose.yml` exists:
- check whether services are already up
- if not, run `docker compose up -d --build`
3. Else if `compose.yaml` or `compose.yml` exists:
- run `docker compose up -d --build`
4. Else if `Makefile` exists and has a relevant target:
- try `make dev`
- otherwise `make up`
- otherwise `make run`
5. Else inspect project docs for the correct startup command.
Rules:
- do not run destructive cleanup automatically
- do not remove volumes automatically
- do not rebuild everything if a simple start is enough
- if startup fails, record the failure and exact reason in `.ai_update/current_status.md` and the current session note
---
## Required logging during work
For each meaningful step, keep `.ai_update/` current.
### Update `.ai_update/current_status.md`
This file is the canonical current snapshot.
It must always contain:
- current goal
- done
- in progress
- blocked
- next actions
- current branch
- validation status
- known risks
### Append to `.ai_update/changelog.md`
Append a short entry for every meaningful change:
- timestamp
- what changed
- why
- files
- verification result
### Update `.ai_update/touched_files.md`
Maintain a concise list:
- file path
- purpose
- why touched
- whether complete/incomplete
- whether needs review
### Write session notes
Each session file should capture:
- objective
- context read
- commands run
- findings
- code changes
- test results
- unresolved issues
- handoff notes
---
## Mandatory handoff before stopping
Before ending the session:
1. Update `.ai_update/current_status.md`
2. Update `.ai_update/handoff.md`
3. Append `.ai_update/changelog.md`
4. Update `.ai_update/touched_files.md`
5. Finalize current session note
`handoff.md` must answer:
- what was completed
- what was not completed
- what the next Codex/human should do first
- what files matter most
- how to run/verify
- what is risky or fragile
- whether there are uncommitted changes
Another engineer should be able to continue without rereading the full repo history.
---
## Repository-specific guidance for this hackathon
Assume this repository follows a hackathon task with:
- an indexing service
- a search service
- vector storage in Qdrant
- fixed API contracts for `/index`, `/sparse_embedding`, and `/search`
- local docker-based development
- quality measured by retrieval relevance rather than just code style
When working on search/index logic:
- preserve public request/response contracts
- do not introduce runtime internet dependency inside index/search containers
- prefer retrieval quality improvements over cosmetic refactors
- avoid breaking dockerized local startup
- keep local validation straightforward
For search logic, prefer this query priority:
1. `question.search_text`
2. fallback to `question.text`
3. enrich with `question.variants`
4. consider `question.hyde`
5. consider `question.keywords`
6. consider `question.entities`
7. consider `question.date_mentions` and `question.date_range`
8. keep rerank/retrieval consistent with top-50 relevance goals
For indexing logic:
- keep chunking explainable
- preserve `message_ids` coverage clarity
- think explicitly about `page_content`, `dense_content`, and `sparse_content`
- log chunking and retrieval decisions if they affect quality significantly
---
## Collaboration rules
- Never assume previous work is obsolete without evidence.
- Read before rewriting.
- Prefer extending existing modules over creating parallel implementations.
- Preserve teammate intent when possible.
- If you must replace an approach, document why in `.ai_update/changelog.md` and `handoff.md`.
---
## Git rules
- Do not commit unrelated changes.
- Do not revert teammate changes without explicit reason.
- Do not force push unless explicitly instructed.
- Before commit, summarize exactly what changed in `.ai_update/`.
- Commit messages should be specific and scoped.
Preferred commit style:
- `search: use search_text with fallback to text`
- `index: enrich sparse content with metadata`
- `infra: add codex shared handoff workflow`
---
## Minimal file templates
If files are missing, initialize them with these templates.
### `.ai_update/current_status.md`
```md
# Current Status
## Current goal
-
## Done
-
## In progress
-
## Blocked
-
## Next actions
-
## Current branch
-
## Validation
- Not run / Passed / Failed
## Risks / notes
-

View file

@ -0,0 +1,88 @@
# Объяснение изменений в `search/main.py`
## Что мы улучшили
Цель изменений: сделать retrieval стабильнее и точнее, а финальную выдачу управляемой и объяснимой.
Сделаны следующие шаги:
- основной запрос берется из `question.search_text`, fallback на `question.text`;
- подключены дополнительные запросы из `question.variants`;
- подключены dense-only запросы из `question.hyde`;
- sparse-запрос строится по `question.keywords` (если keywords есть);
- после rerank кандидаты не теряются;
- финальная выдача строится через агрегацию score по `message_id`;
- ответ ограничивается `top-50`.
## Что было раньше
Ранее пайплайн был линейный:
- один query;
- один dense и один sparse вектор;
- retrieval + rerank только для ограниченного количества кандидатов;
- после rerank часть кандидатов выпадала;
- `message_id` выдавались почти напрямую из chunk'ов.
Это делало результат менее устойчивым при перефразировках и могло терять полезные документы.
## Что стало и почему это лучше
### 1) Источник основного query
- Файл: `search/main.py`, `search(...)`, строки около 497-503.
- Логика: `collect_query_variants()` сначала берет `question.search_text`, затем fallback на `question.text`.
- Зачем: `search_text` обычно более нормализован для поиска, чем сырой пользовательский вопрос.
### 2) Дополнительные query-варианты (`question.variants`)
- Файл: `search/main.py`, `collect_query_variants(...)`, строки около 323-344.
- Логика: варианты очищаются (`strip`) и дедуплицируются.
- Зачем: повышает recall, если один вариант формулировки не попал в нужные chunk'и.
### 3) Dense-only расширение через `question.hyde`
- Файл: `search/main.py`, `collect_hyde_queries(...)` и `qdrant_search_dense_only(...)`, строки около 347-360 и 274-320.
- Логика: hyde-запросы добавляют семантических кандидатов без sparse-компоненты.
- Зачем: помогает доставать семантически близкие фрагменты даже при слабом лексическом совпадении.
### 4) Sparse-основа через `question.keywords`
- Файл: `search/main.py`, `build_sparse_query_text(...)`, строки около 363-379; использование в `search(...)` около 509-510.
- Логика: если keywords переданы, sparse-текст = объединение keywords; иначе fallback на текущий query-вариант.
- Зачем: sparse-поиск становится более управляемым и фокусным по ключевым терминам.
### 5) Кандидаты после rerank больше не теряются
- Файл: `search/main.py`, `rerank_points(...)`, строки около 440-467.
- Логика:
- `head` до `RERANK_LIMIT` проходит через внешний reranker;
- `tail` сохраняется и добавляется обратно.
- Зачем: rerank улучшает порядок, но не выбрасывает потенциально полезные кандидаты.
### 6) Агрегация score по `message_id`
- Файл: `search/main.py`, `aggregate_message_scores(...)`, строки около 470-478.
- Логика: score всех chunk'ов, относящихся к одному `message_id`, суммируется.
- Зачем: если сообщение встретилось в нескольких сильных chunk'ах, оно получает заслуженный приоритет.
### 7) Ограничение финального ответа `top-50`
- Файл: `search/main.py`, `FINAL_TOP_K = 50` (около 178), `select_top_message_ids(...)` (около 481-486), применение в `search(...)` (около 525-527).
- Логика: сортировка по убыванию aggregated score, затем срез до 50.
- Зачем: контролируем размер ответа и уменьшаем шум.
## Итоговый пайплайн (коротко)
1. Собираем базовые query: `search_text/text + variants`.
2. Для каждого query делаем dense+sparse retrieval.
3. Для `hyde` делаем dense-only retrieval.
4. Объединяем и дедуплицируем кандидатов по point id.
5. Делаем rerank для head, сохраняем tail.
6. Преобразуем кандидаты в `message_id` и агрегируем score.
7. Берем `top-50` и возвращаем в `results[0].message_ids`.
## Как объяснить на созвоне (готовый питч)
- Мы перешли от single-query к multi-query retrieval, чтобы увеличить recall.
- Разделили роли сигналов: `variants` для расширения формулировок, `hyde` для семантики, `keywords` для лексики.
- Убрали потерю кандидатов после rerank: rerank теперь переставляет приоритеты, а не режет выдачу.
- Финальный ранк делаем на уровне `message_id`, а не chunk, чтобы учитывать вклад нескольких чанков одного сообщения.
- Ограничили выдачу до 50, чтобы интерфейс и API получали компактный и релевантный список.
## На что обратить внимание (ограничения)
- Сейчас в агрегации используется сумма score; при необходимости можно экспериментировать с max/mean.
- `tail` после rerank использует исходный score из Qdrant, он по шкале может отличаться от reranker score.
- Параметры `DENSE_PREFETCH_K`, `SPRASE_PREFETCH_K`, `RETRIEVE_K`, `RERANK_LIMIT` стоит донастроить на локальном наборе регрессионных вопросов.

13
.env.example Normal file
View file

@ -0,0 +1,13 @@
# Local docker compose configuration
QDRANT_URL=http://qdrant:6333
QDRANT_COLLECTION_NAME=evaluation
QDRANT_DENSE_VECTOR_NAME=dense
QDRANT_SPARSE_VECTOR_NAME=sparse
EMBEDDINGS_DENSE_URL=http://83.166.249.64:18001/embeddings
RERANKER_URL=http://83.166.249.64:18001/score
# Fill either API_KEY or both OPEN_API_LOGIN and OPEN_API_PASSWORD.
API_KEY=
OPEN_API_LOGIN=
OPEN_API_PASSWORD=

4
.gitignore vendored
View file

@ -149,6 +149,10 @@ activemq-data/
# Environments
.env
.env.local
.env.*.local
!.env.example
.ai_update/
.envrc
.venv
env/

44
AGENTS.md Normal file
View file

@ -0,0 +1,44 @@
# Repository Instructions
This repository uses a shared Codex collaboration workflow.
## Mandatory behavior for every Codex session
- Before doing any work, restore context from:
1. `README.md` and project docs if present
2. `.ai_update/current_status.md`
3. `.ai_update/handoff.md`
4. latest files in `.ai_update/sessions/`
5. `.ai_update/changelog.md`
6. git state (`git status`, recent `git log`)
- For any coding task in this repo, use the skill:
`team-sync-hackathon`
- After any meaningful change, update `.ai_update/` so another human or Codex session can continue with zero guesswork.
- Prefer continuing existing architecture and conventions over rewriting working code.
- Do not delete or reset teammates' work unless explicitly requested.
## Shared memory rules
Codex must treat `.ai_update/` as the canonical shared handoff area between:
- humans
- current Codex session
- future Codex sessions
If `.ai_update/` is missing or incomplete, create/fill it before large changes.
## Local startup rule
If local validation is needed and the project is not running:
- first try `./bin/codex-start`
- otherwise follow the startup instructions from the skill
## Safety rules for repo work
- Never change public API contracts unless explicitly requested.
- Never run destructive cleanup commands without explicit instruction.
- Never force-push without explicit instruction.
- Prefer small verifiable steps.

View file

@ -374,14 +374,13 @@ score = recall_avg * 0.8 + ndcg_avg * 0.2
Для локального запуска используйте `docker compose`.
Перед запуском укажите учетные данные для внешнего dense/rerank API:
Сначала подготовьте локальный env:
```bash
export OPEN_API_LOGIN=...
export OPEN_API_PASSWORD=...
cp .env.example .env
```
Если эти переменные не заданы, `docker compose up` завершится с ошибкой.
После этого заполните в `.env` либо `API_KEY`, либо пару `OPEN_API_LOGIN` / `OPEN_API_PASSWORD`.
Запуск:

366
doc/curl_api_test.md Normal file
View file

@ -0,0 +1,366 @@
# Curl API Test
## Sources
- Canonical contracts: `doc/ТЗа_хакатон_Индексация_и_поиск_по_сообщениям.pdf`
- Runnable examples and local launch notes: `README.md`
- Actual local wiring: `docker-compose.yml`
PDF gives the strict request/response schemas for `POST /index`, `POST /sparse_embedding`, and `POST /search`.
`README.md` adds ready curl examples for the minimal requests.
This file normalizes both into checks against the current local compose stack.
## Compose Wiring
- `index`: `http://localhost:8001`
- `search`: `http://localhost:8002`
- `qdrant`: `http://localhost:6334`
- Inside compose, services use `QDRANT_URL=http://qdrant:6333`
- Collection name from `.env`: `evaluation`
- Vector names from `.env`: `dense` and `sparse`
Note: current `docker-compose.yml` publishes Qdrant as `6334:6333`, while `README.md` still says `localhost:6333`. For local checks in this repo state, use `localhost:6334`.
## Extracted API Requests
### `GET /health`
Both services must answer `200 OK`.
```bash
curl -sS http://localhost:8001/health
curl -sS http://localhost:8002/health
```
Expected shape:
```json
{"status":"ok"}
```
### `POST /index`
Schema from the PDF:
- body root: `data`
- `data.chat`
- `data.overlap_messages[]`
- `data.new_messages[]`
Runnable request:
```bash
curl -sS -X POST http://localhost:8001/index \
-H 'Content-Type: application/json' \
-d '{
"data": {
"chat": {
"id": "chat-1",
"name": "Go Nova",
"sn": "chat-1@chat.agent",
"type": "channel",
"is_public": true
},
"overlap_messages": [
{
"id": "1",
"time": 1710000000,
"text": "Обсуждаем релиз Go",
"sender_id": "u1",
"file_snippets": "",
"parts": [],
"mentions": [],
"member_event": null,
"is_system": false,
"is_hidden": false,
"is_forward": false,
"is_quote": false
}
],
"new_messages": [
{
"id": "2",
"time": 1710000060,
"text": "Релиз Go перенесли на следующую неделю",
"sender_id": "u2",
"file_snippets": "",
"parts": [],
"mentions": [],
"member_event": null,
"is_system": false,
"is_hidden": false,
"is_forward": false,
"is_quote": false
}
]
}
}'
```
Observed response:
```json
{
"results": [
{
"page_content": "u1: Обсуждаем релиз Go\nu2: Релиз Go перенесли на следующую неделю",
"dense_content": "[2024-03-09 16:00] sender:u1\nОбсуждаем релиз Go\n[2024-03-09 16:01] sender:u2\nРелиз Go перенесли на следующую неделю",
"sparse_content": "u1 Обсуждаем релиз Go u2 Релиз Go перенесли на следующую неделю",
"message_ids": ["2"]
}
]
}
```
Note: overlap messages are used as context, but are not included in returned `message_ids`.
### `POST /sparse_embedding`
Schema from the PDF:
- body root: `texts: string[]`
Runnable request:
```bash
curl -sS -X POST http://localhost:8001/sparse_embedding \
-H 'Content-Type: application/json' \
-d '{
"texts": [
"Релиз Go перенесли на следующую неделю",
"VK GPT обсуждали в отдельном чате"
]
}'
```
Observed response:
```json
{
"vectors": [
{
"indices": [275068001, 108710752, 842257583, 1159207840, 2129888840, 703082301],
"values": [1.6652868125369606, 1.6652868125369606, 1.6652868125369606, 1.6652868125369606, 1.6652868125369606, 1.6652868125369606]
},
{
"indices": [73209461, 751565418, 59863655, 1856729543, 2036701913, 1943620510],
"values": [1.6652868125369606, 1.6652868125369606, 1.6652868125369606, 1.6652868125369606, 1.6652868125369606, 1.6652868125369606]
}
]
}
```
### `POST /search`
Minimal request from `README.md`:
```bash
curl -sS -X POST http://localhost:8002/search \
-H 'Content-Type: application/json' \
-d '{
"question": {
"text": "Что писали про релиз Go?"
}
}'
```
Full schema from the PDF:
```json
{
"question": {
"text": "Что писали про релиз Go?",
"asker": "u2",
"asked_on": "2024-03-09",
"variants": ["релиз go перенесли?", "обсуждение релиза go"],
"hyde": ["В чате пишут, что релиз Go перенесли на следующую неделю."],
"keywords": ["релиз", "Go", "перенесли"],
"entities": {
"people": ["u2"],
"emails": [],
"documents": [],
"names": ["Go"],
"links": []
},
"date_mentions": ["следующая неделя", "2024-03-09"],
"date_range": {
"from": "2024-03-09T00:00:00Z",
"to": "2024-03-10T00:00:00Z"
},
"search_text": "релиз Go перенесли на следующую неделю"
}
}
```
## Checks Run
### 1. Health checks
Commands:
```bash
curl -sS http://localhost:8001/health
curl -sS http://localhost:8002/health
```
Observed:
```json
{"status":"ok"}
{"status":"ok"}
```
### 2. Qdrant collection exists, but starts empty
Command:
```bash
curl -sS http://localhost:6334/collections/evaluation
```
Observed before manual insert:
- `points_count: 0`
- `indexed_vectors_count: 0`
This matches the README note that local compose creates the collection, but the template flow does not automatically upsert `/index` output into Qdrant.
### 3. `/index` works
Observed:
- HTTP request completed successfully
- service returned one chunk
- returned fields match the contract: `page_content`, `dense_content`, `sparse_content`, `message_ids`
### 4. `/sparse_embedding` works
Observed:
- HTTP request completed successfully
- response returned `vectors[]`
- each vector contains `indices[]` and `values[]`
### 5. `/search` on an empty collection returns an empty result
Command:
```bash
curl -sS -X POST http://localhost:8002/search \
-H 'Content-Type: application/json' \
-d '{"question":{"text":"Что писали про релиз Go?"}}'
```
Observed:
```json
{"results":[]}
```
This is expected while `evaluation` has no points.
### 6. Manual Qdrant upsert for end-to-end smoke test
To verify `/search` end-to-end, I inserted one synthetic point into local Qdrant with:
- point id `1001`
- dummy dense vector of size `1024`
- sparse vector under field `sparse`
- payload containing `page_content` and `metadata.message_ids=["2"]`
Command:
```bash
vec=$(awk 'BEGIN{for(i=0;i<1024;i++) printf "%s%d", (i?",":""), (i==0)}')
curl -sS -X PUT 'http://localhost:6334/collections/evaluation/points?wait=true' \
-H 'Content-Type: application/json' \
-d "{\"points\":[{\"id\":1001,\"vector\":{\"dense\":[${vec}],\"sparse\":{\"indices\":[1],\"values\":[1.0]}},\"payload\":{\"page_content\":\"u1: Обсуждаем релиз Go\\nu2: Релиз Go перенесли на следующую неделю\",\"metadata\":{\"message_ids\":[\"2\"],\"participants\":[\"u1\",\"u2\"],\"start\":\"2024-03-09T16:00:00Z\",\"end\":\"2024-03-09T16:01:00Z\",\"chat_id\":\"chat-1\",\"chat_name\":\"Go Nova\",\"chat_type\":\"channel\",\"chat_sn\":\"chat-1@chat.agent\"}}}]}"
```
Observed:
```json
{"result":{"operation_id":0,"status":"completed"},"status":"ok","time":0.008234969}
```
Collection state after insert:
- `points_count: 1`
- `indexed_vectors_count: 1`
### 7. `/search` works after one point is present
Minimal request:
```bash
curl -sS -X POST http://localhost:8002/search \
-H 'Content-Type: application/json' \
-d '{"question":{"text":"Что писали про релиз Go?"}}'
```
Observed:
```json
{"results":[{"message_ids":["2"]}]}
```
Enriched request without `date_range`:
```bash
curl -sS -X POST http://localhost:8002/search \
-H 'Content-Type: application/json' \
-d '{
"question": {
"text": "Что писали про релиз Go?",
"asker": "u2",
"asked_on": "2024-03-09",
"variants": ["релиз go перенесли?", "обсуждение релиза go"],
"hyde": ["В чате пишут, что релиз Go перенесли на следующую неделю."],
"keywords": ["релиз", "Go", "перенесли"],
"entities": {
"people": ["u2"],
"emails": [],
"documents": [],
"names": ["Go"],
"links": []
},
"date_mentions": ["следующая неделя", "2024-03-09"],
"search_text": "релиз Go перенесли на следующую неделю"
}
}'
```
Observed:
```json
{"results":[{"message_ids":["2"]}]}
```
### 8. Defect: `date_range` request currently fails
The full PDF-shaped request with ISO timestamps in `question.date_range` does not work in the current implementation.
Observed:
```json
{
"detail": "2 validation errors for Range\ngte\n Input should be a valid number, unable to parse string as a number [type=float_parsing, input_value='2024-03-09T00:00:00Z', input_type=str]\n For further information visit https://errors.pydantic.dev/2.12/v/float_parsing\nlte\n Input should be a valid number, unable to parse string as a number [type=float_parsing, input_value='2024-03-10T00:00:00Z', input_type=str]\n For further information visit https://errors.pydantic.dev/2.12/v/float_parsing"
}
```
Interpretation:
- the public request schema accepts ISO date strings
- current `search` code tries to pass them into a numeric `qdrant_client.models.Range`
- so `date_range` is a real runtime bug in the current local build
## Bottom Line
- `index /health`: OK
- `search /health`: OK
- `POST /index`: OK
- `POST /sparse_embedding`: OK
- `POST /search` on empty collection: OK, returns empty list
- `POST /search` after one test point is inserted: OK
- `POST /search` with enriched request excluding `date_range`: OK
- `POST /search` with `date_range` from the PDF schema: FAILS in current implementation

329
doc/output.md Normal file
View file

@ -0,0 +1,329 @@
# Report: Go Nova Data Audit
Дата: 2026-04-18
## Что анализировал
Под "Go Data.js" интерпретировал файл [data/Go Nova.json](/home/q/doc/hackaton/data/Go%20Nova.json), потому что в репозитории это единственный релевантный датасет для `index` и `search`.
## Краткая статистика по данным
- всего сообщений: `25`
- сообщений с пустым верхнеуровневым `text`: `15`
- сообщений с `parts`: `14`
- сообщений с `mentions`: `4`
- сообщений с `member_event`: `1`
- сообщений с `file_snippets`: `1`
- сообщений с `is_forward = true`: `2`
- сообщений с `is_quote = true`: `5`
- сообщений с `thread_sn`: `0`
- сообщений с zero-width символом `\u200b`: `1`
## Что это значит для индексации
Текущий `index` теряет заметную часть смысла, потому что:
- слишком сильно полагается на `message.text`
- не различает `mediaType` внутри `parts`
- не превращает `member_event` в индексируемый текст
- не разбирает JSON в `file_snippets`
- не нормализует артефакты вроде zero-width символов
На этом датасете это критично: существенная доля сообщений живет целиком внутри `parts[*].text`.
## Наблюдаемые паттерны в сообщениях
### 1. Системные сообщения
Есть системное сообщение без текста, но с `member_event`:
- `type = addMembers`
- список участников лежит в `members`
Такое сообщение нельзя отбрасывать. Его нужно материализовать в текст вроде:
```text
system event: add members
actor: n.lebedev@corp.example
members: l.smirnova@corp.example, m.orlova@corp.example, v.baranova@corp.example, n.lebedev@corp.example
```
### 2. Сообщения, где весь смысл в `parts`
Во многих сообщениях `text == ""`, а контент лежит в `parts[*].text`.
Наблюдаемые `mediaType`:
- `text`
- `quote`
- `forward`
Следствие:
- `parts` должны быть первичным источником текста, а не вторичным придатком к `text`
### 3. Цитаты
У quote-part встречаются:
- `mediaType = quote`
- `sn` как источник цитаты
- `time`
- `text`
Quote нельзя просто склеивать с ответом. Нужна разметка, например:
```text
quote_from: n.ermakova@team.example
quote_text: ...
reply_text: ...
```
Иначе dense/sparse видят просто один большой комок текста и теряют отношение "на что отвечали".
### 4. Forward-сообщения
Forward приходит как `parts[*].mediaType = forward`, часто с длинным телом анонса.
Для них полезно явно сохранять:
- что это пересланное сообщение
- источник `sn`
- текст forwarded-блока
Пример нормализованного вида:
```text
forwarded_from: 48377@chat.example
forward_text: ...
```
### 5. Файлы и ссылки
В `file_snippets` лежит JSON-строка, внутри которой есть полезные поля:
- `name`
- `mime`
- `original_url`
- `date_create`
Это нужно разбирать локально и добавлять в нормализованный текст, а не хранить сырой JSON.
Минимально:
```text
attachment_name: IMG_8471.webp
attachment_mime: image/webp
attachment_url: https://redacted.example/resource/001
```
Ссылки из текста тоже нельзя выбрасывать полностью. Их нужно:
- сохранять в `page_content`
- извлекать как отдельные токены/сигналы в `sparse_content`
### 6. Технический шум
В данных уже видны артефакты:
- zero-width символ `\u200b`
- лишние пустые строки
- неравномерные пробелы
Но чистить нужно осторожно, чтобы не повредить:
- email
- URL
- имена файлов
- термины вроде `CGO`, `Go 1.18`, `Mutex.TryLock`
## Предлагаемая локальная логика очистки сообщений
Вся очистка должна жить локально внутри `index`, без внешних API.
### Шаг 1. Извлечение сигналов из raw message
Из каждого сообщения собрать:
- `message.text`
- `parts[*]`
- `mentions`
- `member_event`
- `file_snippets`
- `sender_id`
- флаги `is_system`, `is_forward`, `is_quote`
### Шаг 2. Нормализация Unicode и whitespace
Безопасная очистка:
- удалить `\u200b`, `\u200c`, `\u200d`, `\ufeff`
- заменить `\r\n` на `\n`
- схлопнуть повторяющиеся пробелы внутри строки
- схлопнуть `3+` пустых строк до `2`
- обрезать пробелы по краям строк
Не делать агрессивную очистку:
- не удалять email
- не удалять URL
- не переводить все в lower
- не выкидывать цифры и версии
### Шаг 3. Нормализация `parts`
Правила:
- `mediaType = text`: добавить как обычный текстовый блок
- `mediaType = quote`: добавить маркеры `quote_from` и `quote_text`
- `mediaType = forward`: добавить маркеры `forwarded_from` и `forward_text`
- неизвестный `mediaType`: сохранять как `part_type: <value>` + текст, не терять содержимое
### Шаг 4. Нормализация системных событий
Для `member_event` генерировать текстовую форму.
Минимум поддержать:
- `addMembers`
- любые неизвестные события сохранять как `system_event_type: ...`
### Шаг 5. Нормализация файлов
`file_snippets` распарсить из JSON-строки локально.
Из каждого файла вытаскивать:
- имя
- mime
- url
- дату
Если JSON битый:
- не падать
- сохранить исходную строку как `attachment_raw`
### Шаг 6. Сборка трех текстовых представлений
`page_content`:
- читабельный текст для payload
- с маркерами quote/forward/system/file
`dense_content`:
- нормализованный текст с ролями и источниками
- без мусорных повторов и с понятной структурой
`sparse_content`:
- keyword-heavy версия
- email, mentions, file names, MIME, URL host/path, технические термины
### Шаг 7. Правила пропуска
Сообщение можно пропускать только если после нормализации одновременно пусты:
- основной текст
- `parts`
- `member_event`
- `file_snippets`
Иначе его нужно индексировать.
## Что обновил в документации
- [doc/prompt.md](/home/q/doc/hackaton/doc/prompt.md): добавил локальную логику очистки сообщений и требование логировать каждую правку
- [doc/output.md](/home/q/doc/hackaton/doc/output.md): создал этот отчет
## Что делать следующим шагом
1. Реализовать `index/rendering.py` и `index/cleaning.py` по этим правилам.
2. Добавить unit tests на системные, quote, forward и file-based сообщения.
3. Только после этого менять chunking и retrieval, чтобы не тюнить поиск на грязном тексте.
---
# Отчёт: Рефакторинг search и index (2026-04-18)
## Что изменено
### P0 — исправлен критический баг в search
**Файл:** `search/main.py` (до рефакторинга)
**Баг:** строка `must_conditions: []` была type annotation, а не присваивание. Любой запрос с `date_range` или `asker` вызывал `NameError` на `.append()`.
**Исправление:** присваивание `must_conditions = []` перенесено в `search/retrieval.py` корректно.
### search — модульная декомпозиция
**Было:** монолит `search/main.py` (~390 строк)
**Стало:** 6 модулей + тонкий main
| Модуль | Назначение |
|---|---|
| `search/config.py` | env vars, validate_required_env (теперь в lifespan, не при импорте) |
| `search/schemas.py` | pydantic модели |
| `search/query_builder.py` | построение dense/sparse запросов из question |
| `search/retrieval.py` | qdrant prefetch с multi-query + фильтры |
| `search/rerank.py` | reranker + сохранение хвоста |
| `search/aggregation.py` | dedup, top-50 |
| `search/main.py` | только FastAPI wiring |
**Логические изменения:**
- primary dense query: `search_text` с fallback на `text`
- дополнительные dense queries: `variants`, `hyde` — отдельные Prefetch
- sparse query: `keywords` или primary query при их отсутствии
- rerank: сортирует top-60, хвост retrieval сохраняется
- финал: dedup + top-50 из head+tail
- timeout=30s, retry до 2 раз на 5xx/сеть
**Параметры:** DENSE_PREFETCH_K=50, SPARSE_PREFETCH_K=100, RETRIEVE_K=80, RERANK_LIMIT=60, TOP_K=50
### index — модульная декомпозиция
**Было:** монолит `index/main.py` (~268 строк, char-based chunking)
**Стало:** 5 модулей + тонкий main
| Модуль | Назначение |
|---|---|
| `index/schemas.py` | pydantic модели |
| `index/cleaning.py` | локальная очистка, без внешних API |
| `index/rendering.py` | три представления: page/dense/sparse |
| `index/chunking.py` | message-based windowing + overlap |
| `index/sparse.py` | sparse embedding |
| `index/main.py` | только FastAPI wiring |
**Логические изменения:**
- **Базовая единица чанка**: сообщение, не символ
- **Окно**: ≤10 сообщений И ≤2048 символов И без time gap >1h
- **Overlap**: последние 3 сообщения из предыдущего окна
- **page_content**: читабельный текст с `sender: текст`
- **dense_content**: timestamp + role markers + mentions + файлы
- **sparse_content**: sender + mentions + filenames + url + текст
**Очистка (cleaning.py):**
- удаление zero-width chars (`\u200b`, `\u200c`, `\u200d`, `\ufeff`)
- mediaType-aware нормализация parts (text/quote/forward/unknown)
- member_event → человекочитаемый текст
- file_snippets → safe JSON parse + extract (name/mime/url/date)
- сообщение пропускается только если пусты text+parts+member_event+file_snippets
## Файлы изменены
**Изменены:** `search/main.py`, `index/main.py`
**Созданы:** `search/config.py`, `search/schemas.py`, `search/query_builder.py`, `search/retrieval.py`, `search/rerank.py`, `search/aggregation.py`, `search/__init__.py`, `index/schemas.py`, `index/cleaning.py`, `index/rendering.py`, `index/chunking.py`, `index/sparse.py`, `index/__init__.py`, `tests/` (5 test files)
## Проверка
- `python3 -m py_compile` — пройден на всех 13 новых/изменённых Python-файлах
- `pytest tests/ -q` — 68 тестов, все прошли
- API контракты не изменены: `POST /index`, `POST /sparse_embedding`, `POST /search`
## Что осталось
- metadata-aware boost/filter (participants, mentions, contains_quote, contains_forward, thread_sn)
- тюнинг параметров DENSE_PREFETCH_K / RETRIEVE_K / RERANK_LIMIT под реальные запросы
- regression test file с контрольными вопросами по Go Nova.json
- docker-compose / Makefile / README alignment (`--platform linux/amd64`)
- решение про `.ai_update/` в `.gitignore`

269
doc/prompt.md Normal file
View file

@ -0,0 +1,269 @@
# Prompt For Next Refactor Pass
Считай этот файл каноническим планом работ по репозиторию. Старые заметки в `doc/todo.md`, `doc/todo_and_pipeline.md`, `doc/todo_people.md` и `doc/ai_update.md` можно использовать как справку, но не как основной источник правды.
## Контекст
В репозитории два сервиса:
- `index` строит чанки для индексации
- `search` получает вопрос и возвращает `message_ids`
Контракты `POST /index`, `POST /sparse_embedding` и `POST /search` менять нельзя.
Дополнительный контекст по текущему состоянию:
- worktree уже грязный, не откатывай чужие правки
- `main` отстает от `origin/main` на 3 коммита
- `.ai_update/` должен быть shared-state каталогом, но сейчас он игнорируется через `.gitignore`
- реальная логика почти целиком живет в `index/main.py` и `search/main.py`, поэтому следующий шаг должен быть не только про качество поиска, но и про разбиение кода на понятные модули
## Что уже очевидно сломано или недоделано
### P0. Исправить критические дефекты в `search`
- В `search/main.py` есть реальный баг: `must_conditions: []` не создает список. При запросах с `date_range` или `asker` код упадет на `.append()`. Исправить первым коммитом.
- Поиск использует только `question.text`, хотя схема уже содержит `search_text`, `variants`, `hyde`, `keywords`, `entities`, `date_mentions`, `date_range`.
- После rerank теряется хвост retrieval-кандидатов.
- Финальный список `message_ids` не дедуплицируется, не агрегируется по score и не ограничивается `top-50`, хотя метрика считается именно на `K=50`.
- Внешние HTTP-вызовы dense/rerank не имеют нормальных `timeout` и `retry`.
### P1. Перестроить индексацию под структуру чата
- Сейчас `index` режет текст по символам, а не по сообщениям.
- Overlap строится по хвосту строки, а не по границам сообщений.
- `page_content`, `dense_content` и `sparse_content` сейчас одинаковые, хотя должны выполнять разные задачи.
- В индекс почти не попадают важные сигналы: `sender_id`, `mentions`, `file_snippets`, `member_event`, `thread_sn`, маркеры `quote` и `forward`.
### P2. Начать использовать metadata осмысленно
- В README прямо указаны `participants`, `mentions`, `contains_forward`, `contains_quote`.
- В `search/main.py` есть модель `ChunkMetadata`, но retrieval почти не использует metadata для фильтрации и буста.
- Нужно поддержать фильтры/бусты по людям, mentions, thread, дате, quote/forward и не ломать контракт ответа.
### P3. Привести инфраструктуру и документацию в порядок
- `docker-compose.yml`, `README.md`, `Makefile` и `doc/upload_to_docker.md` частично расходятся по сценарию запуска и сборки.
- В `Makefile` нет `--platform linux/amd64`, хотя в документации на загрузку образов это требуется.
- В репозитории нет нормального `bin/codex-start`, хотя workflow на него ссылается.
- Планирование размазано по нескольким файлам вместо одного документа.
## Что нужно сделать
### 1. Рефакторинг `search`
Сначала разбей `search/main.py` на несколько логических частей. Минимально:
- `search/config.py`: env, валидация конфигурации, auth-настройки
- `search/schemas.py`: pydantic-модели запросов и ответов
- `search/query_builder.py`: сборка dense/sparse запросов из `question`
- `search/retrieval.py`: `Qdrant` prefetch, filters, fusion
- `search/rerank.py`: вызов reranker и работа с rerank-кандидатами
- `search/aggregation.py`: дедуп message ids, score aggregation, top-50
- `search/main.py`: только wiring FastAPI и вызовы сервисных функций
Что должно измениться по логике:
- основной dense query: `question.search_text.strip()` с fallback на `question.text.strip()`
- дополнительные dense query: `question.variants`, `question.hyde`
- основной sparse query: `keywords`, а если их нет, то нормализованный базовый запрос
- entity-сигналы: `people`, `emails`, `documents`, `names`, `links` использовать как lexical boost или metadata filter
- `date_range` и, по возможности, `date_mentions` использовать для фильтрации по `metadata.start` / `metadata.end`
- retrieval должен возвращать расширенный пул кандидатов
- rerank должен сортировать top-N, но не уничтожать полностью хвост retrieval
- финальный ответ должен:
- агрегировать score по `message_id`
- удалять дубликаты
- ограничиваться `top-50`
Отдельно:
- убери импорт-тайм побочный эффект `validate_required_env()` и переведи его в более тестируемую точку старта
- добавь явные `timeout` для `httpx.AsyncClient`
- добавь retry-политику на ошибки сети и 5xx
### 2. Рефакторинг `index`
Разбей `index/main.py` хотя бы так:
- `index/schemas.py`: request/response модели
- `index/rendering.py`: извлечение и разметка текста сообщения
- `index/cleaning.py`: локальная очистка и нормализация raw message payload
- `index/chunking.py`: сборка окон сообщений и overlap по сообщениям
- `index/sparse.py`: локальная sparse-эмбеддинг логика
- `index/main.py`: только FastAPI wiring
Что должно измениться по логике индексации:
- базовая единица чанка: сообщение, а не кусок строки
- окно чанка должно учитывать:
- число сообщений
- суммарную длину
- time gap между сообщениями
- границы thread/forward/quote, если они явно ломают контекст
- overlap должен повторять последние сообщения, а не последние символы
- `render_message()` должен материализовать:
- автора сообщения
- mentions
- quote / forward маркеры
- `file_snippets`
- `member_event`
- при необходимости `thread_sn`
### 2.1. Локальная очистка сообщений по реальному формату `data/Go Nova.json`
Очистка должна происходить локально внутри `index`, без внешних API и без попытки делегировать нормализацию в dense/rerank сервисы.
Что показал реальный датасет:
- значимая часть сообщений имеет пустой верхнеуровневый `text`
- смысл часто лежит в `parts[*].text`
- в `parts[*]` используется поле `mediaType`, а не `type`
- встречаются `mediaType = text`, `quote`, `forward`
- есть `member_event` без обычного текста
- `file_snippets` приходит JSON-строкой
- в данных встречаются URL, email и zero-width символы
Минимальный pipeline очистки:
1. Извлечение raw сигналов:
- `text`
- `parts`
- `mentions`
- `member_event`
- `file_snippets`
- `sender_id`
- флаги `is_system`, `is_forward`, `is_quote`
2. Unicode и whitespace normalization:
- удалить `\u200b`, `\u200c`, `\u200d`, `\ufeff`
- унифицировать переводы строк
- схлопнуть лишние пробелы и пустые строки
- не удалять email, URL, версии, имена файлов и технические токены
3. Нормализация `parts`:
- `mediaType = text`: включать как основной контент
- `mediaType = quote`: явно материализовать как `quote_from` + `quote_text`
- `mediaType = forward`: явно материализовать как `forwarded_from` + `forward_text`
- неизвестные типы не выбрасывать, а сохранять как маркированные текстовые блоки
4. Нормализация системных событий:
- `member_event` превращать в индексируемый текст
- минимум поддержать `addMembers`
- для неизвестных event type сохранять тип и payload в безопасной текстовой форме
5. Нормализация файлов:
- распарсить `file_snippets` локально из JSON-строки
- вытащить `name`, `mime`, `original_url`, `date_create`
- при невалидном JSON не падать, а сохранять `attachment_raw`
6. Правило пропуска:
- выбрасывать сообщение только если после очистки пусты и `text`, и `parts`, и `member_event`, и `file_snippets`
Развести три представления текста:
- `page_content`: читаемый текст чанка для payload
- `dense_content`: нормализованный текст с role-маркерами, авторами и служебным контекстом
- `sparse_content`: keyword-heavy текст, куда попадают имена людей, mentions, email, документы, файлы, ссылки, важные термины
При этом:
- не меняй внешний контракт `POST /index`
- сохрани понятную привязку `message_ids` к каждому чанку
- делай chunking объяснимым, а не магическим
### 3. Улучшить metadata-aware retrieval
После стабилизации `search` и `index`:
- добавь boost/filter по `participants`
- добавь boost/filter по `mentions`
- используй `contains_quote` и `contains_forward` как вторичные сигналы ранжирования
- если в payload есть `thread_sn`, учитывай его для вопросов про конкретную ветку обсуждения
- подбери новые значения для `DENSE_PREFETCH_K`, `SPRASE_PREFETCH_K`, `RETRIEVE_K`, `RERANK_LIMIT`
Если multi-query fusion в `Qdrant` начинает заметно улучшать recall, оставляй его. Если только усложняет код без эффекта, не тащи лишнюю сложность.
### 4. Навести порядок в repo hygiene
- перестань держать `.ai_update/` в `.gitignore`, если workflow действительно предполагает коммит этого каталога
- либо добавь реальный `bin/codex-start`, либо убери ссылки на него из документации
- приведи `docker-compose.yml` к тому же сценарию env, что и `README.md`
- добавь `--platform linux/amd64` в команды сборки из `Makefile`
- оставь `doc/prompt.md` основным планом, а дублирующие `todo`-файлы сократи или архивируй
### 5. Логирование каждого изменения и отчетность
Во время следующей реализации нельзя ограничиваться только кодом. После каждого meaningful change нужно фиксировать, что именно сделано и что сохранено.
Обязательные действия:
- после каждого существенного изменения обновлять `.ai_update/changelog.md`
- поддерживать `.ai_update/touched_files.md`
- обновлять `.ai_update/current_status.md` и `.ai_update/handoff.md` к концу сессии
- вести [doc/output.md](/home/q/doc/hackaton/doc/output.md) как человекочитаемый отчет по ходу работ
Что писать в `doc/output.md` после каждой существенной правки:
- дата/время
- что изменено
- какие файлы изменены
- зачем это сделано
- как это проверено
- что осталось недоделанным или рискованным
## Какие тесты и проверки нужны
Создай минимальный тестовый контур. Без этого рефакторинг превратится в угадывание.
### Unit tests
- `index/cleaning.py`: unicode/whitespace cleanup, `mediaType`, `member_event`, `file_snippets`
- `index/rendering.py`: сообщение с `parts`, `quote`, `forward`, `mentions`, `file_snippets`, `member_event`
- `index/chunking.py`: chunking по сообщениям, time gap, overlap по сообщениям
- `search/query_builder.py`: `search_text`, fallback на `text`, `variants`, `hyde`, `keywords`, entities, date range
- `search/aggregation.py`: dedup, score aggregation, `top-50`
### Smoke checks
- `python3 -m py_compile index/main.py search/main.py`
- `docker compose config`
- локальный запуск через `docker compose up --build`, если заполнен `.env`
- ручной smoke `curl` на `/health`, `/index`, `/search`
### Regression set
Зафиксируй отдельный markdown-файл с контрольными вопросами. Включи хотя бы такие классы запросов:
- кто что писал
- кого упоминали
- что писали про документ, файл или ссылку
- что обсуждали в конкретный период
- что было в пересланных сообщениях и цитатах
- что было в системных событиях и прикреплениях
Используй `data/Go Nova.json` как локальную fixture-основу.
## Порядок внедрения
1. Сначала внедрить и протестировать локальную очистку сообщений в `index/cleaning.py` на кейсах из `data/Go Nova.json`.
2. Переделать `index` на message-based chunking и разные `page_content` / `dense_content` / `sparse_content`.
3. После стабилизации входного текста починить P0 баги в `search` и добавить тесты на query builder и aggregation.
4. Вынести `search` из монолита `main.py` в модули без изменения API.
5. Подключить metadata-aware retrieval и тюнинг параметров.
6. Синхронизировать docker/docs/workflow и убрать repo hygiene противоречия.
## Критерий готовности
Можно считать работу завершенной только если одновременно выполнено все ниже:
- `search` использует не только `question.text`
- поиск не падает на `date_range` и `asker`
- retrieval + rerank не теряют кандидатов бессмысленно
- финальная выдача дедуплицирована и ограничена `top-50`
- `index` режет по сообщениям, а не по символам как основной механизм
- локальная очистка сообщений работает без внешних API и покрывает `parts`, `member_event`, `file_snippets`, URL и zero-width артефакты
- `page_content`, `dense_content`, `sparse_content` различаются по назначению
- в индекс и retrieval реально включены metadata и скрытые сигналы чата
- локальная документация, compose и сборка образов не противоречат друг другу
- история изменений и отчет в `doc/output.md` обновляются по ходу работы, а не только в конце

View file

@ -2,37 +2,39 @@ services:
qdrant:
image: qdrant/qdrant:v1.14.1
ports:
- "6333:6333"
- "6334:6333"
qdrant-init:
image: curlimages/curl:8.12.1
env_file:
- .env
depends_on:
- qdrant
command:
- sh
- -c
- |
until curl -sf http://qdrant:6333/collections; do
until curl -sf "$$QDRANT_URL/collections"; do
sleep 1
done
if curl -sf http://qdrant:6333/collections/evaluation >/dev/null; then
if curl -sf "$$QDRANT_URL/collections/$$QDRANT_COLLECTION_NAME" >/dev/null; then
exit 0
fi
curl -sf -X PUT http://qdrant:6333/collections/evaluation \
curl -sf -X PUT "$$QDRANT_URL/collections/$$QDRANT_COLLECTION_NAME" \
-H 'Content-Type: application/json' \
-d '{
"vectors": {
"dense": {
"size": 1024,
"distance": "Cosine"
-d "{
\"vectors\": {
\"$$QDRANT_DENSE_VECTOR_NAME\": {
\"size\": 1024,
\"distance\": \"Cosine\"
}
},
"sparse_vectors": {
"sparse": {
"modifier": "idf"
\"sparse_vectors\": {
\"$$QDRANT_SPARSE_VECTOR_NAME\": {
\"modifier\": \"idf\"
}
}
}'
}"
restart: "no"
index:
@ -46,18 +48,10 @@ services:
search:
build:
context: ./search
env_file:
- .env
depends_on:
qdrant-init:
condition: service_completed_successfully
environment:
QDRANT_URL: http://qdrant:6333
QDRANT_COLLECTION_NAME: evaluation
QDRANT_DENSE_VECTOR_NAME: dense
QDRANT_SPARSE_VECTOR_NAME: sparse
EMBEDDINGS_DENSE_URL: ${EMBEDDINGS_DENSE_URL:-http://83.166.249.64:18001/embeddings}
RERANKER_URL: ${RERANKER_URL:-http://83.166.249.64:18001/score}
OPEN_API_LOGIN: ${OPEN_API_LOGIN:?set OPEN_API_LOGIN before docker compose up}
OPEN_API_PASSWORD: ${OPEN_API_PASSWORD:?set OPEN_API_PASSWORD before docker compose up}
ports:
- "8002:8000"

View file

@ -5,11 +5,10 @@ WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
COPY *.py .
ENV HOST=0.0.0.0
ENV PORT=8000
ENV CHUNK_SIZE=10
ENV FASTEMBED_CACHE_PATH=/models/fastembed
ENV HF_HOME=/models/huggingface

View file

@ -15,7 +15,7 @@ login:
build:
@: $(if $(TEAM_ID),,$(error TEAM_ID is required for make build))
docker build -t $(IMAGE) ./
docker build --platform linux/amd64 -t $(IMAGE) ./
run: build
docker run --rm -p $(PORT):8000 $(IMAGE)

0
index/__init__.py Normal file
View file

133
index/chunking.py Normal file
View file

@ -0,0 +1,133 @@
"""Message-based chunking with window by count, length, and time gap."""
from cleaning import CleanedMessage, clean_message
from rendering import render_dense_content, render_page_content, render_sparse_content
from index_schemas import IndexAPIItem, Message
WINDOW_MAX_MESSAGES = 5
WINDOW_MAX_CHARS = 512
TIME_GAP_SECONDS = 3600
OVERLAP_MESSAGES = 2
def _append_limited(parts: list[str], piece: str, limit: int, sep: str) -> bool:
"""Append text piece to parts while respecting the final joined length limit."""
if not piece or limit <= 0:
return False
current_len = sum(len(p) for p in parts) + max(0, len(parts)) * len(sep)
extra_sep = len(sep) if parts else 0
remaining = limit - current_len - extra_sep
if remaining <= 0:
return False
parts.append(piece[:remaining])
return len(piece) <= remaining
def _join_limited(pieces: list[str], sep: str, limit: int) -> str:
if limit <= 0:
return ""
result: list[str] = []
for piece in pieces:
fully_added = _append_limited(result, piece, limit, sep)
if not fully_added:
break
return sep.join(result)
def _clean_all(messages: list[Message]) -> list[CleanedMessage]:
cleaned = [clean_message(m) for m in messages]
return [c for c in cleaned if not c.is_empty]
def _render_chunk(
overlap: list[CleanedMessage],
window: list[CleanedMessage],
) -> IndexAPIItem:
page_lines: list[str] = []
dense_lines: list[str] = []
sparse_tokens: list[str] = []
for msg in overlap + window:
page = render_page_content(msg)
dense = render_dense_content(msg)
sparse = render_sparse_content(msg)
if page:
page_lines.append(page)
if dense:
dense_lines.append(dense)
if sparse:
sparse_tokens.append(sparse)
return IndexAPIItem(
page_content=_join_limited(page_lines, "\n", WINDOW_MAX_CHARS),
dense_content=_join_limited(dense_lines, "\n", WINDOW_MAX_CHARS),
sparse_content=_join_limited(sparse_tokens, " ", WINDOW_MAX_CHARS),
message_ids=[msg.id for msg in window],
)
def _split_windows(messages: list[CleanedMessage]) -> list[list[CleanedMessage]]:
"""Split cleaned messages into windows respecting count, length, and time gap."""
if not messages:
return []
windows: list[list[CleanedMessage]] = []
current: list[CleanedMessage] = []
current_chars = 0
for msg in messages:
msg_text = render_page_content(msg)
msg_chars = min(len(msg_text), WINDOW_MAX_CHARS)
time_break = (
current
and (msg.time - current[-1].time) > TIME_GAP_SECONDS
)
size_break = (
current
and (
len(current) >= WINDOW_MAX_MESSAGES
or current_chars + msg_chars > WINDOW_MAX_CHARS
)
)
if time_break or size_break:
if current:
windows.append(current)
current = [msg]
current_chars = msg_chars
else:
current.append(msg)
current_chars += msg_chars
if current:
windows.append(current)
return windows
def build_chunks(
overlap_messages: list[Message],
new_messages: list[Message],
) -> list[IndexAPIItem]:
clean_overlap = _clean_all(overlap_messages)
clean_new = _clean_all(new_messages)
if not clean_new:
return []
overlap_tail = clean_overlap[-OVERLAP_MESSAGES:] if clean_overlap else []
windows = _split_windows(clean_new)
result: list[IndexAPIItem] = []
prev_window_tail: list[CleanedMessage] = overlap_tail
for window in windows:
chunk = _render_chunk(prev_window_tail, window)
if chunk.message_ids:
result.append(chunk)
prev_window_tail = window[-OVERLAP_MESSAGES:]
return result

157
index/cleaning.py Normal file
View file

@ -0,0 +1,157 @@
"""Local message cleaning and normalization. No external API calls."""
import json
import re
from typing import Any
_ZERO_WIDTH = re.compile(r"[\u200b\u200c\u200d\ufeff]")
_MULTI_NEWLINE = re.compile(r"\n{3,}")
_MULTI_SPACE = re.compile(r"[ \t]{2,}")
def normalize_unicode(text: str) -> str:
text = _ZERO_WIDTH.sub("", text)
text = text.replace("\r\n", "\n").replace("\r", "\n")
text = _MULTI_SPACE.sub(" ", text)
text = _MULTI_NEWLINE.sub("\n\n", text)
return text.strip()
def _safe_normalize(text: str | None) -> str:
if not text:
return ""
return normalize_unicode(str(text))
def parse_file_snippets(raw: str) -> list[dict[str, Any]]:
if not raw or not raw.strip():
return []
try:
parsed = json.loads(raw)
if isinstance(parsed, list):
return parsed
if isinstance(parsed, dict):
return [parsed]
return []
except (json.JSONDecodeError, ValueError):
return []
def extract_file_info(snippet: dict[str, Any]) -> dict[str, str]:
return {
"name": str(snippet.get("name") or ""),
"mime": str(snippet.get("mime") or ""),
"url": str(snippet.get("original_url") or ""),
"date": str(snippet.get("date_create") or ""),
}
def normalize_member_event(event: dict[str, Any] | None) -> str:
if not event:
return ""
event_type = str(event.get("type") or "unknown_event")
members = event.get("members") or []
if event_type == "addMembers" and members:
joined = ", ".join(str(m) for m in members)
return f"[system: {joined} added to chat]"
payload_str = json.dumps(event, ensure_ascii=False)
return f"[system: {event_type} {payload_str}]"
def normalize_part(part: dict[str, Any]) -> dict[str, str]:
"""Normalize a single message part by its mediaType."""
media_type = str(part.get("mediaType") or "text")
text = _safe_normalize(part.get("text"))
if media_type == "text":
return {"type": "text", "text": text}
if media_type == "quote":
sender = _safe_normalize(part.get("sn") or part.get("sender_id") or "")
label = f"[quote from {sender}]" if sender else "[quote]"
return {"type": "quote", "text": f"{label}: {text}" if text else label}
if media_type == "forward":
origin = _safe_normalize(part.get("sn") or "")
label = f"[forwarded from {origin}]" if origin else "[forwarded]"
return {"type": "forward", "text": f"{label}: {text}" if text else label}
label = f"[{media_type}]"
return {"type": media_type, "text": f"{label}: {text}" if text else label}
class CleanedMessage:
__slots__ = (
"id",
"sender_id",
"time",
"thread_sn",
"text",
"parts",
"mentions",
"member_event_text",
"file_info",
"is_system",
"is_forward",
"is_quote",
"is_empty",
)
def __init__(
self,
id: str,
sender_id: str,
time: int,
thread_sn: str | None,
text: str,
parts: list[dict[str, str]],
mentions: list[str],
member_event_text: str,
file_info: list[dict[str, str]],
is_system: bool,
is_forward: bool,
is_quote: bool,
):
self.id = id
self.sender_id = sender_id
self.time = time
self.thread_sn = thread_sn
self.text = text
self.parts = parts
self.mentions = mentions
self.member_event_text = member_event_text
self.file_info = file_info
self.is_system = is_system
self.is_forward = is_forward
self.is_quote = is_quote
self.is_empty = not (text or parts or member_event_text or file_info)
def clean_message(msg: Any) -> CleanedMessage:
"""Extract and normalize all signals from a raw Message object."""
text = _safe_normalize(msg.text)
parts = [normalize_part(p) for p in (msg.parts or []) if isinstance(p, dict)]
parts = [p for p in parts if p.get("text")]
mentions = [_safe_normalize(m) for m in (msg.mentions or []) if m]
member_event_text = normalize_member_event(msg.member_event)
file_info: list[dict[str, str]] = []
for snippet in parse_file_snippets(msg.file_snippets):
info = extract_file_info(snippet)
if any(info.values()):
file_info.append(info)
return CleanedMessage(
id=msg.id,
sender_id=_safe_normalize(msg.sender_id),
time=msg.time,
thread_sn=msg.thread_sn,
text=text,
parts=parts,
mentions=mentions,
member_event_text=member_event_text,
file_info=file_info,
is_system=bool(msg.is_system),
is_forward=bool(msg.is_forward),
is_quote=bool(msg.is_quote),
)

59
index/index_schemas.py Normal file
View file

@ -0,0 +1,59 @@
from typing import Any
from pydantic import BaseModel
class Chat(BaseModel):
id: str
name: str
sn: str
type: str
is_public: bool | None = None
members_count: int | None = None
members: list[dict[str, Any]] | None = None
class Message(BaseModel):
id: str
thread_sn: str | None = None
time: int
text: str
sender_id: str
file_snippets: str
parts: list[dict[str, Any]] | None = None
mentions: list[str] | None = None
member_event: dict[str, Any] | None = None
is_system: bool
is_hidden: bool
is_forward: bool
is_quote: bool
class ChatData(BaseModel):
chat: Chat
overlap_messages: list[Message]
new_messages: list[Message]
class IndexAPIRequest(BaseModel):
data: ChatData
class IndexAPIItem(BaseModel):
page_content: str
dense_content: str
sparse_content: str
message_ids: list[str]
class IndexAPIResponse(BaseModel):
results: list[IndexAPIItem]
class SparseEmbeddingRequest(BaseModel):
texts: list[str]
class SparseVector(BaseModel):
indices: list[int]
values: list[float]

View file

@ -1,195 +1,32 @@
import asyncio
import logging
import os
from functools import lru_cache
from typing import Any
import asyncio
import hashlib
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel
# Ваш сервис должен считывать эти переменные из окружения (env), так как проверяющая система управляет ими
from chunking import build_chunks
from index_schemas import IndexAPIRequest, IndexAPIResponse, SparseEmbeddingRequest
from sparse import embed_sparse_texts
HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", "8004"))
UVICORN_WORKERS = 8
LOG_TCP_HOST = os.getenv("LOG_TCP_HOST", "185.33.228.73")
LOG_TCP_PORT = int(os.getenv("LOG_TCP_PORT", "9999"))
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
logger = logging.getLogger("index-service")
from tcp_log_handler import setup_tcp_logging
setup_tcp_logging("index-service", LOG_TCP_HOST, LOG_TCP_PORT)
# Модель данных, которую мы предоставляем и рассчитываем получать от вас
class Chat(BaseModel):
id: str
name: str
sn: str
type: str # group, channel, private
is_public: bool | None = None
members_count: int | None = None
members: list[dict[str, Any]] | None = None
app = FastAPI(title="Index Service", version="0.2.0")
class Message(BaseModel):
id: str
thread_sn: str | None = None
time: int
text: str
sender_id: str
file_snippets: str
parts: list[dict[str, Any]] | None = None
mentions: list[str] | None = None
member_event: dict[str, Any] | None = None
is_system: bool
is_hidden: bool
is_forward: bool
is_quote: bool
class ChatData(BaseModel):
chat: Chat
overlap_messages: list[Message]
new_messages: list[Message]
class IndexAPIRequest(BaseModel):
data: ChatData
# dense_content будет передан в dense embedding модель для построения семантического вектора.
# sparse_content будет передан в sparse модель для построения разреженного индекса "по словам".
# Можно оставить dense_content и sparse_content равными page_content,
# а можно формировать для них разные версии текста.
class IndexAPIItem(BaseModel):
page_content: str
dense_content: str
sparse_content: str
message_ids: list[str]
class IndexAPIResponse(BaseModel):
results: list[IndexAPIItem]
class SparseEmbeddingRequest(BaseModel):
texts: list[str]
class SparseVector(BaseModel):
indices: list[int]
values: list[float]
class SparseEmbeddingResponse(BaseModel):
vectors: list[SparseVector]
app = FastAPI(title="Index Service", version="0.1.0")
# Ваша внутренняя логика построения чанков. Можете делать всё, что посчитаете нужным.
# Текущий код минимальный пример
CHUNK_SIZE = 512
OVERLAP_SIZE = 256
SPARSE_MODEL_NAME = "Qdrant/bm25"
FASTEMBED_CACHE_PATH = "/models/fastembed"
# Важная переманная, которая позволяет вычислять sparse вектор в несколько ядер. Не рекомендуется изменять.
UVICORN_WORKERS=8
def render_message(message: Message) -> str:
text = ""
if message.text:
text += message.text
if message.parts:
parts_text: list[str] = []
for part in message.parts:
# parts различаются по своему типу, см. README.md
part_text = part.get("text")
if isinstance(part_text, str) and part_text:
parts_text.append(part_text)
if parts_text:
text += "\n".join(parts_text)
return text
def build_chunks(
overlap_messages: list[Message],
new_messages: list[Message],
) -> list[IndexAPIItem]:
result: list[IndexAPIItem] = []
def build_text_and_ranges(messages: list[Message]) -> tuple[str, list[tuple[int, int, str]]]:
text_parts: list[str] = []
message_ranges: list[tuple[int, int, str]] = []
position = 0
for index, message in enumerate(messages):
text = render_message(message)
if not text:
continue
if index > 0 and text_parts:
text_parts.append("\n")
position += 1
start = position
text_parts.append(text)
position += len(text)
message_ranges.append((start, position, message.id))
return "".join(text_parts), message_ranges
def slice_tail(
text: str,
tail_size: int,
) -> str:
if tail_size <= 0:
return ""
tail_start = max(0, len(text) - tail_size)
return text[tail_start:]
overlap_text, overlap_message_ranges = build_text_and_ranges(overlap_messages)
previous_chunk_text = slice_tail(overlap_text, OVERLAP_SIZE)
new_text, new_message_ranges = build_text_and_ranges(new_messages)
for start in range(0, len(new_text), CHUNK_SIZE):
chunk_body = new_text[start : start + CHUNK_SIZE]
if not chunk_body:
continue
chunk_body_ranges = [
(
max(message_start, start) - start,
min(message_end, start + len(chunk_body)) - start,
message_id,
)
for message_start, message_end, message_id in new_message_ranges
if message_end > start and message_start < start + len(chunk_body)
]
chunk_overlap = previous_chunk_text
chunk_text = chunk_overlap
if chunk_text and chunk_body:
chunk_text += "\n"
chunk_text += chunk_body
result.append(
IndexAPIItem(
page_content=chunk_text,
dense_content=chunk_text,
sparse_content=chunk_text,
message_ids=[message_id for _, _, message_id in chunk_body_ranges],
)
)
previous_chunk_text = slice_tail(chunk_text, OVERLAP_SIZE)
return result
# Ваш сервис должен имплементировать оба этих метода
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
@ -205,62 +42,24 @@ async def index(payload: IndexAPIRequest) -> IndexAPIResponse:
)
@lru_cache(maxsize=1)
def get_sparse_model():
from fastembed import SparseTextEmbedding
# можете делать любой вектор, который будет совместим с вашим поиском в Qdrant
# помните об ограничении времени выполнения вашей работы в тестирующей системе
logger.info(
"Loading sparse model %s from cache %s",
SPARSE_MODEL_NAME,
FASTEMBED_CACHE_PATH,
)
return SparseTextEmbedding(model_name=SPARSE_MODEL_NAME)
def embed_sparse_texts(texts: list[str]) -> list[SparseVector]:
model = get_sparse_model()
vectors: list[dict[str, list[int] | list[float]]] = []
for item in model.embed(texts):
vectors.append(
{
"indices": item.indices.tolist(),
"values": item.values.tolist(),
}
)
return vectors
@app.post("/sparse_embedding")
async def sparse_embedding(payload: SparseEmbeddingRequest) -> dict[str, Any]:
# Проверяющая система вызывает этот endpoint при создании коллекции
vectors = await asyncio.to_thread(embed_sparse_texts, payload.texts)
return {"vectors": vectors}
return {"vectors": [{"indices": v.indices, "values": v.values} for v in vectors]}
# красивая обработка ошибок
@app.exception_handler(Exception)
async def exception_handler(request: Request, exc: Exception) -> JSONResponse:
logger.exception(exc)
if isinstance(exc, RequestValidationError):
return JSONResponse(status_code=422, content={"detail": exc.errors()})
return JSONResponse(status_code=500, content={"detail": str(exc)})
def main() -> None:
import uvicorn
uvicorn.run(
"main:app",
host=HOST,
port=PORT,
reload=False,
workers=UVICORN_WORKERS,
)
uvicorn.run("main:app", host=HOST, port=PORT, reload=False, workers=UVICORN_WORKERS)
if __name__ == "__main__":

102
index/rendering.py Normal file
View file

@ -0,0 +1,102 @@
"""Render cleaned messages into page_content, dense_content, sparse_content."""
import datetime
from cleaning import CleanedMessage
def _format_time(ts: int) -> str:
try:
return datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).strftime("%Y-%m-%d %H:%M")
except (OSError, OverflowError, ValueError):
return str(ts)
def render_page_content(msg: CleanedMessage) -> str:
"""Human-readable text for the chunk payload."""
lines: list[str] = []
prefix = f"{msg.sender_id}:"
if msg.text:
lines.append(f"{prefix} {msg.text}")
elif not msg.parts and not msg.member_event_text:
lines.append(prefix)
for part in msg.parts:
lines.append(part["text"])
if msg.member_event_text:
lines.append(msg.member_event_text)
for fi in msg.file_info:
name = fi.get("name") or fi.get("url") or "file"
lines.append(f"[attachment: {name}]")
return "\n".join(lines)
def render_dense_content(msg: CleanedMessage) -> str:
"""Text optimized for semantic dense embedding: role markers + full context."""
lines: list[str] = []
ts = _format_time(msg.time)
header_parts = [f"[{ts}]", f"sender:{msg.sender_id}"]
if msg.is_forward:
header_parts.append("type:forward")
if msg.is_quote:
header_parts.append("type:quote")
if msg.is_system:
header_parts.append("type:system")
if msg.thread_sn:
header_parts.append(f"thread:{msg.thread_sn}")
lines.append(" ".join(header_parts))
if msg.text:
lines.append(msg.text)
for part in msg.parts:
lines.append(part["text"])
if msg.member_event_text:
lines.append(msg.member_event_text)
for fi in msg.file_info:
fi_parts = []
if fi.get("name"):
fi_parts.append(fi["name"])
if fi.get("mime"):
fi_parts.append(fi["mime"])
if fi.get("url"):
fi_parts.append(fi["url"])
if fi_parts:
lines.append(f"[file: {' '.join(fi_parts)}]")
if msg.mentions:
lines.append("mentions: " + ", ".join(msg.mentions))
return "\n".join(lines)
def render_sparse_content(msg: CleanedMessage) -> str:
"""Keyword-heavy text for sparse/BM25 embedding."""
tokens: list[str] = []
tokens.append(msg.sender_id)
tokens.extend(msg.mentions)
if msg.text:
tokens.append(msg.text)
for part in msg.parts:
tokens.append(part["text"])
if msg.member_event_text:
tokens.append(msg.member_event_text)
for fi in msg.file_info:
for key in ("name", "mime", "url"):
val = fi.get(key)
if val:
tokens.append(val)
return " ".join(tokens)

39
index/sparse.py Normal file
View file

@ -0,0 +1,39 @@
import logging
import os
from functools import lru_cache
from index_schemas import SparseVector
SPARSE_MODEL_NAME = "Qdrant/bm25"
FASTEMBED_CACHE_PATH = "/models/fastembed"
MAX_SPARSE_TEXT_CHARS = int(os.getenv("MAX_SPARSE_TEXT_CHARS", "512"))
logger = logging.getLogger("index-service")
@lru_cache(maxsize=1)
def get_sparse_model():
from fastembed import SparseTextEmbedding
logger.info("Loading sparse model %s from cache %s", SPARSE_MODEL_NAME, FASTEMBED_CACHE_PATH)
return SparseTextEmbedding(model_name=SPARSE_MODEL_NAME)
def _prepare_text(text: str) -> str:
if not text:
return ""
return text[:MAX_SPARSE_TEXT_CHARS]
def embed_sparse_texts(texts: list[str]) -> list[SparseVector]:
model = get_sparse_model()
prepared = [_prepare_text(t) for t in texts]
result: list[SparseVector] = []
for item in model.embed(prepared):
result.append(
SparseVector(
indices=[int(i) for i in item.indices.tolist()],
values=[float(v) for v in item.values.tolist()],
)
)
return result

90
index/tcp_log_handler.py Normal file
View file

@ -0,0 +1,90 @@
"""
Non-blocking TCP log handler.
Sends JSON-lines to a remote server in a daemon background thread.
Never blocks the main application drops records when queue is full.
"""
import json
import logging
import queue
import socket
import threading
import time
from datetime import datetime, timezone
class TCPLogHandler(logging.Handler):
def __init__(self, host: str, port: int, service: str, timeout: float = 3.0):
super().__init__()
self.host = host
self.port = port
self.service = service
self.timeout = timeout
self._queue: queue.Queue[str] = queue.Queue(maxsize=2000)
self._sock: socket.socket | None = None
self._lock = threading.Lock()
self._thread = threading.Thread(target=self._worker, daemon=True, name="tcp-log")
self._thread.start()
def emit(self, record: logging.LogRecord) -> None:
try:
entry = {
"ts": datetime.now(tz=timezone.utc).isoformat(),
"level": record.levelname,
"service": self.service,
"logger": record.name,
"msg": self.format(record),
}
self._queue.put_nowait(json.dumps(entry, ensure_ascii=False) + "\n")
except queue.Full:
pass # drop — never block the caller
def _connect(self) -> bool:
try:
sock = socket.create_connection((self.host, self.port), timeout=self.timeout)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
with self._lock:
self._sock = sock
return True
except OSError:
return False
def _close_sock(self) -> None:
with self._lock:
if self._sock:
try:
self._sock.close()
except OSError:
pass
self._sock = None
def _worker(self) -> None:
while True:
line = self._queue.get()
sent = False
while not sent:
with self._lock:
sock = self._sock
if sock is None:
if not self._connect():
time.sleep(5)
continue
with self._lock:
sock = self._sock
try:
sock.sendall(line.encode("utf-8")) # type: ignore[union-attr]
sent = True
except OSError:
self._close_sock()
time.sleep(2)
def setup_tcp_logging(service: str, host: str, port: int) -> TCPLogHandler | None:
"""Attach TCP handler to root logger. Returns handler or None if disabled."""
if not host or not port:
return None
handler = TCPLogHandler(host=host, port=port, service=service)
handler.setFormatter(logging.Formatter("%(message)s"))
logging.getLogger().addHandler(handler)
logging.getLogger().info("TCP log handler started → %s:%d", host, port)
return handler

3
kredit.md Normal file
View file

@ -0,0 +1,3 @@
team_id: 35230
vk login: 56aa86799bb9edc4
vk password: edd89cea9ed0734d00ba6904cf7475d7

107
logserver/server.py Normal file
View file

@ -0,0 +1,107 @@
#!/usr/bin/env python3
"""
TCP log server receives JSON-line logs from index-service and search-service.
Usage:
python3 server.py # listen on 0.0.0.0:9999
python3 server.py --port 9999
python3 server.py --save logs.jsonl # also save to file
"""
import argparse
import json
import logging
import socketserver
import sys
import threading
from datetime import datetime
COLORS = {
"DEBUG": "\033[36m",
"INFO": "\033[0m",
"WARNING": "\033[33m",
"ERROR": "\033[31m",
"CRITICAL": "\033[35m",
}
RESET = "\033[0m"
SERVICE_COLOR = {
"index-service": "\033[34m", # blue
"search-service": "\033[32m", # green
}
_save_file = None
_save_lock = threading.Lock()
def _format(entry: dict) -> str:
ts = entry.get("ts", "")[:23].replace("T", " ")
level = entry.get("level", "INFO")
service = entry.get("service", "?")
msg = entry.get("msg", "")
lc = COLORS.get(level, "")
sc = SERVICE_COLOR.get(service, "\033[0m")
return f"{ts} {sc}{service:<15}{RESET} {lc}{level:<8}{RESET} {msg}"
def _handle_line(raw: str) -> None:
raw = raw.strip()
if not raw:
return
try:
entry = json.loads(raw)
except json.JSONDecodeError:
entry = {"ts": datetime.utcnow().isoformat(), "level": "INFO", "service": "?", "msg": raw}
print(_format(entry), flush=True)
if _save_file:
with _save_lock:
_save_file.write(raw + "\n")
_save_file.flush()
class _Handler(socketserver.StreamRequestHandler):
def handle(self) -> None:
addr = self.client_address[0]
print(f"\033[90m[+] connected: {addr}{RESET}", flush=True)
try:
for raw_bytes in self.rfile:
try:
_handle_line(raw_bytes.decode("utf-8", errors="replace"))
except Exception:
pass
except Exception:
pass
print(f"\033[90m[-] disconnected: {addr}{RESET}", flush=True)
def main() -> None:
global _save_file
parser = argparse.ArgumentParser(description="TCP JSON-line log receiver")
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=9999)
parser.add_argument("--save", metavar="FILE", help="Also save raw JSON lines to this file")
args = parser.parse_args()
if args.save:
_save_file = open(args.save, "a", encoding="utf-8")
print(f"Saving logs to {args.save}", flush=True)
server = socketserver.ThreadingTCPServer((args.host, args.port), _Handler)
server.allow_reuse_address = True
print(f"Listening on {args.host}:{args.port} ...\n", flush=True)
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopped.")
finally:
server.server_close()
if _save_file:
_save_file.close()
if __name__ == "__main__":
main()

41
run_codex.sh Executable file
View file

@ -0,0 +1,41 @@
---
## 3) `bin/codex-start` чтобы всё поднималось само
`SKILL.md` сам по себе **не умеет магически запускать Codex при открытии терминала**. Для этого нужен обычный wrapper-скрипт. Вот рабочий вариант:
```bash
#!/usr/bin/env bash
set -euo pipefail
ROOT="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
cd "$ROOT"
mkdir -p .ai_update/sessions
touch .ai_update/current_status.md
touch .ai_update/handoff.md
touch .ai_update/changelog.md
touch .ai_update/touched_files.md
COMPOSE_FILE=""
if [ -f docker-compose.yml ]; then
COMPOSE_FILE="docker-compose.yml"
elif [ -f compose.yaml ]; then
COMPOSE_FILE="compose.yaml"
elif [ -f compose.yml ]; then
COMPOSE_FILE="compose.yml"
fi
if [ -n "$COMPOSE_FILE" ] && command -v docker >/dev/null 2>&1; then
if ! docker compose ps --status running >/dev/null 2>&1; then
docker compose up -d --build || true
else
RUNNING_COUNT="$(docker compose ps --status running --services 2>/dev/null | wc -l | tr -d ' ')"
if [ "${RUNNING_COUNT:-0}" = "0" ]; then
docker compose up -d --build || true
fi
fi
fi
exec codex "$@"

View file

@ -5,7 +5,7 @@ WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
COPY *.py .
ENV HOST=0.0.0.0
ENV PORT=8000

View file

@ -25,7 +25,7 @@ login:
build:
@: $(if $(TEAM_ID),,$(error TEAM_ID is required for make build))
docker build -t $(IMAGE) ./
docker build --platform linux/amd64 -t $(IMAGE) ./
run: build
@: $(foreach var,$(REQUIRED_RUN_VARS),$(if $($(var)),,$(error $(var) is required for make run)))

0
search/__init__.py Normal file
View file

23
search/aggregation.py Normal file
View file

@ -0,0 +1,23 @@
from typing import Any
from config import TOP_K
from retrieval import extract_message_ids
def aggregate_message_ids(
reranked_head: list[Any],
retrieval_tail: list[Any],
) -> list[str]:
"""Deduplicate and collect top-K message_ids, reranked head first."""
seen: set[str] = set()
result: list[str] = []
for point in reranked_head + retrieval_tail:
for mid in extract_message_ids(point):
if mid not in seen:
seen.add(mid)
result.append(mid)
if len(result) >= TOP_K:
return result
return result

56
search/config.py Normal file
View file

@ -0,0 +1,56 @@
import logging
import os
from typing import Any
EMBEDDINGS_DENSE_MODEL = "Qwen/Qwen3-Embedding-0.6B"
SPARSE_MODEL_NAME = "Qdrant/bm25"
RERANKER_MODEL = "nvidia/llama-nemotron-rerank-1b-v2"
HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", "8003"))
API_KEY = os.getenv("API_KEY")
EMBEDDINGS_DENSE_URL = os.getenv("EMBEDDINGS_DENSE_URL")
RERANKER_URL = os.getenv("RERANKER_URL")
QDRANT_URL = os.getenv("QDRANT_URL")
QDRANT_COLLECTION_NAME = os.getenv("QDRANT_COLLECTION_NAME", "evaluation")
QDRANT_DENSE_VECTOR_NAME = os.getenv("QDRANT_DENSE_VECTOR_NAME", "dense")
QDRANT_SPARSE_VECTOR_NAME = os.getenv("QDRANT_SPARSE_VECTOR_NAME", "sparse")
OPEN_API_LOGIN = os.getenv("OPEN_API_LOGIN")
OPEN_API_PASSWORD = os.getenv("OPEN_API_PASSWORD")
DENSE_PREFETCH_K = 50
SPARSE_PREFETCH_K = 100
RETRIEVE_K = 80
RERANK_LIMIT = 60
TOP_K = 50
HTTP_TIMEOUT = 30.0
HTTP_MAX_RETRIES = 2
REQUIRED_ENV_VARS = ["EMBEDDINGS_DENSE_URL", "RERANKER_URL", "QDRANT_URL"]
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
logger = logging.getLogger("search-service")
def validate_required_env() -> None:
if bool(OPEN_API_LOGIN) != bool(OPEN_API_PASSWORD):
raise RuntimeError("OPEN_API_LOGIN and OPEN_API_PASSWORD must be set together")
if not API_KEY and not (OPEN_API_LOGIN and OPEN_API_PASSWORD):
raise RuntimeError("Either API_KEY or OPEN_API_LOGIN and OPEN_API_PASSWORD must be set")
missing = [name for name in REQUIRED_ENV_VARS if not os.getenv(name)]
if missing:
logger.error("Empty required env vars: %s", ", ".join(missing))
raise RuntimeError(f"Empty required env vars: {', '.join(missing)}")
def get_upstream_kwargs() -> dict[str, Any]:
headers = {"Content-Type": "application/json"}
kwargs: dict[str, Any] = {"headers": headers}
if OPEN_API_LOGIN and OPEN_API_PASSWORD:
kwargs["auth"] = (OPEN_API_LOGIN, OPEN_API_PASSWORD)
return kwargs
if API_KEY:
headers["Authorization"] = f"Bearer {API_KEY}"
return kwargs

View file

@ -1,163 +1,62 @@
import asyncio
import logging
import os
from contextlib import asynccontextmanager
from functools import lru_cache
from typing import Any
import httpx
from fastembed import SparseTextEmbedding
from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from qdrant_client import AsyncQdrantClient, models
from qdrant_client import AsyncQdrantClient
EMBEDDINGS_DENSE_MODEL = "Qwen/Qwen3-Embedding-0.6B"
from aggregation import aggregate_message_ids
from config import (
API_KEY,
HOST,
HTTP_MAX_RETRIES,
HTTP_TIMEOUT,
PORT,
QDRANT_URL,
logger,
validate_required_env,
)
from tcp_log_handler import setup_tcp_logging
# Ваш сервис должен считывать эти переменные из окружения (env), так как проверяющая система управляет ими
HOST = os.getenv("HOST", "0.0.0.0")
PORT = int(os.getenv("PORT", "8003"))
API_KEY = os.getenv("API_KEY")
EMBEDDINGS_DENSE_URL = os.getenv("EMBEDDINGS_DENSE_URL")
QDRANT_DENSE_VECTOR_NAME = os.getenv("QDRANT_DENSE_VECTOR_NAME", "dense")
QDRANT_SPARSE_VECTOR_NAME = os.getenv("QDRANT_SPARSE_VECTOR_NAME", "sparse")
SPARSE_MODEL_NAME = "Qdrant/bm25"
RERANKER_MODEL = "nvidia/llama-nemotron-rerank-1b-v2"
RERANKER_URL = os.getenv("RERANKER_URL")
OPEN_API_LOGIN = os.getenv("OPEN_API_LOGIN")
OPEN_API_PASSWORD = os.getenv("OPEN_API_PASSWORD")
QDRANT_URL = os.getenv("QDRANT_URL")
QDRANT_COLLECTION_NAME = os.getenv("QDRANT_COLLECTION_NAME", "evaluation")
REQUIRED_ENV_VARS = [
"EMBEDDINGS_DENSE_URL",
"RERANKER_URL",
"QDRANT_URL",
]
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
logger = logging.getLogger("search-service")
_LOG_TCP_HOST = os.getenv("LOG_TCP_HOST", "185.33.228.73")
_LOG_TCP_PORT = int(os.getenv("LOG_TCP_PORT", "9999"))
setup_tcp_logging("search-service", _LOG_TCP_HOST, _LOG_TCP_PORT)
from query_builder import (
build_extra_dense_queries,
build_primary_query,
build_sparse_query,
embed_dense,
embed_dense_multi,
embed_sparse,
)
from rerank import rerank_points
from retrieval import qdrant_search
from schemas import SearchAPIItem, SearchAPIRequest, SearchAPIResponse, SparseVector
def validate_required_env() -> None:
if bool(OPEN_API_LOGIN) != bool(OPEN_API_PASSWORD):
raise RuntimeError("OPEN_API_LOGIN and OPEN_API_PASSWORD must be set together")
if not API_KEY and not (OPEN_API_LOGIN and OPEN_API_PASSWORD):
raise RuntimeError("Either API_KEY or OPEN_API_LOGIN and OPEN_API_PASSWORD must be set")
missing_env_vars = [
name for name in REQUIRED_ENV_VARS if os.getenv(name) is None or os.getenv(name) == ""
]
if not missing_env_vars:
return
logger.error("Empty required env vars: %s", ", ".join(missing_env_vars))
raise RuntimeError(f"Empty required env vars: {', '.join(missing_env_vars)}")
validate_required_env()
def get_upstream_request_kwargs() -> dict[str, Any]:
headers = {"Content-Type": "application/json"}
kwargs: dict[str, Any] = {"headers": headers}
if OPEN_API_LOGIN and OPEN_API_PASSWORD:
kwargs["auth"] = (OPEN_API_LOGIN, OPEN_API_PASSWORD)
return kwargs
if API_KEY:
headers["Authorization"] = f"Bearer {API_KEY}"
return kwargs
# Модель данных, которую мы предоставляем и рассчитываем получать от вас
class DateRange(BaseModel):
from_: str = Field(alias="from")
to: str
class Entities(BaseModel):
people: list[str] | None = None
emails: list[str] | None = None
documents: list[str] | None = None
names: list[str] | None = None
links: list[str] | None = None
class Question(BaseModel):
text: str
asker: str = ""
asked_on: str = ""
variants: list[str] | None = None
hyde: list[str] | None = None
keywords: list[str] | None = None
entities: Entities | None = None
date_mentions: list[str] | None = None
date_range: DateRange | None = None
search_text: str = ""
class SearchAPIRequest(BaseModel):
question: Question
class SearchAPIItem(BaseModel):
message_ids: list[str]
class SearchAPIResponse(BaseModel):
results: list[SearchAPIItem]
class DenseEmbeddingItem(BaseModel):
index: int
embedding: list[float]
class DenseEmbeddingResponse(BaseModel):
data: list[DenseEmbeddingItem]
class SparseVector(BaseModel):
indices: list[int] = Field(default_factory=list)
values: list[float] = Field(default_factory=list)
class SparseEmbeddingResponse(BaseModel):
vectors: list[SparseVector]
# Метадата чанков в Qdrant'e, по которой вы можете фильтровать
class ChunkMetadata(BaseModel):
chat_name: str
chat_type: str # channel, group, private, thread
chat_id: str
chat_sn: str
thread_sn: str | None = None
message_ids: list[str]
start: str
end: str
participants: list[str] = Field(default_factory=list)
mentions: list[str] = Field(default_factory=list)
contains_forward: bool = False
contains_quote: bool = False
@lru_cache(maxsize=1)
def get_sparse_model() -> SparseTextEmbedding:
logger.info("Loading local sparse model %s", SPARSE_MODEL_NAME)
return SparseTextEmbedding(model_name=SPARSE_MODEL_NAME)
async def _embed_dense_with_retry(client: httpx.AsyncClient, text: str) -> list[float]:
last_exc: Exception | None = None
for attempt in range(HTTP_MAX_RETRIES + 1):
try:
return await embed_dense(client, text)
except (httpx.TransportError, httpx.HTTPStatusError) as exc:
if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 500:
raise
last_exc = exc
if attempt < HTTP_MAX_RETRIES:
await asyncio.sleep(0.5 * (attempt + 1))
raise RuntimeError(f"Dense embedding failed after retries: {last_exc}")
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.http = httpx.AsyncClient()
app.state.qdrant = AsyncQdrantClient(
url=QDRANT_URL,
api_key=API_KEY,
)
validate_required_env()
app.state.http = httpx.AsyncClient(timeout=HTTP_TIMEOUT)
app.state.qdrant = AsyncQdrantClient(url=QDRANT_URL, api_key=API_KEY)
try:
yield
finally:
@ -165,328 +64,9 @@ async def lifespan(app: FastAPI):
await app.state.qdrant.close()
app = FastAPI(title="Search Service", version="0.1.0", lifespan=lifespan)
app = FastAPI(title="Search Service", version="0.2.0", lifespan=lifespan)
# Внутри шаблона dense и rerank берутся из внешних HTTP endpoint'ов,
# которые предоставляет проверяющая система.
# Текущий код ниже — минимальный пример search pipeline.
DENSE_PREFETCH_K = 10
SPRASE_PREFETCH_K = 30
RETRIEVE_K = 20
RERANK_LIMIT = 10
FINAL_TOP_K = 50
async def embed_dense(client: httpx.AsyncClient, text: str) -> list[float]:
# Dense endpoint ожидает OpenAI-compatible body с input как списком строк.
response = await client.post(
EMBEDDINGS_DENSE_URL,
**get_upstream_request_kwargs(),
json={
"model": os.getenv("EMBEDDINGS_DENSE_MODEL", EMBEDDINGS_DENSE_MODEL),
"input": [text],
},
)
response.raise_for_status()
payload = DenseEmbeddingResponse.model_validate(response.json())
if not payload.data:
raise ValueError("Dense embedding response is empty")
return payload.data[0].embedding
async def embed_sparse(text: str) -> SparseVector:
vectors = list(get_sparse_model().embed([text]))
if not vectors:
raise ValueError("Sparse embedding response is empty")
item = vectors[0]
return SparseVector(
indices=[int(index) for index in item.indices.tolist()],
values=[float(value) for value in item.values.tolist()],
)
# ПЕРЕПИСАТЬ
async def qdrant_search(
client: AsyncQdrantClient,
dense_vector: list[float],
sparse_vector: SparseVector,
question_data: Question
) -> Any | None:
must_conditions: []
# Фильтр по диапазону дат (поле metadata.start в Qdrant) [cite: 147, 148, 175]
if question_data.date_range:
must_conditions.append(
models.FieldCondition(
key="metadata.start",
range=models.Range(
gte=question_data.date_range.from_,
lte=question_data.date_range.to_
)
)
)
# Фильтр по автору вопроса (поле metadata.participants) [cite: 161, 163]
if question_data.asker:
must_conditions.append(
models.FieldCondition(
key="metadata.participants",
match=models.MatchValue(value=question_data.asker)
)
)
# Создаем итоговый объект фильтра, если есть условия
search_filter = models.Filter(must=must_conditions) if must_conditions else None
response = await client.query_points(
collection_name=QDRANT_COLLECTION_NAME,
prefetch=[
models.Prefetch(
query=dense_vector,
using=QDRANT_DENSE_VECTOR_NAME,
limit=DENSE_PREFETCH_K,
filter=search_filter,
),
models.Prefetch(
query=models.SparseVector(
indices=sparse_vector.indices,
values=sparse_vector.values,
),
using=QDRANT_SPARSE_VECTOR_NAME,
limit=SPRASE_PREFETCH_K,
filter=search_filter,
),
],
query=models.FusionQuery(fusion=models.Fusion.RRF),
limit=RETRIEVE_K,
with_payload=True,
)
if not response.points:
return None
return response.points
async def qdrant_search_dense_only(
client: AsyncQdrantClient,
dense_vector: list[float],
question_data: Question,
) -> Any | None:
must_conditions: []
if question_data.date_range:
must_conditions.append(
models.FieldCondition(
key="metadata.start",
range=models.Range(
gte=question_data.date_range.from_,
lte=question_data.date_range.to_,
),
)
)
if question_data.asker:
must_conditions.append(
models.FieldCondition(
key="metadata.participants",
match=models.MatchValue(value=question_data.asker),
)
)
search_filter = models.Filter(must=must_conditions) if must_conditions else None
response = await client.query_points(
collection_name=QDRANT_COLLECTION_NAME,
prefetch=[
models.Prefetch(
query=dense_vector,
using=QDRANT_DENSE_VECTOR_NAME,
limit=DENSE_PREFETCH_K,
filter=search_filter,
),
],
query=models.FusionQuery(fusion=models.Fusion.RRF),
limit=RETRIEVE_K,
with_payload=True,
)
if not response.points:
return None
return response.points
def collect_query_variants(question: Question) -> list[str]:
variants: list[str] = []
seen: set[str] = set()
def add_query(text: str | None) -> None:
if text is None:
return
normalized = text.strip()
if not normalized:
return
if normalized in seen:
return
seen.add(normalized)
variants.append(normalized)
add_query(question.search_text)
add_query(question.text)
for variant in question.variants or []:
add_query(variant)
return variants
def collect_hyde_queries(question: Question, base_queries: list[str]) -> list[str]:
hyde_queries: list[str] = []
seen: set[str] = set(base_queries)
for hyde_query in question.hyde or []:
normalized = hyde_query.strip()
if not normalized:
continue
if normalized in seen:
continue
seen.add(normalized)
hyde_queries.append(normalized)
return hyde_queries
def build_sparse_query_text(question: Question, fallback_query: str) -> str:
keywords: list[str] = []
seen: set[str] = set()
for keyword in question.keywords or []:
normalized = keyword.strip()
if not normalized:
continue
if normalized in seen:
continue
seen.add(normalized)
keywords.append(normalized)
if keywords:
return " ".join(keywords)
return fallback_query
def deduplicate_points(points: list[Any]) -> list[Any]:
unique_points: list[Any] = []
seen_ids: set[str] = set()
for point in points:
point_id = str(getattr(point, "id", ""))
if not point_id:
continue
if point_id in seen_ids:
continue
seen_ids.add(point_id)
unique_points.append(point)
return unique_points
def extract_point_score(point: Any) -> float:
score = getattr(point, "score", 0.0)
if score is None:
return 0.0
return float(score)
def extract_message_ids(point: Any) -> list[str]:
payload = point.payload or {}
metadata = payload.get("metadata") or {}
message_ids = metadata.get("message_ids") or []
return [str(message_id) for message_id in message_ids]
async def get_rerank_scores(
client: httpx.AsyncClient,
label: str,
targets: list[str],
) -> list[float]:
if not targets:
return []
# Rerank endpoint возвращает score для пары query -> candidate text.
response = await client.post(
RERANKER_URL,
**get_upstream_request_kwargs(),
json={
"model": RERANKER_MODEL,
"encoding_format": "float",
"text_1": label,
"text_2": targets,
},
)
response.raise_for_status()
payload = response.json()
data = payload.get("data") or []
return [float(sample["score"]) for sample in data]
async def rerank_points(
client: httpx.AsyncClient,
query: str,
points: list[Any],
) -> list[tuple[Any, float]]:
rerank_candidates = points[:RERANK_LIMIT]
tail_candidates = points[RERANK_LIMIT:]
rerank_targets = [point.payload.get("page_content") for point in rerank_candidates]
scores = await get_rerank_scores(client, query, rerank_targets)
reranked_candidates = [
(point, float(score))
for score, point in sorted(
zip(scores, rerank_candidates, strict=True),
key=lambda item: item[0],
reverse=True,
)
]
tail_with_scores = [
(point, extract_point_score(point))
for _, point in sorted(
[(extract_point_score(point), point) for point in tail_candidates],
key=lambda item: item[0],
reverse=True,
)
]
return reranked_candidates + tail_with_scores
def aggregate_message_scores(scored_points: list[tuple[Any, float]]) -> dict[str, float]:
aggregated_scores: dict[str, float] = {}
for point, point_score in scored_points:
point_message_ids = set(extract_message_ids(point))
for message_id in point_message_ids:
aggregated_scores[message_id] = aggregated_scores.get(message_id, 0.0) + point_score
return aggregated_scores
def select_top_message_ids(aggregated_scores: dict[str, float], limit: int) -> list[str]:
sorted_items = sorted(
aggregated_scores.items(),
key=lambda item: (-item[1], item[0]),
)
return [message_id for message_id, _ in sorted_items[:limit]]
# Ваш сервис должен имплементировать оба этих метода
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
@ -494,41 +74,42 @@ async def health() -> dict[str, str]:
@app.post("/search", response_model=SearchAPIResponse)
async def search(payload: SearchAPIRequest) -> SearchAPIResponse:
queries = collect_query_variants(payload.question)
if not queries:
raise HTTPException(status_code=400, detail="question.search_text or question.text is required")
question = payload.question
primary_query = build_primary_query(question)
if not primary_query:
raise HTTPException(status_code=400, detail="question.text is required")
hyde_queries = collect_hyde_queries(payload.question, queries)
query = queries[0]
client: httpx.AsyncClient = app.state.http
qdrant: AsyncQdrantClient = app.state.qdrant
all_points: list[Any] = []
for query_variant in queries:
dense_vector = await embed_dense(client, query_variant)
sparse_query_text = build_sparse_query_text(payload.question, query_variant)
sparse_vector = await embed_sparse(sparse_query_text)
points = await qdrant_search(qdrant, dense_vector, sparse_vector, payload.question)
if points:
all_points.extend(list(points))
extra_texts = build_extra_dense_queries(question)
sparse_text = build_sparse_query(question)
for hyde_query in hyde_queries:
hyde_dense_vector = await embed_dense(client, hyde_query)
hyde_points = await qdrant_search_dense_only(qdrant, hyde_dense_vector, payload.question)
if hyde_points:
all_points.extend(list(hyde_points))
async def _no_extra() -> list:
return []
best_points = deduplicate_points(all_points)
if not best_points:
extra_task = embed_dense_multi(client, extra_texts) if extra_texts else _no_extra()
primary_dense, extra_dense_vecs, sparse_vec = await asyncio.gather(
_embed_dense_with_retry(client, primary_query),
extra_task,
asyncio.to_thread(embed_sparse, sparse_text),
)
points = await qdrant_search(
qdrant,
primary_dense,
extra_dense_vecs,
sparse_vec,
question,
)
if not points:
return SearchAPIResponse(results=[])
scored_points = await rerank_points(client, query, list(best_points))
aggregated_scores = aggregate_message_scores(scored_points)
message_ids = select_top_message_ids(aggregated_scores, FINAL_TOP_K)
reranked_head, retrieval_tail = await rerank_points(client, primary_query, points)
message_ids = aggregate_message_ids(reranked_head, retrieval_tail)
return SearchAPIResponse(
results=[SearchAPIItem(message_ids=message_ids)]
)
return SearchAPIResponse(results=[SearchAPIItem(message_ids=message_ids)])
@app.exception_handler(Exception)
@ -548,12 +129,7 @@ async def exception_handler(request: Request, exc: Exception) -> JSONResponse:
def main() -> None:
import uvicorn
uvicorn.run(
"main:app",
host=HOST,
port=PORT,
reload=False,
)
uvicorn.run("main:app", host=HOST, port=PORT, reload=False)
if __name__ == "__main__":

99
search/query_builder.py Normal file
View file

@ -0,0 +1,99 @@
import asyncio
import os
import re
from functools import lru_cache
import httpx
from fastembed import SparseTextEmbedding
from config import (
EMBEDDINGS_DENSE_MODEL,
EMBEDDINGS_DENSE_URL,
SPARSE_MODEL_NAME,
get_upstream_kwargs,
logger,
)
from schemas import DenseEmbeddingResponse, Question, SparseVector
@lru_cache(maxsize=1)
def get_sparse_model() -> SparseTextEmbedding:
logger.info("Loading local sparse model %s", SPARSE_MODEL_NAME)
return SparseTextEmbedding(model_name=SPARSE_MODEL_NAME)
async def embed_dense(client: httpx.AsyncClient, text: str) -> list[float]:
response = await client.post(
str(EMBEDDINGS_DENSE_URL),
**get_upstream_kwargs(),
json={
"model": os.getenv("EMBEDDINGS_DENSE_MODEL", EMBEDDINGS_DENSE_MODEL),
"input": [text],
},
)
response.raise_for_status()
payload = DenseEmbeddingResponse.model_validate(response.json())
if not payload.data:
raise ValueError("Dense embedding response is empty")
return payload.data[0].embedding
async def embed_dense_multi(client: httpx.AsyncClient, texts: list[str]) -> list[list[float]]:
tasks = [embed_dense(client, t) for t in texts]
return list(await asyncio.gather(*tasks))
def embed_sparse(text: str) -> SparseVector:
vectors = list(get_sparse_model().embed([text]))
if not vectors:
raise ValueError("Sparse embedding response is empty")
item = vectors[0]
return SparseVector(
indices=[int(i) for i in item.indices.tolist()],
values=[float(v) for v in item.values.tolist()],
)
def _normalize_query(text: str) -> str:
return re.sub(r"\s+", " ", text).strip()
def build_primary_query(question: Question) -> str:
q = question.search_text.strip() if question.search_text else ""
if not q:
q = question.text.strip()
return _normalize_query(q)
def build_extra_dense_queries(question: Question) -> list[str]:
extras: list[str] = []
for v in question.variants or []:
q = _normalize_query(v)
if q:
extras.append(q)
for h in question.hyde or []:
q = _normalize_query(h)
if q:
extras.append(q)
return extras
def build_sparse_query(question: Question) -> str:
kws = question.keywords or []
if kws:
return " ".join(kws)
return build_primary_query(question)
def build_entity_tokens(question: Question) -> list[str]:
tokens: list[str] = []
if question.entities:
for field in (
question.entities.people,
question.entities.emails,
question.entities.documents,
question.entities.names,
question.entities.links,
):
tokens.extend(field or [])
return [t.strip() for t in tokens if t.strip()]

58
search/rerank.py Normal file
View file

@ -0,0 +1,58 @@
from typing import Any
import httpx
from config import RERANK_LIMIT, RERANKER_MODEL, RERANKER_URL, get_upstream_kwargs, logger
from retrieval import extract_page_content
async def get_rerank_scores(
client: httpx.AsyncClient,
query: str,
targets: list[str],
) -> list[float]:
if not targets:
return []
response = await client.post(
str(RERANKER_URL),
**get_upstream_kwargs(),
json={
"model": RERANKER_MODEL,
"encoding_format": "float",
"text_1": query,
"text_2": targets,
},
)
response.raise_for_status()
data = response.json().get("data") or []
return [float(sample["score"]) for sample in data]
async def rerank_points(
client: httpx.AsyncClient,
query: str,
points: list[Any],
) -> tuple[list[Any], list[Any]]:
"""Return (reranked_head, retrieval_tail) so we don't lose candidates."""
if not points:
return [], []
rerank_candidates = points[:RERANK_LIMIT]
tail = points[RERANK_LIMIT:]
targets = [extract_page_content(p) for p in rerank_candidates]
try:
scores = await get_rerank_scores(client, query, targets)
except Exception as exc:
logger.warning("Rerank failed, using retrieval order: %s", exc)
return rerank_candidates, tail
if len(scores) != len(rerank_candidates):
logger.warning("Rerank score count mismatch, using retrieval order")
return rerank_candidates, tail
paired = sorted(zip(scores, rerank_candidates), key=lambda x: x[0], reverse=True)
reranked = [p for _, p in paired]
return reranked, tail

121
search/retrieval.py Normal file
View file

@ -0,0 +1,121 @@
from datetime import datetime
from typing import Any
from qdrant_client import AsyncQdrantClient, models
from config import (
DENSE_PREFETCH_K,
QDRANT_COLLECTION_NAME,
QDRANT_DENSE_VECTOR_NAME,
QDRANT_SPARSE_VECTOR_NAME,
RETRIEVE_K,
SPARSE_PREFETCH_K,
logger,
)
from schemas import Question, SparseVector
def _iso_to_unix(s: str) -> float:
try:
return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()
except (ValueError, AttributeError):
return 0.0
def _build_filter(question: Question) -> models.Filter | None:
must_conditions: list[models.Condition] = []
if question.date_range:
must_conditions.append(
models.FieldCondition(
key="metadata.start",
range=models.Range(
gte=_iso_to_unix(question.date_range.from_),
lte=_iso_to_unix(question.date_range.to),
),
)
)
if question.asker:
must_conditions.append(
models.FieldCondition(
key="metadata.participants",
match=models.MatchValue(value=question.asker),
)
)
return models.Filter(must=must_conditions) if must_conditions else None
async def qdrant_search(
client: AsyncQdrantClient,
primary_dense: list[float],
extra_dense: list[list[float]],
sparse_vector: SparseVector,
question: Question,
) -> list[Any]:
search_filter = _build_filter(question)
prefetch: list[models.Prefetch] = []
# Primary dense
prefetch.append(
models.Prefetch(
query=primary_dense,
using=QDRANT_DENSE_VECTOR_NAME,
limit=DENSE_PREFETCH_K,
filter=search_filter,
)
)
# Extra dense (variants / hyde) - smaller budget per query
extra_k = max(10, DENSE_PREFETCH_K // max(1, len(extra_dense)))
for vec in extra_dense:
prefetch.append(
models.Prefetch(
query=vec,
using=QDRANT_DENSE_VECTOR_NAME,
limit=extra_k,
filter=search_filter,
)
)
# Sparse
prefetch.append(
models.Prefetch(
query=models.SparseVector(
indices=sparse_vector.indices,
values=sparse_vector.values,
),
using=QDRANT_SPARSE_VECTOR_NAME,
limit=SPARSE_PREFETCH_K,
filter=search_filter,
)
)
response = await client.query_points(
collection_name=QDRANT_COLLECTION_NAME,
prefetch=prefetch,
query=models.FusionQuery(fusion=models.Fusion.RRF),
limit=RETRIEVE_K,
with_payload=True,
)
if not response.points:
logger.debug("Qdrant returned 0 points")
return []
logger.debug("Qdrant returned %d points", len(response.points))
return list(response.points)
def extract_message_ids(point: Any) -> list[str]:
payload = point.payload or {}
metadata = payload.get("metadata") or {}
message_ids = metadata.get("message_ids") or []
return [str(mid) for mid in message_ids]
def extract_page_content(point: Any) -> str:
payload = point.payload or {}
return payload.get("page_content") or ""

72
search/schemas.py Normal file
View file

@ -0,0 +1,72 @@
from pydantic import BaseModel, Field
class DateRange(BaseModel):
from_: str = Field(alias="from")
to: str
class Entities(BaseModel):
people: list[str] | None = None
emails: list[str] | None = None
documents: list[str] | None = None
names: list[str] | None = None
links: list[str] | None = None
class Question(BaseModel):
text: str
asker: str = ""
asked_on: str = ""
variants: list[str] | None = None
hyde: list[str] | None = None
keywords: list[str] | None = None
entities: Entities | None = None
date_mentions: list[str] | None = None
date_range: DateRange | None = None
search_text: str = ""
class SearchAPIRequest(BaseModel):
question: Question
class SearchAPIItem(BaseModel):
message_ids: list[str]
class SearchAPIResponse(BaseModel):
results: list[SearchAPIItem]
class DenseEmbeddingItem(BaseModel):
index: int
embedding: list[float]
class DenseEmbeddingResponse(BaseModel):
data: list[DenseEmbeddingItem]
class SparseVector(BaseModel):
indices: list[int] = Field(default_factory=list)
values: list[float] = Field(default_factory=list)
class SparseEmbeddingResponse(BaseModel):
vectors: list[SparseVector]
class ChunkMetadata(BaseModel):
chat_name: str
chat_type: str
chat_id: str
chat_sn: str
thread_sn: str | None = None
message_ids: list[str]
start: str
end: str
participants: list[str] = Field(default_factory=list)
mentions: list[str] = Field(default_factory=list)
contains_forward: bool = False
contains_quote: bool = False

90
search/tcp_log_handler.py Normal file
View file

@ -0,0 +1,90 @@
"""
Non-blocking TCP log handler.
Sends JSON-lines to a remote server in a daemon background thread.
Never blocks the main application drops records when queue is full.
"""
import json
import logging
import queue
import socket
import threading
import time
from datetime import datetime, timezone
class TCPLogHandler(logging.Handler):
def __init__(self, host: str, port: int, service: str, timeout: float = 3.0):
super().__init__()
self.host = host
self.port = port
self.service = service
self.timeout = timeout
self._queue: queue.Queue[str] = queue.Queue(maxsize=2000)
self._sock: socket.socket | None = None
self._lock = threading.Lock()
self._thread = threading.Thread(target=self._worker, daemon=True, name="tcp-log")
self._thread.start()
def emit(self, record: logging.LogRecord) -> None:
try:
entry = {
"ts": datetime.now(tz=timezone.utc).isoformat(),
"level": record.levelname,
"service": self.service,
"logger": record.name,
"msg": self.format(record),
}
self._queue.put_nowait(json.dumps(entry, ensure_ascii=False) + "\n")
except queue.Full:
pass # drop — never block the caller
def _connect(self) -> bool:
try:
sock = socket.create_connection((self.host, self.port), timeout=self.timeout)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
with self._lock:
self._sock = sock
return True
except OSError:
return False
def _close_sock(self) -> None:
with self._lock:
if self._sock:
try:
self._sock.close()
except OSError:
pass
self._sock = None
def _worker(self) -> None:
while True:
line = self._queue.get()
sent = False
while not sent:
with self._lock:
sock = self._sock
if sock is None:
if not self._connect():
time.sleep(5)
continue
with self._lock:
sock = self._sock
try:
sock.sendall(line.encode("utf-8")) # type: ignore[union-attr]
sent = True
except OSError:
self._close_sock()
time.sleep(2)
def setup_tcp_logging(service: str, host: str, port: int) -> TCPLogHandler | None:
"""Attach TCP handler to root logger. Returns handler or None if disabled."""
if not host or not port:
return None
handler = TCPLogHandler(host=host, port=port, service=service)
handler.setFormatter(logging.Formatter("%(message)s"))
logging.getLogger().addHandler(handler)
logging.getLogger().info("TCP log handler started → %s:%d", host, port)
return handler

59
skill.md Normal file
View file

@ -0,0 +1,59 @@
# Skill: Automating Task Tracking and Git Push with Codex
## Overview
This skill involves using **OpenAI Codex** to automate the tracking of task changes, logging updates, and automatically pushing changes to a Git repository. It combines **Codex's ability to generate code** with the power of **Git automation** to keep track of development tasks, log updates, and push them to version control.
## Objectives
1. Track the status of development tasks (e.g., status, assignee, comments).
2. Log all changes to tasks in a local directory (`.ai_update/`).
3. Automatically commit and push changes to a Git repository (GitHub, GitLab, etc.).
4. Provide clear and structured task progress reports.
## Components
- **Codex System Prompt**: Codex will act as the task manager, processing task updates and tracking changes.
- **Log Files**: Changes will be saved as text files in `.ai_update/`.
- **Git Integration**: Each task update will be automatically committed and pushed to a Git repository.
### 1. Codex System Prompt
Codex needs a **system prompt** that instructs it to track tasks, store updates in `.ai_update/`, and commit them to Git. Here is the **system prompt** for Codex:
```python
"""
You are a highly capable task manager for a development team working on a software project. Your job is to:
1. Track the tasks and changes in the project, including assigning tasks to team members and tracking progress.
2. Log every change made to the tasks, including updates to the status, comments, and assigned team members.
3. Ensure that no task is missed and that progress is clearly reported.
4. Store all updates, status changes, and task comments in a folder named `.ai_update/` on the local disk.
5. Automatically commit and push changes to the Git repository (on GitHub, GitLab, or other Git platforms) whenever updates are made.
6. Provide clear and detailed reports on what has been done so far and what tasks remain.
The folder `.ai_update/` will contain a log file where every change to a task is recorded, along with:
- Task name
- Updated status (if changed)
- Assignee (if changed)
- Any new comments (with timestamps)
- A summary of task progress
For each change:
1. Save the task update to the `.ai_update/` folder as a new log file.
2. Commit the new log file with a descriptive commit message, such as "Updated task [task_name] status to [status]".
3. Push the changes to the remote Git repository, ensuring that the changes are tracked properly.
4. If no change occurs in a task, do not log or push.
Ensure that no steps are skipped in the task completion process, and when a task is marked as "completed", ensure that all relevant information has been logged and pushed.
Provide the following feedback format for every task:
1. Task name
2. Current status
3. Comments and updates
4. Task progress summary
The `.ai_update/` folder will serve as a log of your progress. You should always commit the updates to Git in a way that shows a clear history of the changes.
Example of the log entry in `.ai_update/` folder:
Task: [task_name]
- Status: [status]
- Assignee: [assignee_name]
- Comment: [comment]
- Timestamp: [timestamp]
"""

0
tests/__init__.py Normal file
View file

58
tests/test_aggregation.py Normal file
View file

@ -0,0 +1,58 @@
"""Unit tests for search/aggregation.py"""
import sys
import os
_SEARCH_DIR = os.path.join(os.path.dirname(__file__), "..", "search")
sys.path.insert(0, _SEARCH_DIR)
os.environ.setdefault("EMBEDDINGS_DENSE_URL", "http://localhost/embed")
os.environ.setdefault("RERANKER_URL", "http://localhost/rerank")
os.environ.setdefault("QDRANT_URL", "http://localhost:6333")
os.environ.setdefault("API_KEY", "test-key")
from aggregation import aggregate_message_ids
from config import TOP_K
def _point(message_ids: list[str]):
"""Fake qdrant point with payload."""
class FakePoint:
payload = {"metadata": {"message_ids": message_ids}}
return FakePoint()
class TestAggregateMessageIds:
def test_empty_inputs(self):
result = aggregate_message_ids([], [])
assert result == []
def test_basic_dedup(self):
head = [_point(["m1", "m2"]), _point(["m2", "m3"])]
result = aggregate_message_ids(head, [])
assert result.count("m2") == 1
def test_head_before_tail(self):
head = [_point(["head_msg"])]
tail = [_point(["tail_msg"])]
result = aggregate_message_ids(head, tail)
assert result.index("head_msg") < result.index("tail_msg")
def test_top_k_limit(self):
# Create enough points to exceed TOP_K
points = [_point([f"m{i}"]) for i in range(TOP_K + 20)]
result = aggregate_message_ids(points, [])
assert len(result) <= TOP_K
def test_cross_point_dedup(self):
head = [_point(["shared"]), _point(["shared", "unique"])]
result = aggregate_message_ids(head, [])
assert result.count("shared") == 1
assert "unique" in result
def test_tail_fills_after_head(self):
head = [_point(["h1"])]
tail = [_point(["t1"]), _point(["t2"])]
result = aggregate_message_ids(head, tail)
assert "h1" in result
assert "t1" in result
assert "t2" in result

141
tests/test_chunking.py Normal file
View file

@ -0,0 +1,141 @@
"""Unit tests for index/chunking.py"""
import sys
import os
_INDEX_DIR = os.path.join(os.path.dirname(__file__), "..", "index")
sys.path.insert(0, _INDEX_DIR)
from chunking import (
build_chunks,
_split_windows,
WINDOW_MAX_MESSAGES,
WINDOW_MAX_CHARS,
TIME_GAP_SECONDS,
)
from cleaning import CleanedMessage
from index_schemas import Message
def _make_message(id: str, time: int, text: str = "hello", **kwargs) -> Message:
defaults = dict(
thread_sn=None,
sender_id="user@example.com",
file_snippets="",
parts=None,
mentions=None,
member_event=None,
is_system=False,
is_hidden=False,
is_forward=False,
is_quote=False,
)
defaults.update(kwargs)
return Message(id=id, time=time, text=text, **defaults)
def _make_cleaned(id: str, time: int, text: str = "hello") -> CleanedMessage:
return CleanedMessage(
id=id,
sender_id="user@x.com",
time=time,
thread_sn=None,
text=text,
parts=[],
mentions=[],
member_event_text="",
file_info=[],
is_system=False,
is_forward=False,
is_quote=False,
)
class TestBuildChunks:
def test_empty_new_messages(self):
result = build_chunks([], [])
assert result == []
def test_single_message(self):
msgs = [_make_message("m1", 1000000, text="A simple message")]
result = build_chunks([], msgs)
assert len(result) == 1
assert "m1" in result[0].message_ids
def test_message_ids_preserved(self):
msgs = [
_make_message("m1", 1000000, text="First"),
_make_message("m2", 1000100, text="Second"),
]
result = build_chunks([], msgs)
all_ids = [mid for chunk in result for mid in chunk.message_ids]
assert "m1" in all_ids
assert "m2" in all_ids
def test_different_content_fields(self):
msgs = [_make_message("m1", 1000000, text="test")]
result = build_chunks([], msgs)
chunk = result[0]
assert chunk.page_content
assert chunk.dense_content
assert chunk.sparse_content
def test_overlap_appears_in_chunk(self):
overlap = [_make_message("o1", 999000, text="overlap message")]
new_msgs = [_make_message("m1", 1000000, text="new message")]
result = build_chunks(overlap, new_msgs)
assert len(result) >= 1
# overlap ids should NOT be in message_ids (they're context only)
assert "o1" not in result[0].message_ids
assert "m1" in result[0].message_ids
def test_time_gap_splits_window(self):
msgs = [
_make_message("m1", 1000000, text="morning message"),
_make_message("m2", 1000000 + TIME_GAP_SECONDS + 1, text="evening message"),
]
result = build_chunks([], msgs)
# large time gap should create 2 chunks
assert len(result) == 2
def test_empty_messages_skipped(self):
msgs = [
_make_message("m1", 1000000, text=""),
_make_message("m2", 1000100, text="real content"),
]
result = build_chunks([], msgs)
all_ids = [mid for chunk in result for mid in chunk.message_ids]
assert "m1" not in all_ids
assert "m2" in all_ids
def test_hard_limit_for_chunk_content_lengths(self):
long_msg = _make_message("m1", 1000000, text="x" * (WINDOW_MAX_CHARS * 3))
result = build_chunks([], [long_msg])
assert len(result) == 1
chunk = result[0]
assert len(chunk.page_content) <= WINDOW_MAX_CHARS
assert len(chunk.dense_content) <= WINDOW_MAX_CHARS
assert len(chunk.sparse_content) <= WINDOW_MAX_CHARS
assert chunk.message_ids == ["m1"]
class TestSplitWindows:
def test_empty(self):
assert _split_windows([]) == []
def test_single(self):
msgs = [_make_cleaned("m1", 1000000)]
windows = _split_windows(msgs)
assert len(windows) == 1
def test_time_gap_splits(self):
msgs = [
_make_cleaned("m1", 1000000),
_make_cleaned("m2", 1000000 + TIME_GAP_SECONDS + 1),
]
windows = _split_windows(msgs)
assert len(windows) == 2
def test_max_messages_splits(self):
msgs = [_make_cleaned(f"m{i}", 1000000 + i * 10) for i in range(WINDOW_MAX_MESSAGES + 2)]
windows = _split_windows(msgs)
assert len(windows) >= 2

170
tests/test_cleaning.py Normal file
View file

@ -0,0 +1,170 @@
"""Unit tests for index/cleaning.py"""
import sys
import os
_INDEX_DIR = os.path.join(os.path.dirname(__file__), "..", "index")
sys.path.insert(0, _INDEX_DIR)
import pytest
from cleaning import (
normalize_unicode,
parse_file_snippets,
normalize_member_event,
normalize_part,
clean_message,
)
from index_schemas import Message
def _make_message(**kwargs) -> Message:
defaults = dict(
id="msg1",
thread_sn=None,
time=1000000,
text="",
sender_id="user@example.com",
file_snippets="",
parts=None,
mentions=None,
member_event=None,
is_system=False,
is_hidden=False,
is_forward=False,
is_quote=False,
)
defaults.update(kwargs)
return Message(**defaults)
class TestNormalizeUnicode:
def test_removes_zero_width(self):
assert "\u200b" not in normalize_unicode("hello\u200bworld")
assert "\u200c" not in normalize_unicode("a\u200cb")
assert "\ufeff" not in normalize_unicode("\ufefftext")
def test_collapses_whitespace(self):
result = normalize_unicode(" too many spaces ")
assert " " not in result
def test_normalizes_newlines(self):
result = normalize_unicode("line1\r\nline2\rline3")
assert "\r" not in result
def test_collapses_multiple_newlines(self):
result = normalize_unicode("a\n\n\n\nb")
assert "\n\n\n" not in result
def test_preserves_url(self):
url = "https://example.com/path?q=1&page=2"
assert url in normalize_unicode(url)
def test_preserves_email(self):
email = "user@corp.example"
assert email in normalize_unicode(email)
class TestParseFileSnippets:
def test_empty_string(self):
assert parse_file_snippets("") == []
def test_valid_list(self):
raw = '[{"name": "doc.pdf", "mime": "application/pdf"}]'
result = parse_file_snippets(raw)
assert len(result) == 1
assert result[0]["name"] == "doc.pdf"
def test_valid_dict(self):
raw = '{"name": "file.txt"}'
result = parse_file_snippets(raw)
assert len(result) == 1
def test_invalid_json(self):
assert parse_file_snippets("{broken json}") == []
def test_whitespace_only(self):
assert parse_file_snippets(" ") == []
class TestNormalizeMemberEvent:
def test_none_event(self):
assert normalize_member_event(None) == ""
def test_add_members(self):
event = {"type": "addMembers", "members": ["alice@example.com", "bob@example.com"]}
result = normalize_member_event(event)
assert "alice@example.com" in result
assert "added to chat" in result
def test_unknown_event(self):
event = {"type": "banUser", "member": "x@example.com"}
result = normalize_member_event(event)
assert "banUser" in result
class TestNormalizePart:
def test_text_part(self):
part = {"mediaType": "text", "text": "hello"}
result = normalize_part(part)
assert result["type"] == "text"
assert result["text"] == "hello"
def test_quote_part(self):
part = {"mediaType": "quote", "sn": "alice@x.com", "text": "original text"}
result = normalize_part(part)
assert result["type"] == "quote"
assert "alice@x.com" in result["text"]
assert "original text" in result["text"]
def test_forward_part(self):
part = {"mediaType": "forward", "sn": "channel@x.com", "text": "forwarded"}
result = normalize_part(part)
assert result["type"] == "forward"
assert "forwarded" in result["text"]
def test_unknown_media_type(self):
part = {"mediaType": "sticker", "text": ""}
result = normalize_part(part)
assert "sticker" in result["type"]
class TestCleanMessage:
def test_empty_message_is_empty(self):
msg = _make_message()
cleaned = clean_message(msg)
assert cleaned.is_empty
def test_text_message(self):
msg = _make_message(text="Hello world")
cleaned = clean_message(msg)
assert not cleaned.is_empty
assert cleaned.text == "Hello world"
def test_parts_extracted(self):
msg = _make_message(
parts=[{"mediaType": "text", "text": "from parts"}]
)
cleaned = clean_message(msg)
assert not cleaned.is_empty
assert any("from parts" in p["text"] for p in cleaned.parts)
def test_member_event_extracted(self):
msg = _make_message(
is_system=True,
member_event={"type": "addMembers", "members": ["u@x.com"]},
)
cleaned = clean_message(msg)
assert not cleaned.is_empty
assert "u@x.com" in cleaned.member_event_text
def test_file_snippets_parsed(self):
msg = _make_message(
file_snippets='[{"name": "report.pdf", "mime": "application/pdf"}]'
)
cleaned = clean_message(msg)
assert not cleaned.is_empty
assert cleaned.file_info[0]["name"] == "report.pdf"
def test_zero_width_stripped_from_text(self):
msg = _make_message(text="hello\u200bworld")
cleaned = clean_message(msg)
assert "\u200b" not in cleaned.text

118
tests/test_query_builder.py Normal file
View file

@ -0,0 +1,118 @@
"""Unit tests for search/query_builder.py (pure logic only, no HTTP)"""
import sys
import os
_SEARCH_DIR = os.path.join(os.path.dirname(__file__), "..", "search")
sys.path.insert(0, _SEARCH_DIR)
# Stub env vars before importing search modules
os.environ.setdefault("EMBEDDINGS_DENSE_URL", "http://localhost/embed")
os.environ.setdefault("RERANKER_URL", "http://localhost/rerank")
os.environ.setdefault("QDRANT_URL", "http://localhost:6333")
os.environ.setdefault("API_KEY", "test-key")
from schemas import Entities, Question
from query_builder import (
build_primary_query,
build_extra_dense_queries,
build_sparse_query,
build_entity_tokens,
)
def _q(**kwargs) -> Question:
defaults = dict(text="default question")
defaults.update(kwargs)
return Question(**defaults)
class TestBuildPrimaryQuery:
def test_uses_search_text_over_text(self):
q = _q(text="original", search_text="refined query")
assert build_primary_query(q) == "refined query"
def test_fallback_to_text(self):
q = _q(text="fallback text", search_text="")
assert build_primary_query(q) == "fallback text"
def test_strips_whitespace(self):
q = _q(text=" trimmed ")
assert build_primary_query(q) == "trimmed"
def test_collapses_internal_spaces(self):
q = _q(text="too many spaces")
result = build_primary_query(q)
assert " " not in result
class TestBuildExtraDenseQueries:
def test_no_extras_when_none(self):
q = _q(text="q")
assert build_extra_dense_queries(q) == []
def test_includes_variants(self):
q = _q(text="q", variants=["var1", "var2"])
extras = build_extra_dense_queries(q)
assert "var1" in extras
assert "var2" in extras
def test_includes_hyde(self):
q = _q(text="q", hyde=["hypothetical answer"])
extras = build_extra_dense_queries(q)
assert "hypothetical answer" in extras
def test_skips_empty_strings(self):
q = _q(text="q", variants=["", " ", "valid"])
extras = build_extra_dense_queries(q)
assert "" not in extras
assert " " not in extras
assert "valid" in extras
class TestBuildSparseQuery:
def test_uses_keywords_when_present(self):
q = _q(text="question", keywords=["go", "golang", "performance"])
result = build_sparse_query(q)
assert "go" in result
assert "golang" in result
def test_fallback_to_primary_when_no_keywords(self):
q = _q(text="fallback question", search_text="refined")
result = build_sparse_query(q)
assert result == "refined"
def test_empty_keywords_fallback(self):
q = _q(text="my question", keywords=[])
result = build_sparse_query(q)
assert result == "my question"
class TestBuildEntityTokens:
def test_no_entities(self):
q = _q(text="q")
assert build_entity_tokens(q) == []
def test_people_extracted(self):
q = _q(text="q", entities=Entities(people=["Alice", "Bob"]))
tokens = build_entity_tokens(q)
assert "Alice" in tokens
assert "Bob" in tokens
def test_all_entity_fields(self):
q = _q(
text="q",
entities=Entities(
people=["Alice"],
emails=["alice@corp.com"],
documents=["report.pdf"],
names=["Project X"],
links=["https://example.com"],
),
)
tokens = build_entity_tokens(q)
assert len(tokens) == 5
def test_strips_whitespace(self):
q = _q(text="q", entities=Entities(people=[" Alice "]))
tokens = build_entity_tokens(q)
assert "Alice" in tokens

110
tests/test_rendering.py Normal file
View file

@ -0,0 +1,110 @@
"""Unit tests for index/rendering.py"""
import sys
import os
_INDEX_DIR = os.path.join(os.path.dirname(__file__), "..", "index")
sys.path.insert(0, _INDEX_DIR)
from cleaning import CleanedMessage
from rendering import render_page_content, render_dense_content, render_sparse_content
def _make_cleaned(**kwargs) -> CleanedMessage:
defaults = dict(
id="msg1",
sender_id="alice@example.com",
time=1700000000,
thread_sn=None,
text="",
parts=[],
mentions=[],
member_event_text="",
file_info=[],
is_system=False,
is_forward=False,
is_quote=False,
)
defaults.update(kwargs)
return CleanedMessage(**defaults)
class TestRenderPageContent:
def test_text_message(self):
msg = _make_cleaned(text="Hello world")
result = render_page_content(msg)
assert "alice@example.com" in result
assert "Hello world" in result
def test_quote_part(self):
msg = _make_cleaned(
parts=[{"type": "quote", "text": "[quote from bob]: original"}]
)
result = render_page_content(msg)
assert "[quote from bob]" in result
def test_forward_part(self):
msg = _make_cleaned(
parts=[{"type": "forward", "text": "[forwarded from channel]: content"}],
is_forward=True,
)
result = render_page_content(msg)
assert "forwarded" in result
def test_member_event(self):
msg = _make_cleaned(
member_event_text="[system: alice@x.com added to chat]",
is_system=True,
)
result = render_page_content(msg)
assert "added to chat" in result
def test_file_attachment(self):
msg = _make_cleaned(
file_info=[{"name": "report.pdf", "mime": "application/pdf", "url": "", "date": ""}]
)
result = render_page_content(msg)
assert "report.pdf" in result
class TestRenderDenseContent:
def test_includes_timestamp(self):
msg = _make_cleaned(text="test", time=1700000000)
result = render_dense_content(msg)
assert "2023-" in result # UTC date
def test_includes_sender(self):
msg = _make_cleaned(text="hi", sender_id="bob@corp.com")
result = render_dense_content(msg)
assert "sender:bob@corp.com" in result
def test_forward_marker(self):
msg = _make_cleaned(is_forward=True, parts=[{"type": "forward", "text": "[forwarded]: x"}])
result = render_dense_content(msg)
assert "type:forward" in result
def test_mentions_included(self):
msg = _make_cleaned(
text="hey",
mentions=["charlie@corp.com"],
)
result = render_dense_content(msg)
assert "charlie@corp.com" in result
class TestRenderSparseContent:
def test_includes_sender(self):
msg = _make_cleaned(text="hello")
result = render_sparse_content(msg)
assert "alice@example.com" in result
def test_includes_mentions(self):
msg = _make_cleaned(text="ping", mentions=["dave@corp.com"])
result = render_sparse_content(msg)
assert "dave@corp.com" in result
def test_includes_filename(self):
msg = _make_cleaned(
file_info=[{"name": "budget.xlsx", "mime": "", "url": "", "date": ""}]
)
result = render_sparse_content(msg)
assert "budget.xlsx" in result