From 83e77ea8c9d8e4519b76e2fe2461e646c84b2072 Mon Sep 17 00:00:00 2001 From: n1kt4r Date: Thu, 15 Oct 2026 13:40:06 +0500 Subject: [PATCH] Initial commit: Pegas Memory (PG + Qdrant + Memgraph, MCP server, 14 tools) --- .gitignore | 6 + README.md | 72 +++++ ROADMAP.md | 97 +++++++ mcp_server.py | 509 +++++++++++++++++++++++++++++++++ pegas_memory/__init__.py | 1 + pegas_memory/code_graph.py | 164 +++++++++++ pegas_memory/code_symbols.py | 108 +++++++ pegas_memory/config.py | 23 ++ pegas_memory/db.py | 85 ++++++ pegas_memory/dedup.py | 34 +++ pegas_memory/distill.py | 72 +++++ pegas_memory/embedding.py | 27 ++ pegas_memory/inference.py | 104 +++++++ pegas_memory/ingest.py | 84 ++++++ pegas_memory/memgraph_store.py | 136 +++++++++ pegas_memory/outbox_sync.py | 69 +++++ pegas_memory/pipeline.py | 83 ++++++ pegas_memory/qdrant_store.py | 57 ++++ pegas_memory/research.py | 84 ++++++ pegas_memory/verify.py | 158 ++++++++++ requirements.txt | 11 + schema.sql | 159 ++++++++++ scripts/init_db.py | 26 ++ scripts/smoke_test.py | 32 +++ 24 files changed, 2201 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 ROADMAP.md create mode 100644 mcp_server.py create mode 100644 pegas_memory/__init__.py create mode 100644 pegas_memory/code_graph.py create mode 100644 pegas_memory/code_symbols.py create mode 100644 pegas_memory/config.py create mode 100644 pegas_memory/db.py create mode 100644 pegas_memory/dedup.py create mode 100644 pegas_memory/distill.py create mode 100644 pegas_memory/embedding.py create mode 100644 pegas_memory/inference.py create mode 100644 pegas_memory/ingest.py create mode 100644 pegas_memory/memgraph_store.py create mode 100644 pegas_memory/outbox_sync.py create mode 100644 pegas_memory/pipeline.py create mode 100644 pegas_memory/qdrant_store.py create mode 100644 pegas_memory/research.py create mode 100644 pegas_memory/verify.py create mode 100644 requirements.txt create mode 100644 schema.sql create mode 100644 scripts/init_db.py create mode 100644 scripts/smoke_test.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..55368cd --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +.venv/ +__pycache__/ +*.pyc +*.pyo +.pytest_cache/ +*.egg-info/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..639b15c --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# Pegas Memory + +Подсистема памяти и базы знаний агента **Alicorn** (рабочее имя «Pegas»). + +- **PostgreSQL** — единственный источник истины (текст, метаданные, трейсы, верификация). +- **Qdrant** — векторный срез (EmbeddingGemma 2, dim=768). +- **Memgraph** — графовый срез (связи атомов, зависимости кода). +- **Единица знания** — типизированный атом: text | definition | logical_rule | code | table | fact. + +Полный план: [ROADMAP.md](./ROADMAP.md) + +## Структура + +``` +pegas_memory/ + config.py — конфиг (адреса сервисов, имена коллекций) + db.py — PostgreSQL (источник истины) + embedding.py — EmbeddingGemma 2 через Ollama (dim=768) + qdrant_store.py — векторный срез + memgraph_store.py— графовый срез + atoms.py — модель атома, дистилляция, верификация + ingest.py — парсеры text/pdf/epub/web + outbox_sync.py — PG → Qdrant + Memgraph + research.py — цикл исследования пробелов + code_symbols.py — tree-sitter извлечение символов + code_graph.py — граф зависимостей кода (calls/imports/inherits) + inference.py — логическая инференция (forward-chaining в Memgraph) + dedup.py — семантическая дедупликация атомов (эмбеддинги, порог 0.90) + verify.py — верификация: self-critique (LLM) + кросс-подтверждение +schema.sql — DDL PostgreSQL +``` + + +## Подключение к DeepSeek Harness (dsh) + +Память подключается как **MCP-сервер** (паттерн dsh, как SearXNG/Gitea/Outline/Playwright): + +- **Сервер**: [mcp_server.py](./mcp_server.py) — чистый stdlib, JSON-RPC 2.0, stdio. +- **Инструменты** (видны модели как `mcp__pegas-memory__*`): + `memory_search`, `memory_store`, `memory_ingest`, `memory_add_gap`, + `memory_list_gaps`, `memory_research`, `memory_stats`, `memory_get_context`. +- **Регистрация**: в `~/.dsh/profiles/web/cordis.patch.yml` через `@deepseek-ai/dsh-mcp-client`. +- **Live-reload**: профиль web имеет `patchReload: live` — изменения конфига + подхватываются без рестарта (проверено: сервер спавнится автоматически). + +```yaml +- id: mcp-pegas-memory + name: '@deepseek-ai/dsh-mcp-client' + config: + serverName: pegas-memory + transport: stdio + command: /home/neo/workspace/pegas-memory/.venv/bin/python + args: + - /home/neo/workspace/pegas-memory/mcp_server.py + toolCallTimeoutMs: 180000 +``` + +## Доступность зарубежных сайтов и прокси + +Прямое подключение работает для большинства сайтов (github, wikipedia, postgresql.org), +но часть блокирует по российскому IP (memgraph.com — ConnectTimeout) или ставит +Cloudflare anti-bot (neo4j.com — 403). Поведение research-цикла: + +1. Полная загрузка URL (httpx, таймаут 30s). +2. При блокировке — fallback на сниппет из SearXNG (локальный инстанс). + +Для заблокированных сайтов можно включить SOCKS5/HTTP-прокси: +```bash +export PROXY_URL="socks5://127.0.0.1:1080" # hysteria2 +``` +Когда прокси задан, `parse_web` ходит через него; иначе — напрямую. +Поддержка SOCKS5 уже установлена (`httpx[socks]`). diff --git a/ROADMAP.md b/ROADMAP.md new file mode 100644 index 0000000..48c3db4 --- /dev/null +++ b/ROADMAP.md @@ -0,0 +1,97 @@ +# Pegas Memory — Roadmap + +Подсистема памяти и базы знаний агента Alicorn («Pegas»). Реализация на Python как отдельный +сервис, который Rust-ядро Alicorn вызывает через MCP/HTTP. Спецификация — в +`alicorn-agent/ARCHITECTURE.md` (ADR-4, ADR-11, ADR-12). + +## Принятые решения (по итогам обсуждения) + +1. **PostgreSQL = единственный источник истины.** Текст, метаданные, трейсы, верификация, outbox. +2. **Qdrant = векторный срез** (EmbeddingGemma 2, dim=768). Замена pgvector. +3. **Memgraph = графовый срез** (связи атомов, зависимости кода). Замена Apache AGE. +4. **pgvector и AGE — НЕ используем.** Векторы только в Qdrant, граф только в Memgraph. +5. **Единица знания — типизированный «атом»**, не чанк сырого текста. + Типы: `text | definition | logical_rule | code | table | fact`. +6. **Память агента (episodic) ≠ база знаний (curated).** Разные таблицы/субсистемы. +7. **Код хранится отдельно**: content-addressed исходники + tree-sitter символы + граф зависимостей. +8. **Логические знания** — структурные правила (pattern→conclusion) с инференцией в графе. +9. **Исследование пробелов** — замкнутый цикл: detect→plan→acquire→parse→extract→value→verify→commit. + +## Инфраструктура (проверено на сервере) + +| Компонент | Адрес | Состояние | +|---|---|---| +| PostgreSQL 16 | atom-db :5434 (user atom, db content_atom) | работает | +| Qdrant | :6333 | работает | +| Memgraph (bolt) | :7687 | работает | +| Ollama | :11434 | работает | +| EmbeddingGemma 2 | `embeddinggemma-2-740m-Q8_0` | dim=768, R@1 0.922 (лучшая на ru) | +| LLM-генерация | `lfm25-24b-a2b:128k`, `qwen3.5:4b`, `gemma4:E4B` | работает | + +> EmbeddingGemma 2: полная размерность 768 (проверено curl-тестом). Q8_0 — оптимальный +> баланс качество/память (MRR 0.944, не теряет качество). Не путать с v1 `embeddinggemma:300m`. + +## Модель данных (PostgreSQL — канон) + +``` +document — источники (web|pdf|epub|text|code_repo), uri, authority, content_hash +chunk — сырые сегменты (доказательства, locator) — НЕ для retrieval напрямую +atom — типизированные единицы знания (claim + content + confidence + status) +atom_link — связи (inference|cause|reference|contradicts|depends_on|calls|defines) +citation — доказательная база (atom → document → locator → quoted_text) +code_symbol — tree-sitter символы (repo, path, language, name, kind, signature) +episodic_trace— память агента (шаги/действия) — отдельно от базы знаний +research_gap — выявленные пробелы для исследования +outbox — CDC-журнал для синка в Qdrant + Memgraph +``` + +Статусы атома: `candidate → verified | stale | contested`. +Векторы НЕ хранятся в Postgres — только в Qdrant (payload: type/confidence/status/doc_id). + +## Фазы + +### Фаза 1 — Фундамент (MVP) ✅ ЗАВЕРШЕНА +- [x] Структура проекта + venv + зависимости +- [x] DDL Postgres (schema.sql) +- [x] Клиент эмбеддингов EmbeddingGemma 2 (dim=768) +- [x] Клиент Qdrant (коллекция `alicorn_atoms`) +- [x] Клиент Memgraph (узлы atom + рёбра atom_link) +- [x] Ингест: text/pdf/epub/web → чанки +- [x] Дистилляция чанка → атомы (LLM) +- [x] Outbox-синк PG → Qdrant + Memgraph +- [x] Smoke-тест end-to-end + +### Фаза 2 — Исследование пробелов ✅ (реализовано, snippet-fallback) +- [x] Детект пробелов (research_gap + add_gap/list_gaps) +- [x] Планирование (приоритизированная очередь пробелов) +- [x] Acquire: SearXNG web-поиск (3 источника) + pdf/epub/text +- [x] Оценка ценности (verify_atom + min_confidence) +- [~] Верификация (verify_atom готов; полный автономный прогон требует прокси для зарубежных сайтов) + +### Фаза 3 — Код ✅ / Логика ✅ +- [x] tree-sitter: извлечение символов (function/class/method/type) — 43 символа сохранено +- [x] Граф зависимостей кода (calls/imports/inherits) — 310 зависимостей, 162 узла CodeSymbol +- [x] Логические правила + инференция в Memgraph (forward-chaining до fixpoint, 38 рёбер выведено) + +### Фаза 4 — Интеграция с DSH (MCP) ✅ ВЫПОЛНЕНА +- [x] MCP-сервер (memory_search/store/ingest/add_gap/list_gaps/research/stats/get_context) +- [x] Подключение к DSH через @deepseek-ai/dsh-mcp-client (cordis.patch.yml, live-reload) +- [ ] Подключение к Rust-ядру Alicorn (Planner/Researcher) — когда ядро будет готово + +## Дополнительно реализовано +- Опциональный прокси (PROXY_URL, SOCKS5/HTTP) + улучшенный User-Agent для полной загрузки страниц. + +- Фильтрация артефактов графа кода (dst="." от относительных импортов). +- Дедупликация в memory_store (опциональный dedup). + (после MVP) +- Семантическая дедупликация атомов (эмбеддинги EmbeddingGemma 2, порог 0.90, конфигурируемо). +- Фильтрация шумных logical_rule (только kind-рёбра) в инференции. + +## Критические риски +- EmbeddingGemma 2 dim=768 — не совместима с bge-m3 (1024) → отдельная коллекция, без миграции. +- Memgraph in-memory: объём графа ограничен RAM. +- Дистилляция зависит от качества локальной LLM — нужна верификация. + +## Как проверять +- recall@k на реальных запросах против существующей `knowledge_base` (bge-m3 384-dim). +- Latency эмбеддинга + потребление GPU параллельно с LLM (a2000 12 GB). diff --git a/mcp_server.py b/mcp_server.py new file mode 100644 index 0000000..e895c19 --- /dev/null +++ b/mcp_server.py @@ -0,0 +1,509 @@ +#!/usr/bin/env python3 +"""MCP-сервер Pegas Memory — память и база знаний агента Alicorn. + +Оборачивает pegas_memory (PostgreSQL = источник истины, Qdrant = векторы, +Memgraph = граф) в MCP-инструменты для DeepSeek Harness (dsh). + +Транспорт — stdio (JSON-RPC 2.0, MCP 2024-11-05). Только stdlib + pegas_memory. +Запускается из venv проекта: .venv/bin/python mcp_server.py +""" + +from __future__ import annotations + +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +PROTOCOL_VERSION = "2024-11-05" +SERVER_NAME = "pegas-memory" +SERVER_VERSION = "0.1.0" + +# -------------------------------------------------------------------------- +# Ленивые глобальные ресурсы (соединения открываются при первом обращении) +# -------------------------------------------------------------------------- +_state = {"conn": None} + + +def _conn(): + import psycopg + if _state["conn"] is None or _state["conn"].closed: + from pegas_memory import config + _state["conn"] = psycopg.connect(config.PG_DSN, row_factory=dict_row) + return _state["conn"] + + +def dict_row(cursor): + cols = [d.name for d in cursor.description] + return lambda row: dict(zip(cols, row)) + + +def _sync(): + """Прогнать outbox → Qdrant + Memgraph (ошибки не фатальны).""" + from pegas_memory import outbox_sync + try: + outbox_sync.sync_all(_conn()) + except Exception as exc: # Qdrant/Memgraph недоступны — не роняем инструмент + print(f"[pegas-memory] sync skipped: {exc}", file=sys.stderr) + + +# -------------------------------------------------------------------------- +# Инструменты +# -------------------------------------------------------------------------- +def _tool_search(args): + from pegas_memory import qdrant_store + query = str(args.get("query") or "").strip() + if not query: + return json.dumps({"error": "пустой query"}, ensure_ascii=False), True + top_k = min(int(args.get("top_k") or 5), 20) + type_filter = args.get("type") or None + status_filter = args.get("status") or None + hits = qdrant_store.search(query, top_k=top_k, + type_filter=type_filter, status_filter=status_filter) + return json.dumps({"query": query, "hits": hits}, ensure_ascii=False, indent=2), False + + +def _tool_store(args): + from pegas_memory import db, dedup + type_ = str(args.get("type") or "fact").strip() + claim = args.get("claim") + content = str(args.get("content") or "").strip() + if not content: + return json.dumps({"error": "пустой content"}, ensure_ascii=False), True + if type_ not in ("text", "definition", "logical_rule", "code", "table", "fact"): + return json.dumps({"error": f"недопустимый type: {type_}"}, ensure_ascii=False), True + confidence = float(args.get("confidence") or 0.5) + # семантическая дедупликация (опционально) + if args.get("dedup", True): + dup_id = dedup.find_duplicate((claim or "") + " " + content[:800]) + if dup_id is not None: + return json.dumps({"atom_id": dup_id, "created": False, "duplicate_of": dup_id}, + ensure_ascii=False), False + atom_id, created = db.upsert_atom( + _conn(), type=type_, claim=claim, content=content, + confidence=confidence, status="candidate", + metadata=(args.get("metadata") or None), + ) + if created: + db.enqueue_outbox(_conn(), entity_type="atom", entity_id=atom_id) + _conn().commit() + _sync() + return json.dumps({"atom_id": atom_id, "created": created}, ensure_ascii=False), False + + +def _tool_ingest(args): + from pegas_memory import pipeline + text = str(args.get("text") or "").strip() + url = args.get("url") + path = args.get("path") + if text: + stats = pipeline.ingest_text(_conn(), text, title=args.get("title"), min_confidence=0.4) + elif url: + stats = pipeline.ingest_url(_conn(), url, min_confidence=0.4) + elif path: + stats = pipeline.ingest_file(_conn(), path, min_confidence=0.4) + else: + return json.dumps({"error": "нужен text|url|path"}, ensure_ascii=False), True + _sync() + return json.dumps(stats, ensure_ascii=False), False + + +def _tool_add_gap(args): + from pegas_memory import research + question = str(args.get("question") or "").strip() + if not question: + return json.dumps({"error": "пустой question"}, ensure_ascii=False), True + research.add_gap(_conn(), question, + reason=str(args.get("reason") or "user_query"), + priority=float(args.get("priority") or 0.5)) + return json.dumps({"added": question}, ensure_ascii=False), False + + +def _tool_list_gaps(args): + from pegas_memory import research + gaps = research.detect_gaps(_conn()) + return json.dumps({"gaps": gaps}, ensure_ascii=False, indent=2), False + + +def _tool_research(args): + from pegas_memory import research + question = str(args.get("question") or "").strip() + if not question: + return json.dumps({"error": "пустой question"}, ensure_ascii=False), True + result = research.research_question(_conn(), question) + _sync() + return json.dumps(result, ensure_ascii=False, indent=2), False + + +def _tool_stats(args): + c = _conn() + with c.cursor() as cur: + cur.execute("SELECT count(*) AS n FROM pm_atom") + atoms = cur.fetchone()["n"] + cur.execute("SELECT count(*) AS n FROM pm_document") + docs = cur.fetchone()["n"] + cur.execute("SELECT count(*) AS n FROM pm_research_gap WHERE status='open'") + gaps = cur.fetchone()["n"] + return json.dumps({"atoms": atoms, "documents": docs, "open_gaps": gaps}, + ensure_ascii=False), False + + +def _tool_get_context(args): + """Собрать контекст для агента: поиск + связанные атомы.""" + from pegas_memory import qdrant_store + query = str(args.get("query") or "").strip() + if not query: + return json.dumps({"error": "пустой query"}, ensure_ascii=False), True + top_k = min(int(args.get("top_k") or 5), 20) + hits = qdrant_store.search(query, top_k=top_k) + parts = [f"- [{h['score']:.3f}] ({h['type']}) {h['claim']}" for h in hits] + return json.dumps({"query": query, "context": parts, "hits": hits}, + ensure_ascii=False, indent=2), False + + +def _tool_store_code(args): + from pegas_memory import code_symbols + language = str(args.get("language") or "").strip() + source = str(args.get("source") or "") + path = args.get("path") or "" + if not language or not source: + return json.dumps({"error": "нужны language и source"}, ensure_ascii=False), True + repo = args.get("repo") or "" + n = code_symbols.store_symbols(_conn(), source, language, repo, path) + return json.dumps({"symbols_stored": n}, ensure_ascii=False), False + + +def _tool_search_code(args): + query = str(args.get("query") or "").strip() + if not query: + return json.dumps({"error": "пустой query"}, ensure_ascii=False), True + limit = min(int(args.get("top_k") or 10), 50) + c = _conn() + with c.cursor() as cur: + cur.execute( + "SELECT id, repo, path, language, symbol_name, kind, signature, line_start " + "FROM pm_code_symbol WHERE symbol_name ILIKE %s " + "ORDER BY similarity(symbol_name, %s) DESC LIMIT %s", + (f"%{query}%", query, limit), + ) + rows = cur.fetchall() + return json.dumps({"query": query, "symbols": rows}, ensure_ascii=False, indent=2), False + + +def _tool_code_graph(args): + from pegas_memory import code_symbols, code_graph + language = str(args.get("language") or "").strip() + source = str(args.get("source") or "") + path = args.get("path") or "" + if not language or not source: + return json.dumps({"error": "нужны language и source"}, ensure_ascii=False), True + repo = args.get("repo") or "" + ns = code_symbols.store_symbols(_conn(), source, language, repo, path) + nd = code_graph.store_dependencies(_conn(), source, language, repo, path) + return json.dumps({"symbols_stored": ns, "dependencies_stored": nd}, ensure_ascii=False), False + + +def _tool_infer(args): + from pegas_memory import inference, memgraph_store + if args.get("if") and args.get("then"): + # сохранить правило и сразу применить + rule_id = inference.store_rule( + _conn(), claim=args.get("claim") or "rule", + if_=args.get("if"), then=args.get("then"), + ) + _conn().commit() + result = inference.infer(_conn()) + result["edges_total"] = memgraph_store.count_edges() + result["code_symbols"] = memgraph_store.count_code_symbols() + return json.dumps(result, ensure_ascii=False), False + + +def _tool_query_graph(args): + from pegas_memory import memgraph_store + kind = args.get("kind") or None + limit = min(int(args.get("top_k") or 50), 200) + edges = memgraph_store.query_edges(kind=kind, limit=limit) + return json.dumps({"kind": kind, "edges": edges, + "nodes": memgraph_store.count(), + "code_symbols": memgraph_store.count_code_symbols()}, + ensure_ascii=False, indent=2), False + + +def _tool_verify(args): + from pegas_memory import verify + atom_id = args.get("atom_id") + if atom_id is None: + # верифицировать все candidate + result = verify.verify_all(_conn(), self_critique_enabled=bool(args.get("self_critique", True))) + else: + result = verify.verify_atom(_conn(), int(atom_id), self_critique_enabled=bool(args.get("self_critique", True))) + return json.dumps(result, ensure_ascii=False, indent=2), False + + +TOOLS = [ + { + "name": "memory_search", + "description": "Семантический поиск по базе знаний Pegas Memory (Qdrant, EmbeddingGemma 2). " + "Возвращает типизированные атомы (text/definition/logical_rule/code/table/fact) " + "с оценкой сходства. Фильтры: type, status (candidate/verified/stale/contested).", + "inputSchema": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "Поисковый запрос."}, + "top_k": {"type": "integer", "default": 5, "description": "Число результатов (1-20)."}, + "type": {"type": "string", "description": "Фильтр по типу атома (опционально)."}, + "status": {"type": "string", "description": "Фильтр по статусу (опционально)."}, + }, + "required": ["query"], + }, + }, + { + "name": "memory_store", + "description": "Сохранить атом знания в базу (PostgreSQL + Qdrant + Memgraph). " + "type: text|definition|logical_rule|code|table|fact. Для code передавай исходник, " + "для logical_rule — JSON {if, then}, для table — JSON или markdown-таблицу.", + "inputSchema": { + "type": "object", + "properties": { + "type": {"type": "string", "description": "Тип атома."}, + "claim": {"type": "string", "description": "Короткое каноническое утверждение."}, + "content": {"type": "string", "description": "Полное значение атома."}, + "confidence": {"type": "number", "description": "Уверенность 0-1 (default 0.5)."}, + "metadata": {"type": "object", "description": "Дополнительные метаданные (опционально)."}, + "dedup": {"type": "boolean", "description": "Семантическая дедупликация (default true)."}, + }, + "required": ["type", "content"], + }, + }, + { + "name": "memory_ingest", + "description": "Загрузить документ в базу знаний: сырой текст, URL (web) или путь к файлу (pdf/epub/text). " + "Дистиллирует содержимое в атомы через локальную LLM. Возвращает статистику.", + "inputSchema": { + "type": "object", + "properties": { + "text": {"type": "string", "description": "Сырой текст для загрузки."}, + "url": {"type": "string", "description": "URL веб-страницы."}, + "path": {"type": "string", "description": "Путь к файлу (pdf/epub/text)."}, + "title": {"type": "string", "description": "Заголовок документа (для text)."}, + }, + "required": [], + }, + }, + { + "name": "memory_add_gap", + "description": "Зарегистрировать пробел в знаниях для последующего исследования.", + "inputSchema": { + "type": "object", + "properties": { + "question": {"type": "string", "description": "Вопрос/тема пробела."}, + "reason": {"type": "string", "description": "low_coverage|low_confidence|stale|contested|user_query."}, + "priority": {"type": "number", "description": "Приоритет 0-1 (default 0.5)."}, + }, + "required": ["question"], + }, + }, + { + "name": "memory_list_gaps", + "description": "Список открытых пробелов в базе знаний (по приоритету).", + "inputSchema": {"type": "object", "properties": {}, "required": []}, + }, + { + "name": "memory_research", + "description": "Полный цикл исследования вопроса: поиск в интернете (SearXNG), " + "загрузка найденных источников, дистилляция в атомы, сохранение.", + "inputSchema": { + "type": "object", + "properties": { + "question": {"type": "string", "description": "Исследовательский вопрос."}, + }, + "required": ["question"], + }, + }, + { + "name": "memory_stats", + "description": "Статистика базы знаний: число атомов, документов, открытых пробелов.", + "inputSchema": {"type": "object", "properties": {}, "required": []}, + }, + { + "name": "memory_store_code", + "description": "Сохранить исходник кода: извлекает символы (функции/классы) через tree-sitter " + "и кладёт в pm_code_symbol. language: python|javascript|typescript|rust.", + "inputSchema": { + "type": "object", + "properties": { + "language": {"type": "string", "description": "Язык: python|javascript|typescript|rust."}, + "source": {"type": "string", "description": "Исходный код."}, + "repo": {"type": "string", "description": "Имя репозитория (опционально)."}, + "path": {"type": "string", "description": "Путь файла (опционально)."}, + }, + "required": ["language", "source"], + }, + }, + { + "name": "memory_search_code", + "description": "Лексический поиск по символам кода (имена функций/классов) в базе.", + "inputSchema": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "Имя символа или его часть."}, + "top_k": {"type": "integer", "default": 10, "description": "Число результатов."}, + }, + "required": ["query"], + }, + }, + { + "name": "memory_code_graph", + "description": "Построить граф зависимостей кода: извлекает символы (tree-sitter) и зависимости " + "(calls/imports/inherits), сохраняет в PostgreSQL и Memgraph.", + "inputSchema": { + "type": "object", + "properties": { + "language": {"type": "string", "description": "Язык: python|javascript|typescript|rust."}, + "source": {"type": "string", "description": "Исходный код."}, + "repo": {"type": "string", "description": "Имя репозитория (опционально)."}, + "path": {"type": "string", "description": "Путь файла (опционально)."}, + }, + "required": ["language", "source"], + }, + }, + { + "name": "memory_infer", + "description": "Запустить логическую инференцию в Memgraph. Правила — атомы logical_rule " + "с JSON {if, then}. Можно передать if/then, чтобы добавить правило на лету.", + "inputSchema": { + "type": "object", + "properties": { + "if": {"type": "string", "description": "Условие (kind ребра) или список для композиции."}, + "then": {"type": "string", "description": "Вывод (kind ребра)."}, + "claim": {"type": "string", "description": "Описание правила."}, + }, + "required": [], + }, + }, + { + "name": "memory_query_graph", + "description": "Запросить граф: список рёбер (по kind или все), счётчики узлов.", + "inputSchema": { + "type": "object", + "properties": { + "kind": {"type": "string", "description": "Фильтр по kind ребра (опционально)."}, + "top_k": {"type": "integer", "default": 50, "description": "Число рёбер."}, + }, + "required": [], + }, + }, + { + "name": "memory_verify", + "description": "Верифицировать атом знания: self-critique (LLM) + кросс-подтверждение " + "по эмбеддингам из нескольких источников. Статус: candidate→verified/contested/stale.", + "inputSchema": { + "type": "object", + "properties": { + "atom_id": {"type": "integer", "description": "ID атома (если не указан — все candidate)."}, + "self_critique": {"type": "boolean", "description": "Включить LLM self-critique."}, + }, + "required": [], + }, + }, + + + + { + "name": "memory_get_context", + "description": "Собрать релевантный контекст для агента по запросу (поиск + форматирование).", + "inputSchema": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "Запрос контекста."}, + "top_k": {"type": "integer", "default": 5, "description": "Число результатов."}, + }, + "required": ["query"], + }, + }, +] + + +# -------------------------------------------------------------------------- +# JSON-RPC 2.0 (MCP 2024-11-05) +# -------------------------------------------------------------------------- +def handle(msg): + method = msg.get("method", "") + msg_id = msg.get("id") + + if method == "initialize": + return {"jsonrpc": "2.0", "id": msg_id, "result": { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}, + }} + + if method == "notifications/initialized": + return None + + if method in ("ping", "notifications/cancelled"): + return {"jsonrpc": "2.0", "id": msg_id, "result": {}} if msg_id is not None else None + + if method == "tools/list": + return {"jsonrpc": "2.0", "id": msg_id, "result": {"tools": TOOLS}} + + if method == "tools/call": + params = msg.get("params") or {} + tool = params.get("name", "") + args = params.get("arguments") or {} + handlers = { + "memory_search": _tool_search, + "memory_store": _tool_store, + "memory_ingest": _tool_ingest, + "memory_add_gap": _tool_add_gap, + "memory_list_gaps": _tool_list_gaps, + "memory_research": _tool_research, + "memory_stats": _tool_stats, + "memory_store_code": _tool_store_code, + "memory_search_code": _tool_search_code, + "memory_code_graph": _tool_code_graph, + "memory_infer": _tool_infer, + "memory_query_graph": _tool_query_graph, + "memory_verify": _tool_verify, + "memory_get_context": _tool_get_context, + } + fn = handlers.get(tool) + if fn is None: + return {"jsonrpc": "2.0", "id": msg_id, + "error": {"code": -32602, "message": f"Unknown tool: {tool}"}} + try: + text, failed = fn(args) + except Exception as exc: # сервер не должен падать от одного вызова + text = json.dumps({"error": f"{type(exc).__name__}: {exc}"}, ensure_ascii=False) + failed = True + return {"jsonrpc": "2.0", "id": msg_id, + "result": {"content": [{"type": "text", "text": text}], "isError": failed}} + + return {"jsonrpc": "2.0", "id": msg_id, + "error": {"code": -32601, "message": f"Method not found: {method}"}} + + +def main(): + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + msg = json.loads(line) + except json.JSONDecodeError: + response = {"jsonrpc": "2.0", "id": None, + "error": {"code": -32700, "message": "Parse error"}} + else: + try: + response = handle(msg) + except Exception as exc: + response = {"jsonrpc": "2.0", "id": msg.get("id"), + "error": {"code": -32000, "message": f"{type(exc).__name__}: {exc}"}} + if response is not None: + sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n") + sys.stdout.flush() + + +if __name__ == "__main__": + main() diff --git a/pegas_memory/__init__.py b/pegas_memory/__init__.py new file mode 100644 index 0000000..a18b06e --- /dev/null +++ b/pegas_memory/__init__.py @@ -0,0 +1 @@ +"""Pegas Memory — подсистема памяти Alicorn.""" diff --git a/pegas_memory/code_graph.py b/pegas_memory/code_graph.py new file mode 100644 index 0000000..5b3fb21 --- /dev/null +++ b/pegas_memory/code_graph.py @@ -0,0 +1,164 @@ +"""Граф зависимостей кода: calls/imports/inherits через tree-sitter. + +Строит ориентированный граф зависимостей между символами кода: + - calls: функция A вызывает функцию B + - imports: файл импортирует модуль (src = имя файла без расширения) + - inherits: класс B наследует класс A + +Извлекает из tree-sitter AST и сохраняет в pm_code_dependency (PostgreSQL). +""" +from __future__ import annotations + +from .code_symbols import _parser + +_FUNCTION_TYPES = ("function_definition", "function_item", "function_declaration", "method_definition") +_CLASS_TYPES = ("class_definition", "class_declaration") + + +def _sym_name(node, sb: bytes) -> str | None: + n = node.child_by_field_name("name") + return sb[n.start_byte:n.end_byte].decode("utf-8", "replace") if n else None + + +def _call_target(node, sb: bytes) -> str | None: + """Имя вызываемой функции из call-узла.""" + fn = node.child_by_field_name("function") + if fn is None: + return None + if fn.type == "identifier": + return sb[fn.start_byte:fn.end_byte].decode("utf-8", "replace") + if fn.type == "attribute": + parts = [sb[c.start_byte:c.end_byte].decode("utf-8", "replace") + for c in fn.children if c.type == "identifier"] + return parts[-1] if parts else None + return None + + +def _collect_calls(node, owner, sb, add_fn): + """Рекурсивно собрать вызовы внутри тела функции/метода.""" + if node.type == "call": + t = _call_target(node, sb) + if t: + add_fn(owner, t, "calls") + for ch in node.children: + _collect_calls(ch, owner, sb, add_fn) + + +def _collect_inherits(node, owner, sb, add_fn): + """Найти базовые классы (superclasses/heritage_clause).""" + for ch in node.children: + if ch.type in ("argument_list", "superclasses", "heritage_clause", "base_list"): + for sub in _iter(ch): + if sub.type == "identifier": + add_fn(owner, sb[sub.start_byte:sub.end_byte].decode("utf-8", "replace"), "inherits") + elif sub.type == "attribute": + parts = [sb[p.start_byte:p.end_byte].decode("utf-8", "replace") + for p in sub.children if p.type == "identifier"] + if parts: + add_fn(owner, parts[-1], "inherits") + + +def _collect_imports(root, module_name, sb, add_fn): + """Собрать импорты (import X / from Y import Z).""" + for node in _iter(root): + if node.type == "import_statement": + for ch in node.children: + if ch.type in ("dotted_name", "identifier"): + add_fn(module_name, sb[ch.start_byte:ch.end_byte].decode("utf-8", "replace"), "imports") + break + elif node.type == "import_from_statement": + mod = node.child_by_field_name("module_name") + if mod is not None: + target = sb[mod.start_byte:mod.end_byte].decode("utf-8", "replace") + # отфильтровать артефакты: "." (relative import), "..", пустые + if target not in (".", "..", ""): + add_fn(module_name, target, "imports") + + +def _iter(node): + yield node + for ch in node.children: + yield from _iter(ch) + + +def extract_dependencies(source: str, language: str, repo: str = "", path: str = "") -> list[dict]: + """Вернуть зависимости [{repo,path,language,src,dst,kind}] из исходника.""" + parser, _ = _parser(language) + sb = source.encode("utf-8") + tree = parser.parse(sb) + deps: list[dict] = [] + seen: set[tuple] = set() + + def add(src: str, dst: str, kind: str): + if not dst: + return + key = (src, dst, kind) + if key not in seen: + seen.add(key) + deps.append({"repo": repo, "path": path, "language": language, + "src": src, "dst": dst, "kind": kind}) + + module_name = path.rsplit("/", 1)[-1].rsplit(".", 1)[0] if path else "__module__" + + def walk(node): + t = node.type + if t in _FUNCTION_TYPES: + owner = _sym_name(node, sb) + body = node.child_by_field_name("body") + if body is not None and owner: + _collect_calls(body, owner, sb, add) + return # не спускаемся в вложенные scope как отдельные + if t in _CLASS_TYPES: + owner = _sym_name(node, sb) + if owner: + _collect_inherits(node, owner, sb, add) + body = node.child_by_field_name("body") + if body is not None: + for ch in body.children: + walk(ch) # методы класса — вложенные function_definition + return + for ch in node.children: + walk(ch) + + _collect_imports(tree.root_node, module_name, sb, add) + walk(tree.root_node) + return deps + + +def store_dependencies(conn, source: str, language: str, repo: str = "", path: str = "", + sync_memgraph: bool = True) -> int: + """Извлечь и сохранить зависимости кода (+ рёбра Memgraph). + + Возвращает число сохранённых. + """ + from . import memgraph_store + deps = extract_dependencies(source, language, repo, path) + n = 0 + with conn.cursor() as cur: + for d in deps: + cur.execute( + """INSERT INTO pm_code_dependency (repo, path, language, src, dst, kind) + VALUES (%s, %s, %s, %s, %s, %s) + ON CONFLICT (repo, path, src, dst, kind) DO NOTHING""", + (d["repo"], d["path"], d["language"], d["src"], d["dst"], d["kind"]), + ) + n += 1 + if sync_memgraph: + try: + memgraph_store.upsert_code_dependency( + d["src"], d["dst"], d["kind"], repo=d["repo"], path=d["path"], + ) + except Exception: + pass + conn.commit() + return n + + +def store_file_dependencies(conn, path: str, repo: str = "") -> int: + """По расширению файла определить язык и сохранить зависимости.""" + ext = path.rsplit(".", 1)[-1].lower() if "." in path else "" + lang = {".py": "python", ".js": "javascript", ".ts": "typescript", ".rs": "rust"}.get("." + ext) + if lang is None: + return 0 + source = open(path, encoding="utf-8").read() + return store_dependencies(conn, source, lang, repo, path) diff --git a/pegas_memory/code_symbols.py b/pegas_memory/code_symbols.py new file mode 100644 index 0000000..f98aa02 --- /dev/null +++ b/pegas_memory/code_symbols.py @@ -0,0 +1,108 @@ +"""Извлечение символов кода через tree-sitter (Фаза 3). + +Грамматики ставятся отдельными пакетами: tree-sitter-python, tree-sitter-javascript, +tree-sitter-rust и т.д. Каждый пакет экспортирует language(). +""" +from __future__ import annotations +import hashlib + +# Язык -> (имя модуля, признак function/class узлов) +_LANGS = { + "python": ("tree_sitter_python", {"function_definition", "class_definition"}), + "javascript": ("tree_sitter_javascript", {"function_declaration", "class_declaration", "method_definition"}), + "typescript": ("tree_sitter_javascript", {"function_declaration", "class_declaration", "method_definition"}), + "rust": ("tree_sitter_rust", {"function_item", "struct_item", "impl_item", "trait_item", "enum_item"}), +} + + +def _parser(language: str): + """Вернуть (Parser, kind_nodes) для языка.""" + from tree_sitter import Language, Parser + spec = _LANGS.get(language) + if spec is None: + raise ValueError(f"unsupported language: {language}") + mod_name, kind_nodes = spec + mod = __import__(mod_name) + lang = Language(mod.language()) + return Parser(lang), kind_nodes + + +def extract_symbols(source: str, language: str, repo: str = "", path: str = "") -> list[dict]: + """Вернуть символы (function/class/method/type/const) из исходника. + + tree-sitter индексирует байтами UTF-8, поэтому слайсим из bytes (src), + а не из str (source) — иначе кириллица в комментариях сбивает индексы. + """ + parser, kind_nodes = _parser(language) + src = source.encode("utf-8") + tree = parser.parse(src) + symbols: list[dict] = [] + + def walk(node): + if node.type in kind_nodes: + name_node = node.child_by_field_name("name") + name = src[name_node.start_byte:name_node.end_byte].decode("utf-8", "replace") if name_node else "?" + # сигнатура: от начала узла до первого тела/блока (первые 200 байт) + sig = src[node.start_byte:min(node.end_byte, node.start_byte + 200)].decode("utf-8", "replace") + if "function" in node.type or node.type in ("function_item", "function_declaration", "method_definition"): + kind = "function" + elif "class" in node.type or node.type in ("struct_item", "impl_item", "trait_item"): + kind = "class" + elif "enum" in node.type: + kind = "type" + else: + kind = "type" + symbols.append({ + "repo": repo, "path": path, "language": language, + "symbol_name": name, "kind": kind, + "signature": sig, + "body_hash": hashlib.sha256(src[node.start_byte:node.end_byte]).hexdigest(), + "line_start": node.start_point[0] + 1, "line_end": node.end_point[0] + 1, + }) + for child in node.children: + walk(child) + + walk(tree.root_node) + return symbols + + +def store_symbols(conn, source: str, language: str, repo: str = "", path: str = "", + sync_memgraph: bool = True) -> int: + """Извлечь и сохранить символы кода в pm_code_symbol (+ узлы Memgraph). + + Возвращает число сохранённых. + """ + from . import memgraph_store + syms = extract_symbols(source, language, repo, path) + n = 0 + with conn.cursor() as cur: + for s in syms: + cur.execute( + """INSERT INTO pm_code_symbol (repo, path, language, symbol_name, kind, + signature, body_hash, line_start, line_end) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT DO NOTHING""", + (s["repo"], s["path"], s["language"], s["symbol_name"], s["kind"], + s["signature"], s["body_hash"], s["line_start"], s["line_end"]), + ) + n += 1 + if sync_memgraph: + try: + memgraph_store.upsert_code_symbol( + s["symbol_name"], repo=s["repo"], path=s["path"], + language=s["language"], kind=s["kind"], + ) + except Exception: + pass + conn.commit() + return n + + +def store_file(conn, path: str, repo: str = "") -> int: + """По расширению файла определить язык и сохранить символы.""" + ext = path.rsplit(".", 1)[-1].lower() if "." in path else "" + lang = {".py": "python", ".js": "javascript", ".ts": "typescript", ".rs": "rust"}.get("." + ext) + if lang is None: + return 0 + source = open(path, encoding="utf-8").read() + return store_symbols(conn, source, lang, repo, path) diff --git a/pegas_memory/config.py b/pegas_memory/config.py new file mode 100644 index 0000000..defd173 --- /dev/null +++ b/pegas_memory/config.py @@ -0,0 +1,23 @@ +"""Pegas Memory — конфигурация (адреса сервисов).""" +import os + +PG_DSN = os.getenv("PG_DSN", "postgresql://atom:atompass123@127.0.0.1:5434/content_atom") +QDRANT_URL = os.getenv("QDRANT_URL", "http://127.0.0.1:6333") +MEMGRAPH_HOST = os.getenv("MEMGRAPH_HOST", "127.0.0.1") +MEMGRAPH_PORT = int(os.getenv("MEMGRAPH_PORT", "7687")) +OLLAMA_URL = os.getenv("OLLAMA_URL", "http://127.0.0.1:11434") + +EMBED_MODEL = os.getenv("EMBED_MODEL", "embeddinggemma-2-740m-Q8_0:latest") +EMBED_DIM = int(os.getenv("EMBED_DIM", "768")) + +LLM_MODEL = os.getenv("LLM_MODEL", "lfm25-24b-a2b:128k") + +QDRANT_COLLECTION = os.getenv("QDRANT_COLLECTION", "alicorn_atoms") + +ATOM_TYPES = ("text", "definition", "logical_rule", "code", "table", "fact") +ATOM_STATUSES = ("candidate", "verified", "stale", "contested") + +DEDUP_THRESHOLD = float(os.getenv("DEDUP_THRESHOLD", "0.90")) # порог семантической дедупликации + +PROXY_URL = os.getenv("PROXY_URL", "") # опционально: socks5://127.0.0.1:1080 или http://... +USER_AGENT = os.getenv("USER_AGENT", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36") diff --git a/pegas_memory/db.py b/pegas_memory/db.py new file mode 100644 index 0000000..e31f91c --- /dev/null +++ b/pegas_memory/db.py @@ -0,0 +1,85 @@ +"""PostgreSQL — источник истины (psycopg).""" +from __future__ import annotations +import hashlib +import json +import psycopg +from psycopg.rows import dict_row +from . import config + +def connect(): + return psycopg.connect(config.PG_DSN, row_factory=dict_row) + +def _as_text(content) -> str: + """Привести content к строке: dict/list -> JSON, иначе str.""" + if isinstance(content, str): + return content + import json as _json + return _json.dumps(content, ensure_ascii=False, sort_keys=True) + +def content_hash(content) -> str: + return hashlib.sha256(_as_text(content).encode("utf-8")).hexdigest() + +def upsert_document(conn, *, kind: str, uri: str | None, title: str | None, + content_hash: str | None = None, authority: float = 0.5, + meta: dict | None = None) -> int: + """Вставить/вернуть документ по uri.""" + with conn.cursor() as cur: + if uri: + cur.execute("SELECT id FROM pm_document WHERE uri = %s", (uri,)) + row = cur.fetchone() + if row: + return row["id"] + cur.execute( + """INSERT INTO pm_document (kind, uri, title, content_hash, authority, meta) + VALUES (%s, %s, %s, %s, %s, %s) + ON CONFLICT (uri) DO UPDATE SET title = EXCLUDED.title + RETURNING id""", + (kind, uri, title, content_hash, authority, json.dumps(meta or {}, ensure_ascii=False)), + ) + return cur.fetchone()["id"] + +def upsert_atom(conn, *, type: str, claim: str | None, content, + confidence: float = 0.5, status: str = "candidate", + doc_id: int | None = None, locator: dict | None = None, + metadata: dict | None = None) -> tuple[int, bool]: + """Вставить атом по content_hash. Возвращает (id, created). + + content может быть str (text/fact/definition/code) или dict/list + (logical_rule/table) — dict сериализуется в JSON. + """ + content = _as_text(content) + h = content_hash(content) + with conn.cursor() as cur: + cur.execute("SELECT id FROM pm_atom WHERE content_hash = %s", (h,)) + row = cur.fetchone() + if row: + return row["id"], False + cur.execute( + """INSERT INTO pm_atom (type, claim, content, content_hash, confidence, + status, doc_id, locator, metadata) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + RETURNING id""", + (type, claim, content, h, confidence, status, doc_id, + json.dumps(locator or {}, ensure_ascii=False), + json.dumps(metadata or {}, ensure_ascii=False)), + ) + return cur.fetchone()["id"], True + +def upsert_link(conn, *, src: int, dst: int, kind: str, confidence: float = 0.5) -> int: + with conn.cursor() as cur: + cur.execute( + """INSERT INTO pm_atom_link (src, dst, kind, confidence) + VALUES (%s, %s, %s, %s) + ON CONFLICT (src, dst, kind) DO NOTHING + RETURNING id""", + (src, dst, kind, confidence), + ) + row = cur.fetchone() + return row["id"] if row else 0 + +def enqueue_outbox(conn, *, entity_type: str, entity_id: int, op: str = "upsert"): + with conn.cursor() as cur: + cur.execute( + "INSERT INTO pm_outbox (entity_type, entity_id, op) VALUES (%s, %s, %s)", + (entity_type, entity_id, op), + ) diff --git a/pegas_memory/dedup.py b/pegas_memory/dedup.py new file mode 100644 index 0000000..9f53ea0 --- /dev/null +++ b/pegas_memory/dedup.py @@ -0,0 +1,34 @@ +"""Семантическая дедупликация атомов по эмбеддингам. + +Наивный дедуп по content_hash ловит только точные повторы. LLM генерирует +вариации одного факта — их ловим по косинусной близости эмбеддингов в Qdrant. +""" +from __future__ import annotations + +from . import config +DEFAULT_THRESHOLD = float(getattr(config, "DEDUP_THRESHOLD", "0.90")) + + +def find_duplicate(text: str, threshold: float = DEFAULT_THRESHOLD, top_k: int = 3) -> int | None: + """Найти существующий атом, близкий к text (косинус > threshold). + + Возвращает atom_id дубликата или None. + """ + from . import qdrant_store + hits = qdrant_store.search(text, top_k=top_k) + if not hits: + return None + best = hits[0] + if best["score"] >= threshold: + return int(best["id"]) + return None + + +def duplicate_score(text: str, top_k: int = 3) -> tuple[int | None, float]: + """Вернуть (atom_id, score) лучшего совпадения (id=None если ниже порога).""" + from . import qdrant_store + hits = qdrant_store.search(text, top_k=top_k) + if not hits: + return None, 0.0 + best = hits[0] + return int(best["id"]), best["score"] diff --git a/pegas_memory/distill.py b/pegas_memory/distill.py new file mode 100644 index 0000000..34438a4 --- /dev/null +++ b/pegas_memory/distill.py @@ -0,0 +1,72 @@ +"""Дистилляция чанка в типизированные атомы (LLM).""" +from __future__ import annotations +import json +import httpx +from . import config + +_client = httpx.Client(timeout=600) + +SYSTEM = """Ты — экстрактор знаний. Преврати текст в атомарные, типизированные утверждения. +Верни ТОЛЬКО JSON-массив без пояснений. Каждый элемент: +{ + "type": "text|definition|logical_rule|code|table|fact", + "claim": "короткое каноническое утверждение (одно предложение)", + "content": "полное значение атома", + "confidence": 0.0-1.0 +} +Правила: +- definition: термин и его точное определение. +- logical_rule: правило "если X то Y" (content = JSON со ключами if/then). +- code: фрагмент кода или сигнатура функции (content = сам код). +- table: табличные данные (content = JSON или markdown-таблица). +- fact: проверяемый факт с числом или датой. +- text: всё остальное ценное. +- Отбрасывай воду, повторы, навигацию, рекламу. Только новое и ценное. +- Если ничего ценного — верни []. No markdown fences.""" + +def distill_chunk(chunk_text: str, model: str | None = None) -> list[dict]: + """Вернуть список атомов (dict) из чанка.""" + m = model or config.LLM_MODEL + r = _client.post(f"{config.OLLAMA_URL}/api/chat", json={ + "model": m, + "messages": [ + {"role": "system", "content": SYSTEM}, + {"role": "user", "content": chunk_text[:6000]}, + ], + "stream": False, + "options": {"temperature": 0.1}, + }) + r.raise_for_status() + content = r.json()["message"]["content"] + start = content.find("[") + end = content.rfind("]") + if start == -1 or end == -1: + return [] + try: + arr = json.loads(content[start:end + 1]) + except json.JSONDecodeError: + return [] + return [a for a in arr if isinstance(a, dict) and a.get("type") and a.get("content")] + +def verify_atom(atom: dict, context_atoms: list[dict], model: str | None = None) -> float: + """Оценить новизну/ценность атома против существующих (0..1).""" + m = model or config.LLM_MODEL + ctx = "\n".join(f"- {a.get('claim','')}" for a in context_atoms[:8]) + prompt = ( + f"Существующие знания:\n{ctx}\n\n" + f"Новый атом ({atom.get('type')}): {atom.get('claim')}\n{atom.get('content','')[:500]}\n\n" + f"Оцени от 0 до 1, насколько это НОВОЕ и ценное знание (0=дубль/вода, 1=уникальное важное). " + f"Верни только число." + ) + r = _client.post(f"{config.OLLAMA_URL}/api/chat", json={ + "model": m, + "messages": [{"role": "user", "content": prompt}], + "stream": False, + "options": {"temperature": 0.0}, + }) + r.raise_for_status() + out = r.json()["message"]["content"].strip() + try: + return float(out) + except ValueError: + return 0.5 diff --git a/pegas_memory/embedding.py b/pegas_memory/embedding.py new file mode 100644 index 0000000..c6551bf --- /dev/null +++ b/pegas_memory/embedding.py @@ -0,0 +1,27 @@ +"""EmbeddingGemma 2 через Ollama (dim=768).""" +from __future__ import annotations +import httpx +from . import config + +_client = httpx.Client(timeout=120) + +def embed(text: str) -> list[float]: + """Вернуть вектор для одного текста.""" + r = _client.post(f"{config.OLLAMA_URL}/api/embed", json={ + "model": config.EMBED_MODEL, + "input": text, + }) + r.raise_for_status() + data = r.json() + vec = data["embeddings"][0] + assert len(vec) == config.EMBED_DIM, f"unexpected dim {len(vec)}" + return vec + +def embed_batch(texts: list[str]) -> list[list[float]]: + """Вернуть векторы для пачки текстов (один запрос).""" + r = _client.post(f"{config.OLLAMA_URL}/api/embed", json={ + "model": config.EMBED_MODEL, + "input": texts, + }) + r.raise_for_status() + return r.json()["embeddings"] diff --git a/pegas_memory/inference.py b/pegas_memory/inference.py new file mode 100644 index 0000000..84ba4a8 --- /dev/null +++ b/pegas_memory/inference.py @@ -0,0 +1,104 @@ +"""Логическая инференция: forward-chaining правил в Memgraph. + +Правило — атом типа logical_rule, content = JSON: + {"if": "imports", "then": "depends_on"} # одношаговое + {"if": ["depends_on", "depends_on"], "then": "depends_on"} # транзитивное (композиция) + +Инференция итерирует до фиксированной точки: пока правила порождают новые рёбра. +Выведенные рёбра помечаются inferred=true, source='inference'. +""" +from __future__ import annotations +import json + +from . import memgraph_store + + +def _is_kind(token) -> bool: + """kind ребра — одиночный токен без пробелов (imports, depends_on, calls...). + + Отсекает человекочитаемые правила от LLM-дистилляции вида + if='Memgraph хранит граф в RAM' — это не kind ребра. + """ + return isinstance(token, str) and " " not in token and len(token) > 0 and len(token) < 64 + + +def load_rules(conn) -> list[dict]: + """Загрузить машинные правила из pm_atom (type='logical_rule'). + + Правило валидно для инференции, только если if/then — kind-рёбра + (одиночные токены без пробелов). Человекочитаемые правила от + дистилляции отфильтровываются. + """ + with conn.cursor() as cur: + cur.execute( + "SELECT id, claim, content FROM pm_atom WHERE type = 'logical_rule' ORDER BY id" + ) + rows = cur.fetchall() + rules = [] + for r in rows: + try: + spec = json.loads(r["content"]) if isinstance(r["content"], str) else r["content"] + except (json.JSONDecodeError, TypeError): + continue + if "if" not in spec or "then" not in spec: + continue + if_ = spec["if"] + then = spec["then"] + # валидация: then — всегда kind; if — kind или список kind + if not _is_kind(then): + continue + if isinstance(if_, list): + if len(if_) != 2 or not all(_is_kind(k) for k in if_): + continue + elif not _is_kind(if_): + continue + rules.append({"atom_id": r["id"], "claim": r["claim"], "spec": spec}) + return rules + + +def _apply_rule(spec: dict) -> int: + """Применить одно правило. Возвращает число созданных рёбер.""" + if_ = spec["if"] + then = spec["then"] + if isinstance(if_, list): + # композиция: [k1, k2] -> (a)-[k1]->(b)-[k2]->(c) ⇒ (a)-[then]->(c) + if len(if_) == 2: + return memgraph_store.apply_rule_chain(if_[0], if_[1], then) + return 0 + # одношаговое: (a)-[if]->(b) ⇒ (a)-[then]->(b) + return memgraph_store.apply_rule_one_step(if_, then) + + +def infer(conn, max_rounds: int = 10) -> dict: + """Запустить инференцию до фиксированной точки. + + Возвращает {rules, rounds, edges_created}. + """ + rules = load_rules(conn) + total = 0 + rounds = 0 + for _ in range(max_rounds): + created = 0 + for rule in rules: + created += _apply_rule(rule["spec"]) + total += created + rounds += 1 + if created == 0: + break + return {"rules": len(rules), "rounds": rounds, "edges_created": total} + + +def store_rule(conn, *, claim: str, if_: str | list, then: str) -> int: + """Сохранить логическое правило (атом logical_rule) в PG. + + Возвращает id атома. Если правило с тем же content уже есть — вернёт существующий. + """ + from . import db + content = json.dumps({"if": if_, "then": then}, ensure_ascii=False) + atom_id, created = db.upsert_atom( + conn, type="logical_rule", claim=claim, content=content, + confidence=0.9, status="verified", + ) + if created: + db.enqueue_outbox(conn, entity_type="atom", entity_id=atom_id) + return atom_id diff --git a/pegas_memory/ingest.py b/pegas_memory/ingest.py new file mode 100644 index 0000000..1e210d8 --- /dev/null +++ b/pegas_memory/ingest.py @@ -0,0 +1,84 @@ +"""Парсеры источников: text/pdf/epub/web → чанки.""" +from __future__ import annotations +import hashlib +from dataclasses import dataclass, field + +@dataclass +class ParsedChunk: + seq: int + content: str + locator: dict = field(default_factory=dict) + +@dataclass +class ParsedDoc: + kind: str + uri: str | None + title: str | None + chunks: list[ParsedChunk] + content_hash: str = "" + +def parse_text(text: str, *, chunk_size: int = 2000, overlap: int = 200, + title: str | None = None, uri: str | None = None) -> ParsedDoc: + """Разбить сырой текст на чанки с перекрытием.""" + text = text.strip() + chunks: list[ParsedChunk] = [] + start = 0 + seq = 0 + while start < len(text): + end = min(start + chunk_size, len(text)) + # не резать посреди слова + if end < len(text): + space = text.rfind(" ", start, end) + if space > start + chunk_size // 2: + end = space + chunks.append(ParsedChunk(seq=seq, content=text[start:end].strip())) + seq += 1 + if end >= len(text): + break + start = end - overlap + return ParsedDoc(kind="text", uri=uri, title=title, chunks=chunks, + content_hash=hashlib.sha256(text.encode()).hexdigest()) + +def parse_pdf(path: str) -> ParsedDoc: + from pypdf import PdfReader + reader = PdfReader(path) + text = "\n\n".join((page.extract_text() or "") for page in reader.pages) + doc = parse_text(text, title=path.split("/")[-1], uri=path) + doc.kind = "pdf" + return doc + +def parse_epub(path: str) -> ParsedDoc: + import ebooklib + from ebooklib import epub + book = epub.read_epub(path) + parts = [] + for item in book.get_items_of_type(ebooklib.ITEM_DOCUMENT): + parts.append(item.get_content().decode("utf-8", errors="ignore")) + text = "\n\n".join(parts) + doc = parse_text(text, title=path.split("/")[-1], uri=path) + doc.kind = "epub" + return doc + +def _http_client(): + """httpx.Client с опциональным прокси и единым User-Agent.""" + import httpx + from . import config + kwargs = dict(timeout=30, follow_redirects=True, + headers={"User-Agent": config.USER_AGENT}) + if config.PROXY_URL: + kwargs["proxy"] = config.PROXY_URL + return httpx.Client(**kwargs) + +def parse_web(url: str) -> ParsedDoc: + import trafilatura + # fetch через httpx с таймаутом + прокси (trafilatura.fetch_url не имеет таймаута и виснет) + with _http_client() as client: + r = client.get(url) + r.raise_for_status() + text = trafilatura.extract(r.text, include_comments=False, include_tables=True, + url=url) + if not text: + raise ValueError(f"trafilatura: no main content for {url}") + doc = parse_text(text, uri=url, title=url) + doc.kind = "web" + return doc diff --git a/pegas_memory/memgraph_store.py b/pegas_memory/memgraph_store.py new file mode 100644 index 0000000..05042d1 --- /dev/null +++ b/pegas_memory/memgraph_store.py @@ -0,0 +1,136 @@ +"""Memgraph — графовый срез (узлы Atom + CodeSymbol, рёбра RELATES). + +Memgraph совместим с Bolt-протоколом Neo4j — используем официальный neo4j драйвер. +Рёбра единообразны: (любой узел)-[:RELATES {kind, inferred, source}]->(любой узел). +Это позволяет логической инференции работать и над атомами, и над кодом. +""" +from __future__ import annotations +from neo4j import GraphDatabase +from . import config + +_driver = None + +def driver(): + global _driver + if _driver is None: + _driver = GraphDatabase.driver( + f"bolt://{config.MEMGRAPH_HOST}:{config.MEMGRAPH_PORT}" + ) + return _driver + +def upsert_atom(atom_id: int, type: str, claim: str | None, status: str) -> None: + with driver().session() as s: + s.run( + "MERGE (a:Atom {id: $id}) SET a.type = $type, a.claim = $claim, a.status = $status", + id=atom_id, type=type, claim=claim or "", status=status, + ) + +def upsert_link(src: int, dst: int, kind: str, *, inferred: bool = False, + source: str | None = None, confidence: float = 0.5) -> None: + with driver().session() as s: + s.run( + "MATCH (a:Atom {id: $src}), (b:Atom {id: $dst}) " + "MERGE (a)-[r:RELATES {kind: $kind}]->(b) " + "SET r.inferred = $inferred, r.source = $source, r.confidence = $confidence", + src=src, dst=dst, kind=kind, inferred=inferred, source=source, confidence=confidence, + ) + +def delete_atom(atom_id: int) -> None: + with driver().session() as s: + s.run("MATCH (a:Atom {id: $id}) DETACH DELETE a", id=atom_id) + +def upsert_code_symbol(name: str, *, repo: str = "", path: str = "", + language: str = "", kind: str = "") -> None: + """Узел CodeSymbol (дедуп по name+repo+path).""" + with driver().session() as s: + s.run( + "MERGE (c:CodeSymbol {name: $name, repo: $repo, path: $path}) " + "SET c.language = $language, c.kind = $kind", + name=name, repo=repo, path=path, language=language, kind=kind, + ) + +def upsert_code_dependency(src: str, dst: str, kind: str, *, repo: str = "", path: str = "") -> None: + """Ребро зависимости кода: (src)-[:RELATES {kind}]->(dst).""" + with driver().session() as s: + # dst может быть внешним (builtin) — создаём как CodeSymbol без repo/path + s.run( + "MERGE (a:CodeSymbol {name: $src, repo: $repo, path: $path}) " + "MERGE (b:CodeSymbol {name: $dst}) " + "MERGE (a)-[r:RELATES {kind: $kind}]->(b) " + "SET r.source = 'extraction'", + src=src, dst=dst, kind=kind, repo=repo, path=path, + ) + +def apply_rule_one_step(if_kind: str, then_kind: str) -> int: + """Одношаговое правило: (a)-[:RELATES {kind: if}]->(b) ⇒ (a)-[:RELATES {kind: then}]->(b). + + Возвращает число созданных рёбер. + """ + with driver().session() as s: + result = s.run( + "MATCH (a)-[r:RELATES {kind: $if_kind}]->(b) " + "OPTIONAL MATCH (a)-[x:RELATES {kind: $then_kind}]->(b) " + "WITH a, b, x WHERE x IS NULL " + "CREATE (a)-[r2:RELATES {kind: $then_kind, inferred: true, source: 'inference'}]->(b) " + "RETURN count(r2) AS n", + if_kind=if_kind, then_kind=then_kind, + ) + return result.single()["n"] + +def apply_rule_chain(k1: str, k2: str, then_kind: str) -> int: + """Композиция: (a)-[:RELATES {kind: k1}]->(b)-[:RELATES {kind: k2}]->(c) + ⇒ (a)-[:RELATES {kind: then}]->(c). Транзитивность. Возвращает число созданных.""" + with driver().session() as s: + result = s.run( + "MATCH (a)-[:RELATES {kind: $k1}]->(b)-[:RELATES {kind: $k2}]->(c) " + "OPTIONAL MATCH (a)-[x:RELATES {kind: $then_kind}]->(c) " + "WITH a, b, c, x WHERE x IS NULL " + "CREATE (a)-[r2:RELATES {kind: $then_kind, inferred: true, source: 'inference'}]->(c) " + "RETURN count(r2) AS n", + k1=k1, k2=k2, then_kind=then_kind, + ) + return result.single()["n"] + +def count() -> int: + with driver().session() as s: + return s.run("MATCH (a:Atom) RETURN count(a) AS n").single()["n"] + +def count_edges() -> int: + with driver().session() as s: + return s.run("MATCH ()-[r:RELATES]->() RETURN count(r) AS n").single()["n"] + +def count_code_symbols() -> int: + with driver().session() as s: + return s.run("MATCH (c:CodeSymbol) RETURN count(c) AS n").single()["n"] + +def query_edges(kind: str | None = None, limit: int = 50) -> list[dict]: + """Список рёбер (опционально по kind).""" + with driver().session() as s: + if kind: + r = s.run( + "MATCH (a)-[r:RELATES {kind: $kind}]->(b) " + "RETURN a.name AS src, labels(a) AS al, b.name AS dst, labels(b) AS bl, " + "r.kind AS kind, r.inferred AS inferred LIMIT $lim", + kind=kind, lim=limit, + ) + else: + r = s.run( + "MATCH (a)-[r:RELATES]->(b) " + "RETURN a.name AS src, labels(a) AS al, b.name AS dst, labels(b) AS bl, " + "r.kind AS kind, r.inferred AS inferred LIMIT $lim", + lim=limit, + ) + out = [] + for rec in r: + d = {"src": rec["src"], "dst": rec["dst"]} + if "al" in rec and rec["al"]: + d["src_label"] = rec["al"][0] + if "bl" in rec and rec["bl"]: + d["dst_label"] = rec["bl"][0] + for key in ("kind", "inferred"): + try: + d[key] = rec[key] + except KeyError: + pass + out.append(d) + return out diff --git a/pegas_memory/outbox_sync.py b/pegas_memory/outbox_sync.py new file mode 100644 index 0000000..3ea6423 --- /dev/null +++ b/pegas_memory/outbox_sync.py @@ -0,0 +1,69 @@ +"""Outbox-синк: PG → Qdrant + Memgraph.""" +from __future__ import annotations +import json +from . import config, db, qdrant_store, memgraph_store + +def sync_batch(conn, limit: int = 100) -> dict: + """Обработать пачку необработанных outbox-записей.""" + with conn.cursor() as cur: + cur.execute( + "SELECT id, entity_type, entity_id, op FROM pm_outbox " + "WHERE processed = false ORDER BY id LIMIT %s FOR UPDATE SKIP LOCKED", + (limit,), + ) + rows = cur.fetchall() + + stats = {"atom": 0, "link": 0, "error": 0} + for row in rows: + try: + if row["entity_type"] == "atom": + _sync_atom(conn, row) + stats["atom"] += 1 + elif row["entity_type"] == "link": + _sync_link(conn, row) + stats["link"] += 1 + with conn.cursor() as cur: + cur.execute("UPDATE pm_outbox SET processed = true WHERE id = %s", (row["id"],)) + except Exception as e: + stats["error"] += 1 + print(f"[outbox] error id={row['id']}: {e}") + conn.commit() + return stats + +def _sync_atom(conn, row) -> None: + with conn.cursor() as cur: + if row["op"] == "delete": + qdrant_store.delete_atom(row["entity_id"]) + return + cur.execute("SELECT id, type, claim, content, confidence, status, doc_id FROM pm_atom WHERE id = %s", (row["entity_id"],)) + a = cur.fetchone() + if not a: + return + # 1) узел в графе (Memgraph) + memgraph_store.upsert_atom(a["id"], a["type"], a["claim"], a["status"]) + # 2) вектор в Qdrant (эмбеддим claim+content — claim семантически сфокусирован) + embed_text = (a["claim"] or "") + " " + a["content"][:1500] + payload = { + "type": a["type"], "claim": a["claim"] or "", + "confidence": a["confidence"], "status": a["status"], + "doc_id": a["doc_id"], + } + qdrant_store.upsert_atom(a["id"], embed_text, payload) + +def _sync_link(conn, row) -> None: + with conn.cursor() as cur: + cur.execute("SELECT src, dst, kind FROM pm_atom_link WHERE id = %s", (row["entity_id"],)) + l = cur.fetchone() + if not l: + return + memgraph_store.upsert_link(l["src"], l["dst"], l["kind"]) + +def sync_all(conn, batch: int = 100, max_batches: int = 100) -> dict: + total = {"atom": 0, "link": 0, "error": 0} + for _ in range(max_batches): + s = sync_batch(conn, batch) + for k in total: + total[k] += s[k] + if s["atom"] + s["link"] + s["error"] == 0: + break + return total diff --git a/pegas_memory/pipeline.py b/pegas_memory/pipeline.py new file mode 100644 index 0000000..09686f5 --- /dev/null +++ b/pegas_memory/pipeline.py @@ -0,0 +1,83 @@ +"""Ингест-пайплайн: источник → чанки → атомы → PG + outbox.""" +from __future__ import annotations +import psycopg +from . import db, distill, ingest +from .ingest import ParsedDoc + +def ingest_doc(conn, doc: ParsedDoc, *, verify: bool = False, + min_confidence: float = 0.4, dedup: bool = True, + dedup_threshold: float = 0.93) -> dict: + """Прогнать документ через пайплайн. Возвращает статистику. + + dedup: семантическая дедупликация по эмбеддингам (Qdrant) против + существующих атомов — вариации одного факта от LLM не дублируются. + """ + doc_id = db.upsert_document( + conn, kind=doc.kind, uri=doc.uri, title=doc.title, + content_hash=doc.content_hash, + ) + stats = {"doc_id": doc_id, "chunks": 0, "atoms_created": 0, + "atoms_skipped": 0, "duplicates": 0} + + with conn.cursor() as cur: + for ch in doc.chunks: + cur.execute( + "INSERT INTO pm_chunk (doc_id, seq, content, locator) VALUES (%s, %s, %s, %s) RETURNING id", + (doc_id, ch.seq, ch.content, __import__("json").dumps(ch.locator, ensure_ascii=False)), + ) + stats["chunks"] += 1 + + # дистилляция чанка в атомы + atoms = distill.distill_chunk(ch.content) + for a in atoms: + if a.get("confidence", 0.5) < min_confidence: + stats["atoms_skipped"] += 1 + continue + # семантическая дедупликация + if dedup: + from . import dedup as _dedup + dup_id = _dedup.find_duplicate( + (a.get("claim") or "") + " " + str(a["content"])[:800], + threshold=dedup_threshold, + ) + if dup_id is not None: + stats["duplicates"] += 1 + stats["atoms_skipped"] += 1 + continue + atom_id, created = db.upsert_atom( + conn, + type=a["type"], + claim=a.get("claim"), + content=a["content"], + confidence=a.get("confidence", 0.5), + doc_id=doc_id, + locator=ch.locator, + metadata=a.get("metadata") or a.get("content_json"), + ) + if created: + stats["atoms_created"] += 1 + db.enqueue_outbox(conn, entity_type="atom", entity_id=atom_id) + else: + stats["atoms_skipped"] += 1 + conn.commit() + return stats + +def ingest_text(conn, text: str, *, title: str | None = None, uri: str | None = None, + **kw) -> dict: + doc = ingest.parse_text(text, title=title, uri=uri) + return ingest_doc(conn, doc, **kw) + +def ingest_file(conn, path: str, **kw) -> dict: + low = path.lower() + if low.endswith(".pdf"): + doc = ingest.parse_pdf(path) + elif low.endswith(".epub"): + doc = ingest.parse_epub(path) + else: + doc = ingest.parse_text(open(path, encoding="utf-8").read(), + title=path.split("/")[-1], uri=path) + return ingest_doc(conn, doc, **kw) + +def ingest_url(conn, url: str, **kw) -> dict: + doc = ingest.parse_web(url) + return ingest_doc(conn, doc, **kw) diff --git a/pegas_memory/qdrant_store.py b/pegas_memory/qdrant_store.py new file mode 100644 index 0000000..8e00e54 --- /dev/null +++ b/pegas_memory/qdrant_store.py @@ -0,0 +1,57 @@ +"""Qdrant — векторный срез (EmbeddingGemma 2, dim=768).""" +from __future__ import annotations +from qdrant_client import QdrantClient +from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue +from . import config +from .embedding import embed + +_client: QdrantClient | None = None + +def client() -> QdrantClient: + global _client + if _client is None: + _client = QdrantClient(url=config.QDRANT_URL) + return _client + +def ensure_collection(): + c = client() + try: + c.get_collection(config.QDRANT_COLLECTION) + except Exception: + c.create_collection( + collection_name=config.QDRANT_COLLECTION, + vectors_config=VectorParams(size=config.EMBED_DIM, distance=Distance.COSINE), + ) + +def upsert_atom(atom_id: int, text: str, payload: dict) -> None: + """text — что эмбеддим; payload — type/confidence/status/doc_id/claim.""" + vec = embed(text) + c = client() + c.upsert( + collection_name=config.QDRANT_COLLECTION, + points=[PointStruct(id=atom_id, vector=vec, payload=payload)], + ) + +def delete_atom(atom_id: int) -> None: + client().delete( + collection_name=config.QDRANT_COLLECTION, + points_selector=[atom_id], + ) + +def search(query: str, top_k: int = 5, type_filter: str | None = None, + status_filter: str | None = None) -> list[dict]: + """Семантический поиск по атомам.""" + qvec = embed(query) + must = [] + if type_filter: + must.append(FieldCondition(key="type", match=MatchValue(value=type_filter))) + if status_filter: + must.append(FieldCondition(key="status", match=MatchValue(value=status_filter))) + qfilter = Filter(must=must) if must else None + hits = client().query_points( + collection_name=config.QDRANT_COLLECTION, + query=qvec, + limit=top_k, + query_filter=qfilter, + ).points + return [{"id": h.id, "score": h.score, **h.payload} for h in hits] diff --git a/pegas_memory/research.py b/pegas_memory/research.py new file mode 100644 index 0000000..78a22d8 --- /dev/null +++ b/pegas_memory/research.py @@ -0,0 +1,84 @@ +"""Цикл исследования пробелов: detect → plan → acquire → extract → verify → commit.""" +from __future__ import annotations +import json +import httpx +from . import config, db, pipeline + +_client = httpx.Client(timeout=600) + +def detect_gaps(conn) -> list[dict]: + """Найти пробелы в базе знаний.""" + with conn.cursor() as cur: + cur.execute( + "SELECT id, question, reason, priority, status FROM pm_research_gap " + "WHERE status = 'open' ORDER BY priority DESC, id LIMIT 20" + ) + return cur.fetchall() + +def add_gap(conn, question: str, reason: str = "user_query", priority: float = 0.5): + with conn.cursor() as cur: + cur.execute( + "INSERT INTO pm_research_gap (question, reason, priority) VALUES (%s, %s, %s) RETURNING id", + (question, reason, priority), + ) + conn.commit() + +def web_search(query: str, engine: str = "searxng") -> list[dict]: + """Поиск через SearXNG. Возвращает [{title, url, snippet}].""" + # локальный SearXNG (MCP mcp__searxng__search); здесь — прямой HTTP JSON API + import urllib.parse + q = urllib.parse.quote(query) + # типичный endpoint локального SearXNG + for base in ("http://127.0.0.1:8889", "http://127.0.0.1:8080"): + try: + r = _client.get(f"{base}/search", params={"q": query, "format": "json"}, timeout=15) + r.raise_for_status() + results = r.json().get("results", []) + return [{"title": x.get("title"), "url": x.get("url"), "snippet": x.get("content")} + for x in results[:3]] + except Exception: + continue + return [] + +def research_question(conn, question: str, *, top_k: int = 5, model: str | None = None) -> dict: + """Полный цикл исследования одного вопроса. + + 1. web-поиск (SearXNG); + 2. для каждого источника: полная загрузка URL; при недоступности (гео-блок, + нет прокси) — дистилляция сниппета из поиска; + 3. пометка вопроса resolved. + """ + result = {"question": question, "sources": [], "atoms_created": 0} + sources = web_search(question) + result["sources"] = [s["url"] for s in sources if s.get("url")] + + for s in sources: + url = s.get("url") + if not url: + continue + # попытка полной загрузки + ingested = False + try: + stats = pipeline.ingest_url(conn, url, min_confidence=0.4) + result["atoms_created"] += stats["atoms_created"] + ingested = True + except Exception as e: + print(f"[research] ingest {url} failed: {e}") + # fallback: сниппет из поиска (когда сайт недоступен напрямую) + if not ingested and s.get("snippet"): + try: + title = s.get("title") or url + stats = pipeline.ingest_text( + conn, s["snippet"], title=title, uri=url, min_confidence=0.4, + ) + result["atoms_created"] += stats["atoms_created"] + except Exception as e: + print(f"[research] snippet fallback {url} failed: {e}") + + with conn.cursor() as cur: + cur.execute( + "UPDATE pm_research_gap SET status = 'resolved' WHERE question = %s AND status = 'open'", + (question,), + ) + conn.commit() + return result diff --git a/pegas_memory/verify.py b/pegas_memory/verify.py new file mode 100644 index 0000000..f0c6275 --- /dev/null +++ b/pegas_memory/verify.py @@ -0,0 +1,158 @@ +"""Верификация атомов: self-critique + кросс-подтверждение. + +Двухступенчатая верификация знания: + 1. self-critique (LLM) — вердикт по новизне/противоречию против существующих. + 2. кросс-подтверждение (эмбеддинги) — тот же факт из N независимых источников. + +Статусы: + candidate — новый атом, не проверен + verified — подтверждён (self-critique прошёл И/ИЛИ 2+ источника) + contested — противоречит существующему знанию + stale — устарел (не актуально) + +Верификация не удаляет атомы, только меняет статус/confidence. +""" +from __future__ import annotations +import json +import httpx +from . import config, db + +_client = httpx.Client(timeout=600) + +VERDICTS = ("new", "duplicate", "corroborates", "contradicts") + + +def self_critique(atom: dict, context_atoms: list[dict], model: str | None = None) -> dict: + """LLM-вердикт по атому против существующих знаний. + + Возвращает {"verdict", "confidence", "reason"}. + """ + m = model or config.LLM_MODEL + ctx = "\n".join(f"- {a.get('claim','')[:120]}" for a in context_atoms[:8]) + prompt = ( + f"Существующие знания:\n{ctx}\n\n" + f"Новый атом ({atom.get('type')}):\n{atom.get('claim')}\n{atom.get('content','')[:400]}\n\n" + f"Оцени атом относительно существующих знаний. Верни ТОЛЬКО JSON:\n" + f'{{"verdict": "new|duplicate|corroborates|contradicts", "confidence": 0.0-1.0, "reason": "кратко"}}\n' + f"- new: уникальное знание, которого нет в контексте\n" + f"- duplicate: то же самое, что уже есть\n" + f"- corroborates: подтверждает/уточняет существующее\n" + f"- contradicts: противоречит существующему" + ) + r = _client.post(f"{config.OLLAMA_URL}/api/chat", json={ + "model": m, + "messages": [{"role": "user", "content": prompt}], + "stream": False, + "options": {"temperature": 0.0}, + }) + r.raise_for_status() + out = r.json()["message"]["content"].strip() + start, end = out.find("{"), out.rfind("}") + if start == -1 or end == -1: + return {"verdict": "new", "confidence": 0.5, "reason": "parse error"} + try: + v = json.loads(out[start:end + 1]) + except json.JSONDecodeError: + return {"verdict": "new", "confidence": 0.5, "reason": "json error"} + if v.get("verdict") not in VERDICTS: + v["verdict"] = "new" + return v + + +def corroborate_atom(conn, atom_id: int, threshold: float = 0.90) -> int: + """Найти близкие атомы из ДРУГИХ источников (кросс-подтверждение). + + Возвращает число найденных подтверждений (не считая сам атом). + """ + from . import qdrant_store + with conn.cursor() as cur: + cur.execute("SELECT id, doc_id, claim, content FROM pm_atom WHERE id = %s", (atom_id,)) + row = cur.fetchone() + if not row: + return 0 + text = (row["claim"] or "") + " " + (row["content"] or "")[:800] + hits = qdrant_store.search(text, top_k=5) + count = 0 + for h in hits: + if int(h["id"]) == atom_id: + continue + if h["score"] >= threshold and h.get("doc_id") != row["doc_id"]: + count += 1 + return count + + +def verify_atom(conn, atom_id: int, *, self_critique_enabled: bool = True, + model: str | None = None) -> dict: + """Полная верификация одного атома. Возвращает результат. + + - self-critique: вердикт LLM (contradicts → contested). + - кросс-подтверждение: N близких из других источников (N>=1 → verified). + """ + with conn.cursor() as cur: + cur.execute("SELECT id, type, claim, content, confidence, status, doc_id FROM pm_atom WHERE id = %s", (atom_id,)) + atom = cur.fetchone() + if not atom: + return {"atom_id": atom_id, "error": "not found"} + + result = {"atom_id": atom_id, "verdict": None, "corroborations": 0, + "new_status": atom["status"], "new_confidence": atom["confidence"]} + + # 1) кросс-подтверждение по эмбеддингам (быстро, без LLM) + corr = corroborate_atom(conn, atom_id) + result["corroborations"] = corr + + # 2) self-critique (LLM): контекст — ближайшие по смыслу (эмбеддинги), а не по id + if self_critique_enabled: + from . import qdrant_store + qtext = (atom["claim"] or "") + " " + (atom["content"] or "")[:400] + hits = qdrant_store.search(qtext, top_k=8) + context = [{"claim": h.get("claim", "")} for h in hits if int(h["id"]) != atom_id] + v = self_critique(atom, context, model=model) + result["verdict"] = v["verdict"] + result["reason"] = v.get("reason") + + if v["verdict"] == "contradicts": + result["new_status"] = "contested" + elif v["verdict"] == "duplicate": + result["new_status"] = "stale" # дубликат — пометить неактуальным + elif v["verdict"] in ("corroborates", "new") or corr >= 1: + result["new_status"] = "verified" + elif corr >= 1: + result["new_status"] = "verified" + + # confidence: кросс-подтверждение повышает уверенность + new_conf = min(1.0, float(atom["confidence"] or 0.5) + 0.1 * corr) + result["new_confidence"] = round(new_conf, 2) + + with conn.cursor() as cur: + cur.execute( + "UPDATE pm_atom SET status = %s, confidence = %s, updated_at = now() WHERE id = %s", + (result["new_status"], result["new_confidence"], atom_id), + ) + conn.commit() + return result + + +def verify_all(conn, *, limit: int = 50, self_critique_enabled: bool = True, + model: str | None = None) -> dict: + """Верифицировать все candidate-атомы (порциями). Возвращает сводку.""" + with conn.cursor() as cur: + cur.execute("SELECT id FROM pm_atom WHERE status = 'candidate' ORDER BY id LIMIT %s", (limit,)) + ids = [r["id"] for r in cur.fetchall()] + summary = {"total": len(ids), "verified": 0, "contested": 0, "stale": 0, "unchanged": 0} + for aid in ids: + try: + r = verify_atom(conn, aid, self_critique_enabled=self_critique_enabled, model=model) + except Exception as e: + print(f"[verify] atom {aid} failed: {e}") + continue + ns = r.get("new_status") + if ns == "verified": + summary["verified"] += 1 + elif ns == "contested": + summary["contested"] += 1 + elif ns == "stale": + summary["stale"] += 1 + else: + summary["unchanged"] += 1 + return summary diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3da62e9 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +psycopg[binary] +qdrant-client +httpx[socks] +neo4j +pypdf +ebooklib +trafilatura +tree-sitter +tree-sitter-python +tree-sitter-javascript +tree-sitter-rust diff --git a/schema.sql b/schema.sql new file mode 100644 index 0000000..0fcb5da --- /dev/null +++ b/schema.sql @@ -0,0 +1,159 @@ +-- Pegas Memory: PostgreSQL — источник истины +-- Запускать в atom-db (:5434), БД content_atom (или отдельная БД alicorn) + +CREATE EXTENSION IF NOT EXISTS pg_trgm; +CREATE EXTENSION IF NOT EXISTS btree_gin; + +-- ============================================================ +-- ИСТОЧНИКИ +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_document ( + id BIGSERIAL PRIMARY KEY, + kind TEXT NOT NULL, -- web|pdf|epub|text|code_repo|manual + uri TEXT, + title TEXT, + fetched_at TIMESTAMPTZ, + content_hash TEXT, -- дедуп целого источника + authority REAL DEFAULT 0.5, -- репутация источника + meta JSONB DEFAULT '{}', + created_at TIMESTAMPTZ DEFAULT now(), + UNIQUE (uri) +); + +-- ============================================================ +-- ЧАНКИ (сырые сегменты — доказательства, НЕ для retrieval напрямую) +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_chunk ( + id BIGSERIAL PRIMARY KEY, + doc_id BIGINT REFERENCES pm_document(id) ON DELETE CASCADE, + seq INT, + content TEXT NOT NULL, + locator JSONB DEFAULT '{}', -- page/§/URL-якорь/строка + created_at TIMESTAMPTZ DEFAULT now() +); +CREATE INDEX IF NOT EXISTS pm_chunk_doc_idx ON pm_chunk (doc_id, seq); + +-- ============================================================ +-- АТОМЫ ЗНАНИЯ (основная единица retrieval) +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_atom ( + id BIGSERIAL PRIMARY KEY, + type TEXT NOT NULL, -- text|definition|logical_rule|code|table|fact + claim TEXT, -- короткое каноническое утверждение + content TEXT NOT NULL, -- полное значение (для code — исходник/сигнатура) + content_hash TEXT NOT NULL, -- дедуп + confidence REAL DEFAULT 0.5, + status TEXT DEFAULT 'candidate', -- candidate|verified|stale|contested + doc_id BIGINT REFERENCES pm_document(id) ON DELETE SET NULL, + locator JSONB DEFAULT '{}', + metadata JSONB DEFAULT '{}', -- logical_rule: pattern/conclusion; code: repo/path + created_at TIMESTAMPTZ DEFAULT now(), + updated_at TIMESTAMPTZ DEFAULT now() +); +CREATE UNIQUE INDEX IF NOT EXISTS pm_atom_hash_idx ON pm_atom (content_hash); +CREATE INDEX IF NOT EXISTS pm_atom_type_status_idx ON pm_atom (type, status); +CREATE INDEX IF NOT EXISTS pm_atom_claim_idx ON pm_atom USING gin (claim gin_trgm_ops); + +-- ============================================================ +-- СВЯЗИ МЕЖДУ АТОМАМИ (граф → Memgraph) +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_atom_link ( + id BIGSERIAL PRIMARY KEY, + src BIGINT REFERENCES pm_atom(id) ON DELETE CASCADE, + dst BIGINT REFERENCES pm_atom(id) ON DELETE CASCADE, + kind TEXT NOT NULL, -- inference|cause|reference|contradicts|depends_on|calls|defines + confidence REAL DEFAULT 0.5, + source TEXT, -- manual|extraction|inference + inferred BOOLEAN DEFAULT false, -- выведено инференцией (не явно) + created_at TIMESTAMPTZ DEFAULT now(), + UNIQUE (src, dst, kind) +); +CREATE INDEX IF NOT EXISTS pm_link_src_idx ON pm_atom_link (src); +CREATE INDEX IF NOT EXISTS pm_link_dst_idx ON pm_atom_link (dst); + +-- ============================================================ +-- ЦИТИРОВАНИЕ (доказательная база) +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_citation ( + id BIGSERIAL PRIMARY KEY, + atom_id BIGINT REFERENCES pm_atom(id) ON DELETE CASCADE, + doc_id BIGINT REFERENCES pm_document(id) ON DELETE CASCADE, + locator JSONB DEFAULT '{}', + quoted_text TEXT, + created_at TIMESTAMPTZ DEFAULT now() +); +CREATE INDEX IF NOT EXISTS pm_citation_atom_idx ON pm_citation (atom_id); + +-- ============================================================ +-- СИМВОЛЫ КОДА (tree-sitter) +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_code_symbol ( + id BIGSERIAL PRIMARY KEY, + repo TEXT, + path TEXT, + language TEXT, + symbol_name TEXT, + kind TEXT, -- function|class|method|type|const + signature TEXT, -- индексируется для retrieval + body_hash TEXT, -- content-addressed + line_start INT, + line_end INT, + created_at TIMESTAMPTZ DEFAULT now() +); +CREATE INDEX IF NOT EXISTS pm_symbol_name_idx ON pm_code_symbol USING gin (symbol_name gin_trgm_ops); +CREATE INDEX IF NOT EXISTS pm_symbol_lang_idx ON pm_code_symbol (language, kind); + +-- ============================================================ +-- ЗАВИСИМОСТИ КОДА (граф кода: calls/imports/inherits) +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_code_dependency ( + id BIGSERIAL PRIMARY KEY, + repo TEXT, + path TEXT, + language TEXT, + src TEXT, -- имя символа-источника + dst TEXT, -- имя цели + kind TEXT NOT NULL, -- calls|imports|inherits + created_at TIMESTAMPTZ DEFAULT now(), + UNIQUE (repo, path, src, dst, kind) +); +CREATE INDEX IF NOT EXISTS pm_code_dep_src_idx ON pm_code_dependency (src); +CREATE INDEX IF NOT EXISTS pm_code_dep_kind_idx ON pm_code_dependency (kind); + +-- ============================================================ +-- ПАМЯТЬ АГЕНТА (episodic) — ОТДЕЛЬНО от базы знаний +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_episodic_trace ( + id BIGSERIAL PRIMARY KEY, + session_id TEXT, + step INT, + kind TEXT, -- plan|action|observation|reflection + content TEXT NOT NULL, + created_at TIMESTAMPTZ DEFAULT now() +); +CREATE INDEX IF NOT EXISTS pm_episodic_session_idx ON pm_episodic_trace (session_id, step); + +-- ============================================================ +-- ПРОБЕЛЫ ДЛЯ ИССЛЕДОВАНИЯ +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_research_gap ( + id BIGSERIAL PRIMARY KEY, + question TEXT NOT NULL, + reason TEXT, -- low_coverage|low_confidence|stale|contested|user_query + priority REAL DEFAULT 0.5, + status TEXT DEFAULT 'open', -- open|researching|resolved|dismissed + created_at TIMESTAMPTZ DEFAULT now() +); + +-- ============================================================ +-- OUTBOX (CDC → Qdrant + Memgraph) +-- ============================================================ +CREATE TABLE IF NOT EXISTS pm_outbox ( + id BIGSERIAL PRIMARY KEY, + entity_type TEXT NOT NULL, -- atom|link|doc + entity_id BIGINT NOT NULL, + op TEXT NOT NULL, -- upsert|delete + created_at TIMESTAMPTZ DEFAULT now(), + processed BOOLEAN DEFAULT false +); +CREATE INDEX IF NOT EXISTS pm_outbox_pending_idx ON pm_outbox (processed, id); diff --git a/scripts/init_db.py b/scripts/init_db.py new file mode 100644 index 0000000..db6df77 --- /dev/null +++ b/scripts/init_db.py @@ -0,0 +1,26 @@ +"""Инициализация: применить schema.sql + создать коллекцию Qdrant.""" +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from pegas_memory import config, db, qdrant_store + +def main(): + print("Применяю schema.sql к", config.PG_DSN.split('@')[-1]) + sql = pathlib.Path(__file__).resolve().parents[1] / "schema.sql" + conn = db.connect() + with conn.cursor() as cur: + cur.execute(sql.read_text()) + conn.commit() + print(" -> schema OK") + + print("Создаю коллекцию Qdrant", config.QDRANT_COLLECTION, "(dim=%d)" % config.EMBED_DIM) + qdrant_store.ensure_collection() + print(" -> collection OK") + + # проверить эмбеддинг + v = __import__('pegas_memory.embedding', fromlist=['embed']).embed("проверка") + print(" -> embedding dim:", len(v)) + print("DONE") + +if __name__ == "__main__": + main() diff --git a/scripts/smoke_test.py b/scripts/smoke_test.py new file mode 100644 index 0000000..6cc3371 --- /dev/null +++ b/scripts/smoke_test.py @@ -0,0 +1,32 @@ +"""Smoke-тест: текст → атомы → PG → Qdrant → поиск.""" +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from pegas_memory import db, pipeline, outbox_sync, qdrant_store + +SAMPLE = ''' +PostgreSQL — это объектно-реляционная система управления базами данных. +pgvector — расширение PostgreSQL для хранения векторов и поиска по сходству. +Qdrant — специализированная векторная база данных, быстрее pgvector на больших объёмах. +Memgraph — in-memory графовая база данных, использует язык запросов Cypher. +EmbeddingGemma 2 — модель эмбеддингов Google, размерность 768. +''' + +def main(): + conn = db.connect() + print("Ингест тестового текста...") + stats = pipeline.ingest_text(conn, SAMPLE, title="smoke-test", min_confidence=0.3) + print(" stats:", stats) + + print("Outbox-синк → Qdrant...") + s = outbox_sync.sync_all(conn) + print(" sync:", s) + + print("Поиск 'векторная база данных'...") + hits = qdrant_store.search("векторная база данных", top_k=3) + for h in hits: + print(f" [{h['score']:.3f}] ({h['type']}) {h['claim'][:80]}") + print("DONE") + +if __name__ == "__main__": + main()