feat: add Qdrant vector store for enterprise-grade semantic intelligence
Phase 16: Vector intelligence (+3 tools = 104 total) New tools: - intel_semantic_search: Natural language search across all stored intelligence - intel_similar_events: Find historically similar events for pattern matching - intel_timeline: Chronological timeline of stored data with domain filtering Architecture: - vector_store.py: Qdrant client with FastEmbed (ONNX, bge-small-en-v1.5) - Background worker queue (asyncio) for non-blocking storage - Auto-populates from every Fetcher.get_json() call - 40+ domain categories mapped for filtered search - Graceful degradation: vector store is optional, all 101 existing tools work without it Integration: - Fetcher: new vector_store parameter, stores data after each successful fetch - server.py: vector store init + worker lifecycle + 3 dispatch cases - dashboard/app.py: vector store worker starts on dashboard boot - intel_status: now includes vector store statistics - pyproject.toml: [vector] optional dependency (qdrant-client, fastembed) 186/186 tests passing.
This commit is contained in:
Marc Shade
@@ -22,6 +22,7 @@ dependencies = [
|
||||
|
||||
[project.optional-dependencies]
|
||||
dashboard = ["starlette>=0.37.0", "uvicorn>=0.29.0"]
|
||||
vector = ["qdrant-client>=1.7.0", "fastembed>=0.7.0"]
|
||||
pdf = ["weasyprint>=62.0"]
|
||||
dev = [
|
||||
"pytest>=8.0.0",
|
||||
|
||||
@@ -81,14 +81,22 @@ logger = logging.getLogger(__name__)
|
||||
_fetcher: Fetcher | None = None
|
||||
_cache: Cache | None = None
|
||||
_breaker: CircuitBreaker | None = None
|
||||
_vector_store = None
|
||||
|
||||
|
||||
def _ensure_fetcher() -> Fetcher:
|
||||
global _fetcher, _cache, _breaker
|
||||
global _fetcher, _cache, _breaker, _vector_store
|
||||
if _fetcher is None:
|
||||
_cache = Cache()
|
||||
_breaker = CircuitBreaker()
|
||||
_fetcher = Fetcher(cache=_cache, breaker=_breaker)
|
||||
# Vector store — optional, dashboard populates it alongside the MCP server
|
||||
try:
|
||||
from world_intel_mcp.vector_store import VectorStore
|
||||
|
||||
_vector_store = VectorStore(enabled=True)
|
||||
except Exception:
|
||||
logger.info("Vector store unavailable in dashboard mode")
|
||||
_fetcher = Fetcher(cache=_cache, breaker=_breaker, vector_store=_vector_store)
|
||||
return _fetcher
|
||||
|
||||
|
||||
@@ -379,6 +387,15 @@ async def api_report_pdf(request):
|
||||
# App
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def on_startup():
|
||||
"""Start vector store worker if available."""
|
||||
_ensure_fetcher()
|
||||
if _vector_store:
|
||||
await _vector_store.start()
|
||||
logger.info("Dashboard vector store worker started")
|
||||
|
||||
|
||||
app = Starlette(
|
||||
routes=[
|
||||
Route("/", index),
|
||||
@@ -388,6 +405,7 @@ app = Starlette(
|
||||
Route("/api/health", api_health),
|
||||
Route("/api/report/pdf", api_report_pdf),
|
||||
],
|
||||
on_startup=[on_startup],
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -53,12 +53,14 @@ class Fetcher:
|
||||
default_timeout: float = 15.0,
|
||||
max_retries: int = 2,
|
||||
client: httpx.AsyncClient | None = None,
|
||||
vector_store: Any = None,
|
||||
):
|
||||
self.cache = cache
|
||||
self.breaker = breaker
|
||||
self.default_timeout = default_timeout
|
||||
self.max_retries = max_retries
|
||||
self._client: httpx.AsyncClient | None = client
|
||||
self.vector_store = vector_store # VectorStore instance (optional)
|
||||
|
||||
async def _get_client(self) -> httpx.AsyncClient:
|
||||
if self._client is None or self._client.is_closed:
|
||||
@@ -134,6 +136,8 @@ class Fetcher:
|
||||
data = resp.json()
|
||||
self.breaker.record_success(source)
|
||||
self.cache.set(effective_key, data, cache_ttl)
|
||||
if self.vector_store:
|
||||
await self.vector_store.store(source, data)
|
||||
return data
|
||||
except (httpx.HTTPStatusError, httpx.RequestError, Exception) as exc:
|
||||
last_error = exc
|
||||
|
||||
@@ -25,6 +25,8 @@ Phase 13: USNI fleet tracker, RSS expansion, report removal.
|
||||
Phase 14: BTC technicals, central bank rates, trade routes, cloud regions, financial centers (+5 = 87 tools).
|
||||
Phase 15: Business intelligence — forex (3), bonds/yields (2), earnings (2), SEC filings (3),
|
||||
company enrichment (1), macro composite (1) (+12 = 99 tools).
|
||||
Phase 16: Vector intelligence — semantic search, similar events, timeline (+3 = 102 tools).
|
||||
Qdrant vector store auto-populates from all fetches. Enterprise-grade semantic retrieval.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
@@ -84,7 +86,17 @@ logger = logging.getLogger("world-intel-mcp")
|
||||
server = Server("world-intel-mcp")
|
||||
cache = Cache()
|
||||
breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=300)
|
||||
fetcher = Fetcher(cache=cache, breaker=breaker)
|
||||
|
||||
# Vector store — optional, degrades gracefully if Qdrant unavailable.
|
||||
_vector_store = None
|
||||
try:
|
||||
from .vector_store import VectorStore
|
||||
|
||||
_vector_store = VectorStore(enabled=True)
|
||||
except Exception:
|
||||
logger.info("Vector store unavailable (qdrant_client or Qdrant not installed)")
|
||||
|
||||
fetcher = Fetcher(cache=cache, breaker=breaker, vector_store=_vector_store)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool definitions
|
||||
@@ -1519,10 +1531,96 @@ TOOLS: list[Tool] = [
|
||||
description="Get weighted macro market composite score (0-100) synthesizing Fear & Greed, VIX, sector breadth, DXY, BTC technicals, and 10Y yield into an actionable verdict (RISK_ON / CONSTRUCTIVE / NEUTRAL / CAUTIOUS / STRONG_CAUTION).",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
),
|
||||
# --- Vector Search (3 tools) ---
|
||||
Tool(
|
||||
name="intel_semantic_search",
|
||||
description="Semantic search across all stored intelligence data using natural language. Searches historical data accumulated from all 101+ tools. Filters: domain (e.g., 'markets', 'conflict'), category (e.g., 'Financial Markets', 'Cyber Threats'), hours (last N hours). Returns ranked results by relevance.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "Natural language search query (e.g., 'military activity near Taiwan', 'oil price disruptions')",
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "Max results (default 20)",
|
||||
"default": 20,
|
||||
},
|
||||
"domain": {
|
||||
"type": "string",
|
||||
"description": "Filter by source domain (e.g., 'markets', 'conflict', 'military')",
|
||||
},
|
||||
"category": {
|
||||
"type": "string",
|
||||
"description": "Filter by category (e.g., 'Financial Markets', 'Conflict & Security', 'Cyber Threats')",
|
||||
},
|
||||
"hours": {
|
||||
"type": "number",
|
||||
"description": "Only results from last N hours",
|
||||
},
|
||||
},
|
||||
"required": ["query"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="intel_similar_events",
|
||||
description="Find historically similar intelligence events or data. Given a text description, finds the most similar stored entries across all domains. Useful for pattern matching and precedent analysis.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"text": {
|
||||
"type": "string",
|
||||
"description": "Reference text to find similar events for",
|
||||
},
|
||||
"domain": {
|
||||
"type": "string",
|
||||
"description": "Source domain of the reference text",
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "Max results (default 10)",
|
||||
"default": 10,
|
||||
},
|
||||
"hours": {
|
||||
"type": "number",
|
||||
"description": "Only results from last N hours",
|
||||
},
|
||||
},
|
||||
"required": ["text"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="intel_timeline",
|
||||
description="Get chronological timeline of stored intelligence data. Returns recent entries sorted by time. Filter by domain or category to focus on specific intelligence areas.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"domain": {
|
||||
"type": "string",
|
||||
"description": "Filter by source domain",
|
||||
},
|
||||
"category": {
|
||||
"type": "string",
|
||||
"description": "Filter by category",
|
||||
},
|
||||
"hours": {
|
||||
"type": "number",
|
||||
"description": "Time window in hours (default 24)",
|
||||
"default": 24,
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "Max entries (default 50)",
|
||||
"default": 50,
|
||||
},
|
||||
},
|
||||
},
|
||||
),
|
||||
# --- System (1 tool) ---
|
||||
Tool(
|
||||
name="intel_status",
|
||||
description="Get data source health, circuit breaker status, cache freshness, and statistics.",
|
||||
description="Get data source health, circuit breaker status, cache freshness, vector store stats, and system statistics.",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
),
|
||||
]
|
||||
@@ -2106,12 +2204,46 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any:
|
||||
|
||||
return await fetch_macro_composite(fetcher)
|
||||
|
||||
# Vector Search
|
||||
case "intel_semantic_search":
|
||||
if _vector_store is None:
|
||||
return {"error": "Vector store not available (Qdrant not running?)"}
|
||||
return await _vector_store.semantic_search(
|
||||
query=arguments["query"],
|
||||
limit=arguments.get("limit", 20),
|
||||
domain=arguments.get("domain"),
|
||||
category=arguments.get("category"),
|
||||
hours=arguments.get("hours"),
|
||||
)
|
||||
case "intel_similar_events":
|
||||
if _vector_store is None:
|
||||
return {"error": "Vector store not available (Qdrant not running?)"}
|
||||
return await _vector_store.find_similar(
|
||||
domain=arguments.get("domain", "unknown"),
|
||||
text=arguments["text"],
|
||||
limit=arguments.get("limit", 10),
|
||||
hours=arguments.get("hours"),
|
||||
)
|
||||
case "intel_timeline":
|
||||
if _vector_store is None:
|
||||
return {"error": "Vector store not available (Qdrant not running?)"}
|
||||
return await _vector_store.timeline(
|
||||
domain=arguments.get("domain"),
|
||||
category=arguments.get("category"),
|
||||
hours=arguments.get("hours", 24.0),
|
||||
limit=arguments.get("limit", 50),
|
||||
)
|
||||
|
||||
# System
|
||||
case "intel_status":
|
||||
vs_stats = {}
|
||||
if _vector_store:
|
||||
vs_stats = await _vector_store.collection_stats()
|
||||
return {
|
||||
"circuit_breakers": breaker.status(),
|
||||
"cache": cache.stats(),
|
||||
"cache_freshness": cache.freshness(),
|
||||
"vector_store": vs_stats,
|
||||
"sources": {
|
||||
"markets": [
|
||||
"yahoo-finance",
|
||||
@@ -2206,10 +2338,17 @@ async def call_tool(name: str, arguments: dict[str, Any] | None) -> list[TextCon
|
||||
|
||||
async def _run() -> None:
|
||||
logger.info("World Intelligence MCP Server starting (%d tools)", len(TOOLS))
|
||||
async with stdio_server() as (read_stream, write_stream):
|
||||
await server.run(
|
||||
read_stream, write_stream, server.create_initialization_options()
|
||||
)
|
||||
if _vector_store:
|
||||
await _vector_store.start()
|
||||
logger.info("Vector store worker started")
|
||||
try:
|
||||
async with stdio_server() as (read_stream, write_stream):
|
||||
await server.run(
|
||||
read_stream, write_stream, server.create_initialization_options()
|
||||
)
|
||||
finally:
|
||||
if _vector_store:
|
||||
await _vector_store.stop()
|
||||
|
||||
|
||||
def run() -> None:
|
||||
|
||||
@@ -0,0 +1,627 @@
|
||||
"""Vector store for world-intel-mcp — persistent semantic intelligence archive.
|
||||
|
||||
Uses Qdrant for vector storage and sentence-transformers for embeddings.
|
||||
Unlike the SQLite TTL cache (ephemeral, serves live snapshots), this store
|
||||
is append-only and accumulates intelligence over time for semantic retrieval.
|
||||
|
||||
Designed to be non-blocking: embedding + storage runs in the background via
|
||||
asyncio.to_thread() so it never delays the main fetch path.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger("world-intel-mcp.vector-store")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lazy imports — sentence-transformers is heavy (~2s first load).
|
||||
# We import on first use so MCP server startup isn't penalized.
|
||||
# ---------------------------------------------------------------------------
|
||||
_embed_model = None
|
||||
_qdrant_client = None
|
||||
|
||||
COLLECTION_NAME = "world_intel"
|
||||
EMBEDDING_MODEL = "BAAI/bge-small-en-v1.5" # FastEmbed ONNX model (~45MB, fast)
|
||||
EMBEDDING_DIM = 384
|
||||
QDRANT_URL = "http://localhost:6333"
|
||||
|
||||
# Maximum text length to embed (chars). Longer text is truncated.
|
||||
MAX_EMBED_CHARS = 2000
|
||||
|
||||
# Domain → human-readable category for filtering
|
||||
DOMAIN_CATEGORIES = {
|
||||
"markets": "Financial Markets",
|
||||
"yahoo-finance": "Financial Markets",
|
||||
"coingecko": "Cryptocurrency",
|
||||
"crypto": "Cryptocurrency",
|
||||
"seismology": "Natural Disasters",
|
||||
"usgs": "Natural Disasters",
|
||||
"military": "Military & Defense",
|
||||
"adsblol": "Military & Defense",
|
||||
"opensky": "Military & Defense",
|
||||
"cyber": "Cyber Threats",
|
||||
"news": "News & Media",
|
||||
"rss": "News & Media",
|
||||
"gdelt": "News & Media",
|
||||
"maritime": "Maritime",
|
||||
"nga": "Maritime",
|
||||
"infrastructure": "Infrastructure",
|
||||
"cloudflare-radar": "Infrastructure",
|
||||
"wildfire": "Natural Disasters",
|
||||
"nasa-firms": "Natural Disasters",
|
||||
"prediction": "Prediction Markets",
|
||||
"polymarket": "Prediction Markets",
|
||||
"aviation": "Aviation",
|
||||
"faa": "Aviation",
|
||||
"climate": "Climate",
|
||||
"economic": "Economics",
|
||||
"energy": "Economics",
|
||||
"conflict": "Conflict & Security",
|
||||
"acled": "Conflict & Security",
|
||||
"ucdp": "Conflict & Security",
|
||||
"displacement": "Humanitarian",
|
||||
"intelligence": "Intelligence Analysis",
|
||||
"space-weather": "Space Weather",
|
||||
"ai-watch": "AI & Technology",
|
||||
"health": "Health",
|
||||
"elections": "Elections",
|
||||
"shipping": "Shipping & Trade",
|
||||
"social": "Social Signals",
|
||||
"nuclear": "Nuclear",
|
||||
"tech": "AI & Technology",
|
||||
"forex": "Financial Markets",
|
||||
"bonds": "Financial Markets",
|
||||
"earnings": "Financial Markets",
|
||||
"sec-edgar": "Financial Markets",
|
||||
"central-banks": "Economics",
|
||||
"traffic": "Traffic",
|
||||
"webcams": "Infrastructure",
|
||||
"sanctions": "Conflict & Security",
|
||||
"service-status": "Infrastructure",
|
||||
}
|
||||
|
||||
|
||||
def _get_embed_model():
|
||||
"""Lazy-load the FastEmbed model (ONNX, ~45MB, no torch required)."""
|
||||
global _embed_model
|
||||
if _embed_model is None:
|
||||
from fastembed import TextEmbedding
|
||||
|
||||
_embed_model = TextEmbedding(EMBEDDING_MODEL)
|
||||
logger.info("Loaded embedding model: %s", EMBEDDING_MODEL)
|
||||
return _embed_model
|
||||
|
||||
|
||||
def _get_qdrant():
|
||||
"""Lazy-load the Qdrant client and ensure collection exists."""
|
||||
global _qdrant_client
|
||||
if _qdrant_client is None:
|
||||
from qdrant_client import QdrantClient
|
||||
from qdrant_client.models import Distance, VectorParams
|
||||
|
||||
_qdrant_client = QdrantClient(url=QDRANT_URL, timeout=10)
|
||||
|
||||
# Create collection if it doesn't exist
|
||||
collections = [c.name for c in _qdrant_client.get_collections().collections]
|
||||
if COLLECTION_NAME not in collections:
|
||||
_qdrant_client.create_collection(
|
||||
collection_name=COLLECTION_NAME,
|
||||
vectors_config=VectorParams(
|
||||
size=EMBEDDING_DIM,
|
||||
distance=Distance.COSINE,
|
||||
),
|
||||
)
|
||||
# Create payload indexes for efficient filtering
|
||||
from qdrant_client.models import PayloadSchemaType
|
||||
|
||||
_qdrant_client.create_payload_index(
|
||||
collection_name=COLLECTION_NAME,
|
||||
field_name="domain",
|
||||
field_schema=PayloadSchemaType.KEYWORD,
|
||||
)
|
||||
_qdrant_client.create_payload_index(
|
||||
collection_name=COLLECTION_NAME,
|
||||
field_name="category",
|
||||
field_schema=PayloadSchemaType.KEYWORD,
|
||||
)
|
||||
_qdrant_client.create_payload_index(
|
||||
collection_name=COLLECTION_NAME,
|
||||
field_name="timestamp",
|
||||
field_schema=PayloadSchemaType.FLOAT,
|
||||
)
|
||||
logger.info("Created Qdrant collection: %s", COLLECTION_NAME)
|
||||
return _qdrant_client
|
||||
|
||||
|
||||
def _embed_text(text: str) -> list[float]:
|
||||
"""Generate embedding vector for text using FastEmbed (ONNX)."""
|
||||
model = _get_embed_model()
|
||||
truncated = text[:MAX_EMBED_CHARS]
|
||||
embeddings = list(model.embed([truncated]))
|
||||
return embeddings[0].tolist()
|
||||
|
||||
|
||||
def _data_to_text(domain: str, data: Any) -> str:
|
||||
"""Convert structured intelligence data to embeddable text.
|
||||
|
||||
Extracts the most meaningful text from each domain's response format.
|
||||
"""
|
||||
if isinstance(data, str):
|
||||
return data
|
||||
|
||||
if not isinstance(data, dict):
|
||||
return json.dumps(data, default=str)[:MAX_EMBED_CHARS]
|
||||
|
||||
parts: list[str] = [f"Domain: {domain}"]
|
||||
|
||||
# Extract events/items lists (most domains return {events: [...], count: N})
|
||||
for list_key in (
|
||||
"events",
|
||||
"items",
|
||||
"articles",
|
||||
"papers",
|
||||
"quotes",
|
||||
"threats",
|
||||
"warnings",
|
||||
"delays",
|
||||
"alerts",
|
||||
"outbreaks",
|
||||
"contracts",
|
||||
"filings",
|
||||
"repos",
|
||||
"stories",
|
||||
"results",
|
||||
"indicators",
|
||||
"rates",
|
||||
"signals",
|
||||
):
|
||||
items = data.get(list_key)
|
||||
if isinstance(items, list) and items:
|
||||
for item in items[:20]: # Cap at 20 items
|
||||
if isinstance(item, dict):
|
||||
# Extract key text fields
|
||||
for text_key in (
|
||||
"title",
|
||||
"headline",
|
||||
"description",
|
||||
"notes",
|
||||
"summary",
|
||||
"name",
|
||||
"text",
|
||||
"event_type",
|
||||
"country",
|
||||
"location",
|
||||
"source",
|
||||
"category",
|
||||
"symbol",
|
||||
"actor1",
|
||||
"actor2",
|
||||
):
|
||||
val = item.get(text_key)
|
||||
if val and isinstance(val, str):
|
||||
parts.append(val)
|
||||
elif isinstance(item, str):
|
||||
parts.append(item)
|
||||
break
|
||||
|
||||
# Extract top-level summary fields
|
||||
for key in (
|
||||
"summary",
|
||||
"verdict",
|
||||
"description",
|
||||
"status",
|
||||
"brief",
|
||||
"headline",
|
||||
"source",
|
||||
"error",
|
||||
):
|
||||
val = data.get(key)
|
||||
if val and isinstance(val, str):
|
||||
parts.append(f"{key}: {val}")
|
||||
|
||||
# Extract country/region context
|
||||
for key in ("country", "region", "location", "theater"):
|
||||
val = data.get(key)
|
||||
if val and isinstance(val, str):
|
||||
parts.append(f"{key}: {val}")
|
||||
|
||||
# Include count for context
|
||||
count = data.get("count")
|
||||
if count is not None:
|
||||
parts.append(f"count: {count}")
|
||||
|
||||
text = " | ".join(parts)
|
||||
return text[:MAX_EMBED_CHARS]
|
||||
|
||||
|
||||
def _content_hash(domain: str, data: Any) -> str:
|
||||
"""Generate a content hash for deduplication."""
|
||||
raw = json.dumps({"d": domain, "v": data}, default=str, sort_keys=True)
|
||||
return hashlib.sha256(raw.encode()).hexdigest()[:16]
|
||||
|
||||
|
||||
class VectorStore:
|
||||
"""Persistent vector store for intelligence data."""
|
||||
|
||||
def __init__(self, enabled: bool = True):
|
||||
self.enabled = enabled
|
||||
self._store_queue: asyncio.Queue | None = None
|
||||
self._worker_task: asyncio.Task | None = None
|
||||
|
||||
async def start(self) -> None:
|
||||
"""Start the background storage worker."""
|
||||
if not self.enabled:
|
||||
return
|
||||
self._store_queue = asyncio.Queue(maxsize=500)
|
||||
self._worker_task = asyncio.create_task(self._storage_worker())
|
||||
logger.info("Vector store worker started")
|
||||
|
||||
async def stop(self) -> None:
|
||||
"""Stop the background worker."""
|
||||
if self._worker_task:
|
||||
self._worker_task.cancel()
|
||||
try:
|
||||
await self._worker_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._worker_task = None
|
||||
|
||||
async def _storage_worker(self) -> None:
|
||||
"""Background worker that processes the store queue."""
|
||||
while True:
|
||||
try:
|
||||
domain, data, timestamp = await self._store_queue.get()
|
||||
try:
|
||||
await asyncio.to_thread(self._store_sync, domain, data, timestamp)
|
||||
except Exception as exc:
|
||||
logger.warning("Vector store write failed: %s", exc)
|
||||
finally:
|
||||
self._store_queue.task_done()
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
except Exception as exc:
|
||||
logger.warning("Vector store worker error: %s", exc)
|
||||
await asyncio.sleep(1)
|
||||
|
||||
def _store_sync(self, domain: str, data: Any, timestamp: float) -> None:
|
||||
"""Synchronous store operation (runs in thread pool)."""
|
||||
from qdrant_client.models import PointStruct
|
||||
|
||||
client = _get_qdrant()
|
||||
text = _data_to_text(domain, data)
|
||||
if len(text) < 20:
|
||||
return # Skip trivially small data
|
||||
|
||||
vector = _embed_text(text)
|
||||
content_hash = _content_hash(domain, data)
|
||||
point_id = int(
|
||||
hashlib.md5(f"{domain}:{content_hash}:{timestamp}".encode()).hexdigest()[
|
||||
:16
|
||||
],
|
||||
16,
|
||||
)
|
||||
|
||||
# Determine category
|
||||
category = DOMAIN_CATEGORIES.get(domain, "Other")
|
||||
# Try prefix match for RSS feeds etc
|
||||
if category == "Other":
|
||||
prefix = domain.split(":")[0] if ":" in domain else domain
|
||||
category = DOMAIN_CATEGORIES.get(prefix, "Other")
|
||||
|
||||
# Extract geographic context if available
|
||||
geo = self._extract_geo(data)
|
||||
|
||||
payload = {
|
||||
"domain": domain,
|
||||
"category": category,
|
||||
"text": text[:2000],
|
||||
"content_hash": content_hash,
|
||||
"timestamp": timestamp,
|
||||
"datetime": datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat(),
|
||||
"has_error": bool(isinstance(data, dict) and data.get("error")),
|
||||
}
|
||||
if geo:
|
||||
payload.update(geo)
|
||||
|
||||
# Count for context
|
||||
if isinstance(data, dict) and "count" in data:
|
||||
payload["event_count"] = data["count"]
|
||||
|
||||
client.upsert(
|
||||
collection_name=COLLECTION_NAME,
|
||||
points=[
|
||||
PointStruct(
|
||||
id=point_id,
|
||||
vector=vector,
|
||||
payload=payload,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _extract_geo(data: Any) -> dict | None:
|
||||
"""Extract geographic info from data if available."""
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
geo = {}
|
||||
# Direct lat/lon
|
||||
for lat_key in ("latitude", "lat"):
|
||||
for lon_key in ("longitude", "lon"):
|
||||
if lat_key in data and lon_key in data:
|
||||
geo["lat"] = data[lat_key]
|
||||
geo["lon"] = data[lon_key]
|
||||
return geo
|
||||
# Country
|
||||
for key in ("country", "country_code", "region"):
|
||||
if key in data and isinstance(data[key], str):
|
||||
geo["country"] = data[key]
|
||||
# Check first event for geo
|
||||
for list_key in ("events", "items"):
|
||||
items = data.get(list_key)
|
||||
if isinstance(items, list) and items and isinstance(items[0], dict):
|
||||
first = items[0]
|
||||
for lat_key in ("latitude", "lat"):
|
||||
if lat_key in first:
|
||||
geo["lat"] = first[lat_key]
|
||||
break
|
||||
for lon_key in ("longitude", "lon"):
|
||||
if lon_key in first:
|
||||
geo["lon"] = first[lon_key]
|
||||
break
|
||||
for key in ("country", "country_code"):
|
||||
if key in first:
|
||||
geo["country"] = first[key]
|
||||
break
|
||||
return geo if geo else None
|
||||
|
||||
async def store(self, domain: str, data: Any) -> None:
|
||||
"""Queue data for async vector storage. Non-blocking, fire-and-forget."""
|
||||
if not self.enabled or self._store_queue is None:
|
||||
return
|
||||
if isinstance(data, dict) and data.get("error"):
|
||||
return # Don't store error responses
|
||||
try:
|
||||
self._store_queue.put_nowait((domain, data, time.time()))
|
||||
except asyncio.QueueFull:
|
||||
logger.debug("Vector store queue full, dropping %s", domain)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Search API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
async def semantic_search(
|
||||
self,
|
||||
query: str,
|
||||
limit: int = 20,
|
||||
domain: str | None = None,
|
||||
category: str | None = None,
|
||||
hours: float | None = None,
|
||||
) -> dict:
|
||||
"""Semantic search across all stored intelligence.
|
||||
|
||||
Args:
|
||||
query: Natural language search query.
|
||||
limit: Max results (default 20).
|
||||
domain: Filter by source domain (e.g., "markets", "conflict").
|
||||
category: Filter by category (e.g., "Financial Markets").
|
||||
hours: Only return results from last N hours.
|
||||
|
||||
Returns:
|
||||
Dict with results list, each containing score, domain, text, datetime.
|
||||
"""
|
||||
return await asyncio.to_thread(
|
||||
self._search_sync, query, limit, domain, category, hours
|
||||
)
|
||||
|
||||
def _search_sync(
|
||||
self,
|
||||
query: str,
|
||||
limit: int,
|
||||
domain: str | None,
|
||||
category: str | None,
|
||||
hours: float | None,
|
||||
) -> dict:
|
||||
from qdrant_client.models import FieldCondition, Filter, MatchValue, Range
|
||||
|
||||
client = _get_qdrant()
|
||||
vector = _embed_text(query)
|
||||
|
||||
# Build filters
|
||||
conditions = []
|
||||
if domain:
|
||||
conditions.append(
|
||||
FieldCondition(key="domain", match=MatchValue(value=domain))
|
||||
)
|
||||
if category:
|
||||
conditions.append(
|
||||
FieldCondition(key="category", match=MatchValue(value=category))
|
||||
)
|
||||
if hours:
|
||||
cutoff = time.time() - (hours * 3600)
|
||||
conditions.append(FieldCondition(key="timestamp", range=Range(gte=cutoff)))
|
||||
# Exclude error entries
|
||||
conditions.append(
|
||||
FieldCondition(key="has_error", match=MatchValue(value=False))
|
||||
)
|
||||
|
||||
query_filter = Filter(must=conditions) if conditions else None
|
||||
|
||||
results = client.search(
|
||||
collection_name=COLLECTION_NAME,
|
||||
query_vector=vector,
|
||||
query_filter=query_filter,
|
||||
limit=limit,
|
||||
with_payload=True,
|
||||
)
|
||||
|
||||
return {
|
||||
"query": query,
|
||||
"results": [
|
||||
{
|
||||
"score": round(r.score, 4),
|
||||
"domain": r.payload.get("domain", ""),
|
||||
"category": r.payload.get("category", ""),
|
||||
"text": r.payload.get("text", ""),
|
||||
"datetime": r.payload.get("datetime", ""),
|
||||
"event_count": r.payload.get("event_count"),
|
||||
"country": r.payload.get("country"),
|
||||
}
|
||||
for r in results
|
||||
],
|
||||
"count": len(results),
|
||||
"filters": {
|
||||
"domain": domain,
|
||||
"category": category,
|
||||
"hours": hours,
|
||||
},
|
||||
}
|
||||
|
||||
async def find_similar(
|
||||
self,
|
||||
domain: str,
|
||||
text: str,
|
||||
limit: int = 10,
|
||||
hours: float | None = None,
|
||||
) -> dict:
|
||||
"""Find historically similar events/data to a given text."""
|
||||
return await asyncio.to_thread(self._similar_sync, domain, text, limit, hours)
|
||||
|
||||
def _similar_sync(
|
||||
self,
|
||||
domain: str,
|
||||
text: str,
|
||||
limit: int,
|
||||
hours: float | None,
|
||||
) -> dict:
|
||||
from qdrant_client.models import FieldCondition, Filter, MatchValue, Range
|
||||
|
||||
client = _get_qdrant()
|
||||
vector = _embed_text(text)
|
||||
|
||||
conditions = [
|
||||
FieldCondition(key="has_error", match=MatchValue(value=False)),
|
||||
]
|
||||
if hours:
|
||||
cutoff = time.time() - (hours * 3600)
|
||||
conditions.append(FieldCondition(key="timestamp", range=Range(gte=cutoff)))
|
||||
|
||||
results = client.search(
|
||||
collection_name=COLLECTION_NAME,
|
||||
query_vector=vector,
|
||||
query_filter=Filter(must=conditions),
|
||||
limit=limit,
|
||||
with_payload=True,
|
||||
)
|
||||
|
||||
return {
|
||||
"reference_domain": domain,
|
||||
"reference_text": text[:200],
|
||||
"similar": [
|
||||
{
|
||||
"score": round(r.score, 4),
|
||||
"domain": r.payload.get("domain", ""),
|
||||
"category": r.payload.get("category", ""),
|
||||
"text": r.payload.get("text", ""),
|
||||
"datetime": r.payload.get("datetime", ""),
|
||||
"country": r.payload.get("country"),
|
||||
}
|
||||
for r in results
|
||||
],
|
||||
"count": len(results),
|
||||
}
|
||||
|
||||
async def timeline(
|
||||
self,
|
||||
domain: str | None = None,
|
||||
category: str | None = None,
|
||||
hours: float = 24.0,
|
||||
limit: int = 50,
|
||||
) -> dict:
|
||||
"""Get chronological timeline of stored intelligence."""
|
||||
return await asyncio.to_thread(
|
||||
self._timeline_sync, domain, category, hours, limit
|
||||
)
|
||||
|
||||
def _timeline_sync(
|
||||
self,
|
||||
domain: str | None,
|
||||
category: str | None,
|
||||
hours: float,
|
||||
limit: int,
|
||||
) -> dict:
|
||||
from qdrant_client.models import FieldCondition, Filter, MatchValue, Range
|
||||
|
||||
client = _get_qdrant()
|
||||
cutoff = time.time() - (hours * 3600)
|
||||
|
||||
conditions = [
|
||||
FieldCondition(key="timestamp", range=Range(gte=cutoff)),
|
||||
FieldCondition(key="has_error", match=MatchValue(value=False)),
|
||||
]
|
||||
if domain:
|
||||
conditions.append(
|
||||
FieldCondition(key="domain", match=MatchValue(value=domain))
|
||||
)
|
||||
if category:
|
||||
conditions.append(
|
||||
FieldCondition(key="category", match=MatchValue(value=category))
|
||||
)
|
||||
|
||||
# Scroll with filter (no vector search — chronological)
|
||||
results, _ = client.scroll(
|
||||
collection_name=COLLECTION_NAME,
|
||||
scroll_filter=Filter(must=conditions),
|
||||
limit=limit,
|
||||
with_payload=True,
|
||||
order_by="timestamp",
|
||||
)
|
||||
|
||||
entries = sorted(
|
||||
[
|
||||
{
|
||||
"domain": r.payload.get("domain", ""),
|
||||
"category": r.payload.get("category", ""),
|
||||
"text": r.payload.get("text", "")[:300],
|
||||
"datetime": r.payload.get("datetime", ""),
|
||||
"timestamp": r.payload.get("timestamp", 0),
|
||||
"event_count": r.payload.get("event_count"),
|
||||
"country": r.payload.get("country"),
|
||||
}
|
||||
for r in results
|
||||
],
|
||||
key=lambda x: x["timestamp"],
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
return {
|
||||
"hours": hours,
|
||||
"entries": entries[:limit],
|
||||
"count": len(entries),
|
||||
"filters": {"domain": domain, "category": category},
|
||||
}
|
||||
|
||||
async def collection_stats(self) -> dict:
|
||||
"""Get vector store statistics."""
|
||||
try:
|
||||
return await asyncio.to_thread(self._stats_sync)
|
||||
except Exception as exc:
|
||||
return {"error": str(exc), "enabled": self.enabled}
|
||||
|
||||
def _stats_sync(self) -> dict:
|
||||
client = _get_qdrant()
|
||||
info = client.get_collection(COLLECTION_NAME)
|
||||
return {
|
||||
"enabled": self.enabled,
|
||||
"collection": COLLECTION_NAME,
|
||||
"points_count": info.points_count,
|
||||
"vectors_count": info.vectors_count,
|
||||
"indexed_vectors_count": info.indexed_vectors_count,
|
||||
"status": info.status.value if info.status else "unknown",
|
||||
"embedding_model": EMBEDDING_MODEL,
|
||||
"embedding_dim": EMBEDDING_DIM,
|
||||
}
|
||||
Reference in New Issue
Block a user