diff --git a/.env.example b/.env.example index 5e651598b..3f66d0ae6 100644 --- a/.env.example +++ b/.env.example @@ -17,6 +17,7 @@ INFOQUEST_API_KEY=your-infoquest-api-key # FIRECRAWL_API_KEY=your-firecrawl-api-key # VOLCENGINE_API_KEY=your-volcengine-api-key # OPENAI_API_KEY=your-openai-api-key +# OPENVIKING_API_KEY=your-openviking-trusted-root-api-key # GEMINI_API_KEY=your-gemini-api-key # DEEPSEEK_API_KEY=your-deepseek-api-key # NOVITA_API_KEY=your-novita-api-key # OpenAI-compatible, see https://novita.ai diff --git a/README.md b/README.md index 420f04429..ece7f1df9 100644 --- a/README.md +++ b/README.md @@ -960,6 +960,15 @@ Then uncomment the `group: browser` tool entries in `config.yaml` (`browser_navi Most agents forget everything the moment a conversation ends. DeerFlow remembers. +DeerFlow also includes an optional `openviking` memory backend. It connects to +an independent OpenViking server over HTTP, submits completed turns through +OpenViking Sessions, and recalls remote memories for prompt injection while +leaving DeerMem as the default. The initial integration supports +`memory.mode: middleware`. Submitted-message watermarks prevent a failed +Session commit from duplicating already accepted messages on retry; see +[OpenViking memory backend](docs/OPENVIKING.md) for configuration and Docker +startup. + Across sessions, DeerFlow builds a persistent memory of your profile, preferences, and accumulated knowledge. The more you use it, the better it knows you — your writing style, your technical stack, your recurring workflows. Memory is stored locally and stays under your control. Memory updates now skip duplicate fact entries at apply time, so repeated preferences and context do not accumulate endlessly across sessions. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index fa323801b..fc9f80458 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -229,7 +229,10 @@ Blocking-IO runtime gate (`tests/blocking_io/`): offloading the uploads-directory scan off the event loop); `test_uploads_router.py` (locks Gateway upload/list/delete endpoints offloading upload directory creation, staged writes, chmod/cleanup, - directory scans/deletes, and remote sandbox sync off the event loop); and + directory scans/deletes, and remote sandbox sync off the event loop); + `test_openviking_memory_backend.py` (locks the OpenViking backend's async + add/context/search entrypoints offloading synchronous HTTP and watermark + filesystem IO); and `test_workspace_changes_recorder.py` (locks the offload around the snapshot text cache lifecycle — roots resolution, `mkdtemp`, and the `shutil.rmtree` on both the capture-failure branch and `record_workspace_changes`' `finally`). @@ -847,6 +850,22 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ **Workflow**: - `memory.mode: middleware` (default) keeps the passive path: `MemoryMiddleware` filters messages (user inputs + final AI responses), captures `user_id` via `resolve_runtime_user_id(runtime)`, queues conversation with the captured `user_id`, and the debounced background thread invokes the LLM to extract context updates and facts using the stored `user_id`. `DynamicContextMiddleware` passes the same resolved identity to the memory read path. On standalone Agent Server runs, server-owned auth identity is also resolved during lead-agent construction, normalized through `make_safe_user_id` for DeerFlow storage, and explicitly reused for custom-agent config/SOUL, user skills, skill policy, and prompt assembly; ordinary client `user_id` values cannot override `langgraph_auth_user_id`. On the embedded Gateway path, `inject_authenticated_user_context` removes client-supplied `langgraph_auth_user` / `langgraph_auth_user_id` from both RunnableConfig sections before graph construction, so those reserved fields cannot impersonate Agent Server auth. +- The optional `openviking` backend under + `packages/harness/deerflow/agents/memory/backends/openviking/` is a + remote-only HTTP adapter. Select it with + `memory.manager_class: openviking` and keep `memory.mode: middleware`. It + commits filtered turns to OpenViking Sessions and maps remote memory search + results into the shared contract. It hashes `(user_id, agent_name)` into a + safe OpenViking trusted-user identity for hard scope isolation and keeps + bounded message watermarks below + `{storage_path}/openviking/sessions/`. The watermark separately records + submitted and committed message IDs: once batch submission succeeds, a + later update never resubmits those messages or retries an ambiguous commit. + A future batch can commit the still-open Session together with new messages. Session + locks are weakly cached, async entrypoints offload synchronous HTTP and file + IO, and graceful shutdown rejects new work before draining all in-flight + client operations within its timeout. It does not implement DeerMem fact + CRUD/import/export and must not import the OpenViking embedded runtime. - `memory.mode: tool` skips `MemoryMiddleware` and registers `memory_search`, `memory_add`, `memory_update`, and `memory_delete` on the agent. The model decides when to search, add, update, or delete facts; this is opt-in/experimental and should not be described as better than middleware mode without eval evidence. - Both modes share `FileMemoryStorage`, per-user/per-agent isolation, prompt injection, manual CRUD primitives, and the updater backend. - Middleware mode queue debounces (30s default), batches updates, and commits global summaries plus the selected/default agent's fact delta through a user-level lock, optimistic user-memory revisions, per-fact revisions, and a recoverable target-file journal. Only explicitly marked point operations may rebase a stale shared revision, and only while every addressed fact still satisfies its original absent/revision precondition. Snapshot-derived clear/trim/consolidation operations instead reload the complete document and recompute their intent on a manifest conflict, with a bounded retry. Typed manifest/fact conflict subclasses keep that decision independent of exception text, and same-ID creates and stale same-fact writes fail. Scope-lock objects are weakly cached so inactive users do not grow a process-lifetime map. Cache validation does not scale with the fact-file count: its token combines the shared JSON's `(mtime_ns, size, revision)`, so the persisted revision invalidates stale caches even when a coarse-mtime filesystem reports identical metadata for same-size writes; direct out-of-band Markdown edits require `reload()`. Atomic replacement also syncs the parent directory on POSIX so the rename is durable. DeerMem translates private storage conflict/corruption exceptions to the backend-neutral MemoryManager contract; the Gateway maps them to HTTP 409 and a stable HTTP 500 response respectively. A normal default-manager read automatically migrates legacy facts from the global JSON into `__default__`; it also adopts the earlier implicit `lead-agent` fact bucket only when that directory has no custom-agent `config.yaml`, and rejects unexpected files instead of deleting them. The v1-to-v2 migration is one-way for the running application: operators must stop DeerFlow and snapshot the configured storage root before upgrade. Before any destructive v2 write, every migrated JSON source is durably retained as `{manifest_filename}.v1.bak`; a missing-write or mismatched existing backup aborts without modifying v1 data. Legacy per-agent JSON is deleted only after its non-empty summaries are safely adopted or confirmed identical; summary conflicts keep the source file and fail loudly. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/README.md b/backend/packages/harness/deerflow/agents/memory/backends/README.md index 846008550..861762831 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/README.md +++ b/backend/packages/harness/deerflow/agents/memory/backends/README.md @@ -4,6 +4,7 @@ Each subfolder under `agents/memory/backends/` is a pluggable memory backend. Sw - `deermem/` - the default backend (deer-flow's own: structured facts + JSON storage). - `noop/` - an empty backend and the **template** to copy when adding a new one. +- `openviking/` - optional remote OpenViking backend over HTTP (middleware mode). This guide tells you **which files to touch** when you change, swap, or add a memory system. Paths are relative to `backend/` unless noted. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/__init__.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/__init__.py new file mode 100644 index 000000000..06feb3998 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/__init__.py @@ -0,0 +1,7 @@ +"""OpenViking HTTP memory backend.""" + +from .openviking_manager import OpenVikingMemoryManager + +MANAGER_CLASS = OpenVikingMemoryManager + +__all__ = ["OpenVikingMemoryManager"] diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/client.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/client.py new file mode 100644 index 000000000..894959f66 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/client.py @@ -0,0 +1,232 @@ +"""Thin synchronous HTTP client for the OpenViking server API.""" + +from __future__ import annotations + +import time +from typing import Any + +import httpx + +from .config import OpenVikingConfig +from .models import OpenVikingCommitResult, OpenVikingIdentity, OpenVikingMessage, OpenVikingSearchHit, OpenVikingSessionContext + + +class OpenVikingClientError(RuntimeError): + """Base error raised by the remote OpenViking adapter.""" + + def __init__(self, operation: str, message: str, *, status_code: int | None = None, code: str | None = None): + super().__init__(message) + self.operation = operation + self.status_code = status_code + self.code = code + + +class OpenVikingAuthenticationError(OpenVikingClientError): + pass + + +class OpenVikingTimeoutError(OpenVikingClientError): + pass + + +class OpenVikingUnavailableError(OpenVikingClientError): + pass + + +class OpenVikingProtocolError(OpenVikingClientError): + pass + + +class OpenVikingHttpClient: + """OpenViking API wrapper with bounded timeout and conservative retries.""" + + def __init__(self, config: OpenVikingConfig, *, transport: httpx.BaseTransport | None = None): + self._config = config + timeout = httpx.Timeout( + connect=config.connect_timeout_seconds, + read=config.read_timeout_seconds, + write=config.write_timeout_seconds, + pool=config.pool_timeout_seconds, + ) + self._client = httpx.Client(base_url=config.base_url, timeout=timeout, transport=transport) + + def close(self) -> None: + self._client.close() + + def health(self) -> bool: + response = self._request("health", "GET", "/health", identity=None, retryable=True) + if response.status_code != 200: + return False + try: + payload = response.json() + except ValueError: + return False + return payload.get("status") == "ok" + + def ensure_session(self, identity: OpenVikingIdentity, session_id: str) -> None: + self._request_result( + "session.ensure", + "GET", + f"/api/v1/sessions/{session_id}", + identity=identity, + params={"auto_create": "true"}, + retryable=True, + ) + + def add_messages(self, identity: OpenVikingIdentity, session_id: str, messages: list[OpenVikingMessage]) -> int: + if not messages: + return 0 + added = 0 + # OpenViking caps one batch at 100 messages. + for offset in range(0, len(messages), 100): + batch = messages[offset : offset + 100] + result = self._request_result( + "messages.add", + "POST", + f"/api/v1/sessions/{session_id}/messages/batch", + identity=identity, + json={"messages": [message.as_request() for message in batch]}, + retryable=False, + ) + added += int(result.get("added", len(batch))) + return added + + def commit_session(self, identity: OpenVikingIdentity, session_id: str) -> OpenVikingCommitResult: + result = self._request_result( + "session.commit", + "POST", + f"/api/v1/sessions/{session_id}/commit", + identity=identity, + json={"keep_recent_count": 0}, + retryable=False, + ) + return OpenVikingCommitResult( + status=str(result.get("status") or ""), + task_id=str(result["task_id"]) if result.get("task_id") else None, + archive_uri=str(result["archive_uri"]) if result.get("archive_uri") else None, + archived=bool(result.get("archived", False)), + ) + + def search( + self, + identity: OpenVikingIdentity, + query: str, + *, + top_k: int, + category: str | None = None, + session_id: str | None = None, + ) -> list[OpenVikingSearchHit]: + body: dict[str, Any] = { + "query": query, + "target_uri": "viking://user/memories", + "context_type": "memory", + "node_limit": top_k, + } + if session_id: + body["session_id"] = session_id + if self._config.score_threshold is not None: + body["score_threshold"] = self._config.score_threshold + result = self._request_result( + "search", + "POST", + "/api/v1/search/search" if session_id else "/api/v1/search/find", + identity=identity, + json=body, + retryable=True, + ) + values = result.get("memories", []) + if not isinstance(values, list): + raise OpenVikingProtocolError("search", "OpenViking response field result.memories is not a list") + hits = [OpenVikingSearchHit.from_response(value) for value in values if isinstance(value, dict)] + if category: + category_key = category.casefold() + hits = [hit for hit in hits if hit.category.casefold() == category_key] + return hits[:top_k] + + def get_session_context( + self, + identity: OpenVikingIdentity, + session_id: str, + *, + token_budget: int, + ) -> OpenVikingSessionContext: + result = self._request_result( + "session.context", + "GET", + f"/api/v1/sessions/{session_id}/context", + identity=identity, + params={"token_budget": token_budget}, + retryable=True, + ) + messages = result.get("messages", []) + return OpenVikingSessionContext( + latest_archive_overview=str(result.get("latest_archive_overview") or ""), + messages=messages if isinstance(messages, list) else [], + estimated_tokens=int(result.get("estimatedTokens") or 0), + ) + + def _headers(self, identity: OpenVikingIdentity | None) -> dict[str, str]: + headers = {"Accept": "application/json"} + if self._config.api_key: + headers["X-API-Key"] = self._config.api_key + if identity is not None and self._config.auth_mode == "trusted": + headers["X-OpenViking-Account"] = identity.account + headers["X-OpenViking-User"] = identity.user + return headers + + def _request_result(self, operation: str, method: str, path: str, *, identity: OpenVikingIdentity, retryable: bool, **kwargs: Any) -> dict[str, Any]: + response = self._request(operation, method, path, identity=identity, retryable=retryable, **kwargs) + try: + payload = response.json() + except ValueError as exc: + raise OpenVikingProtocolError(operation, "OpenViking returned non-JSON data", status_code=response.status_code) from exc + if not isinstance(payload, dict) or payload.get("status") != "ok": + error = payload.get("error", {}) if isinstance(payload, dict) else {} + code = str(error.get("code") or "UNKNOWN") if isinstance(error, dict) else "UNKNOWN" + message = str(error.get("message") or "OpenViking request failed") if isinstance(error, dict) else "OpenViking request failed" + error_type = OpenVikingAuthenticationError if response.status_code in {401, 403} else OpenVikingProtocolError + raise error_type(operation, message, status_code=response.status_code, code=code) + result = payload.get("result") + if not isinstance(result, dict): + raise OpenVikingProtocolError(operation, "OpenViking response field result is not an object", status_code=response.status_code) + return result + + def _request( + self, + operation: str, + method: str, + path: str, + *, + identity: OpenVikingIdentity | None, + retryable: bool, + **kwargs: Any, + ) -> httpx.Response: + attempts = self._config.max_retries + 1 if retryable else 1 + for attempt in range(attempts): + try: + response = self._client.request(method, path, headers=self._headers(identity), **kwargs) + except httpx.TimeoutException as exc: + if attempt + 1 < attempts: + time.sleep(0.05 * (2**attempt)) + continue + raise OpenVikingTimeoutError(operation, "OpenViking request timed out") from exc + except httpx.TransportError as exc: + if attempt + 1 < attempts: + time.sleep(0.05 * (2**attempt)) + continue + raise OpenVikingUnavailableError(operation, "OpenViking is unavailable") from exc + if response.status_code in {429, 502, 503, 504} and attempt + 1 < attempts: + time.sleep(0.05 * (2**attempt)) + continue + if response.status_code >= 400: + try: + payload = response.json() + except ValueError: + payload = {} + error = payload.get("error", {}) if isinstance(payload, dict) else {} + code = str(error.get("code") or "HTTP_ERROR") if isinstance(error, dict) else "HTTP_ERROR" + message = str(error.get("message") or f"OpenViking HTTP {response.status_code}") if isinstance(error, dict) else f"OpenViking HTTP {response.status_code}" + error_type = OpenVikingAuthenticationError if response.status_code in {401, 403} else OpenVikingProtocolError + raise error_type(operation, message, status_code=response.status_code, code=code) + return response + raise OpenVikingUnavailableError(operation, "OpenViking request failed") diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/config.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/config.py new file mode 100644 index 000000000..51aa630c4 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/config.py @@ -0,0 +1,127 @@ +"""Configuration for the OpenViking HTTP memory backend.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, Literal +from urllib.parse import urlparse + + +@dataclass(frozen=True) +class OpenVikingConfig: + """Parsed backend-private configuration. + + The backend is intentionally remote-only: DeerFlow talks to an independent + OpenViking server and does not import the OpenViking Python runtime. + """ + + base_url: str + storage_path: str + auth_mode: Literal["trusted", "dev"] + account: str + api_key: str | None + connect_timeout_seconds: float + read_timeout_seconds: float + write_timeout_seconds: float + pool_timeout_seconds: float + max_retries: int + search_top_k: int + score_threshold: float | None + max_injection_chars: int + injection_query: str + startup_policy: Literal["fail_fast", "warn"] + read_failure_policy: Literal["fail_open", "raise"] + write_failure_policy: Literal["log_and_drop", "raise"] + allow_insecure_http: bool + allow_insecure_dev: bool + max_seen_message_ids: int + + @classmethod + def from_backend_config(cls, backend_config: dict[str, Any] | None) -> OpenVikingConfig: + cfg = dict(backend_config or {}) + retrieval = _mapping(cfg.pop("retrieval", {}), "retrieval") + failure_policy = _mapping(cfg.pop("failure_policy", {}), "failure_policy") + + api_key_env = str(cfg.pop("api_key_env", "OPENVIKING_API_KEY")).strip() + api_key = os.environ.get(api_key_env) if api_key_env else None + + result = cls( + base_url=str(cfg.pop("base_url", "http://127.0.0.1:1933")).rstrip("/"), + storage_path=str(cfg.pop("storage_path", "")), + auth_mode=str(cfg.pop("auth_mode", "trusted")).lower(), # type: ignore[arg-type] + account=str(cfg.pop("account", "deerflow")).strip(), + api_key=api_key, + connect_timeout_seconds=float(cfg.pop("connect_timeout_seconds", 2.0)), + read_timeout_seconds=float(cfg.pop("read_timeout_seconds", 10.0)), + write_timeout_seconds=float(cfg.pop("write_timeout_seconds", 10.0)), + pool_timeout_seconds=float(cfg.pop("pool_timeout_seconds", 2.0)), + max_retries=int(cfg.pop("max_retries", 1)), + search_top_k=int(retrieval.pop("top_k", 8)), + score_threshold=_optional_float(retrieval.pop("score_threshold", None)), + max_injection_chars=int(retrieval.pop("max_injection_chars", 12_000)), + injection_query=str( + retrieval.pop( + "injection_query", + "user profile preferences important entities events ongoing goals constraints and prior decisions", + ) + ).strip(), + startup_policy=str(cfg.pop("startup_policy", "fail_fast")).lower(), # type: ignore[arg-type] + read_failure_policy=str(failure_policy.pop("read", "fail_open")).lower(), # type: ignore[arg-type] + write_failure_policy=str(failure_policy.pop("write", "log_and_drop")).lower(), # type: ignore[arg-type] + allow_insecure_http=bool(cfg.pop("allow_insecure_http", False)), + allow_insecure_dev=bool(cfg.pop("allow_insecure_dev", False)), + max_seen_message_ids=int(cfg.pop("max_seen_message_ids", 512)), + ) + + unknown = sorted([*cfg, *(f"retrieval.{key}" for key in retrieval), *(f"failure_policy.{key}" for key in failure_policy)]) + if unknown: + raise ValueError(f"Unknown OpenViking backend_config fields: {', '.join(unknown)}") + result._validate() + return result + + def _validate(self) -> None: + parsed = urlparse(self.base_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("OpenViking base_url must be an absolute http(s) URL") + if parsed.scheme == "http" and not self.allow_insecure_http and parsed.hostname not in {"127.0.0.1", "localhost", "openviking"}: + raise ValueError("OpenViking plain HTTP is allowed only for localhost/openviking; set allow_insecure_http=true for a trusted internal network") + if self.auth_mode not in {"trusted", "dev"}: + raise ValueError("OpenViking auth_mode must be 'trusted' or 'dev'") + if self.auth_mode == "dev" and not self.allow_insecure_dev: + raise ValueError("OpenViking auth_mode='dev' requires allow_insecure_dev=true") + if self.auth_mode == "trusted" and not self.account: + raise ValueError("OpenViking trusted auth requires a non-empty account") + for field_name in ("connect_timeout_seconds", "read_timeout_seconds", "write_timeout_seconds", "pool_timeout_seconds"): + if getattr(self, field_name) <= 0: + raise ValueError(f"OpenViking {field_name} must be > 0") + if not 0 <= self.max_retries <= 5: + raise ValueError("OpenViking max_retries must be between 0 and 5") + if not 1 <= self.search_top_k <= 100: + raise ValueError("OpenViking retrieval.top_k must be between 1 and 100") + if self.score_threshold is not None and not 0 <= self.score_threshold <= 1: + raise ValueError("OpenViking retrieval.score_threshold must be between 0 and 1") + if not 256 <= self.max_injection_chars <= 100_000: + raise ValueError("OpenViking retrieval.max_injection_chars must be between 256 and 100000") + if not self.injection_query: + raise ValueError("OpenViking retrieval.injection_query must not be empty") + if self.startup_policy not in {"fail_fast", "warn"}: + raise ValueError("OpenViking startup_policy must be 'fail_fast' or 'warn'") + if self.read_failure_policy not in {"fail_open", "raise"}: + raise ValueError("OpenViking failure_policy.read must be 'fail_open' or 'raise'") + if self.write_failure_policy not in {"log_and_drop", "raise"}: + raise ValueError("OpenViking failure_policy.write must be 'log_and_drop' or 'raise'") + if not 16 <= self.max_seen_message_ids <= 10_000: + raise ValueError("OpenViking max_seen_message_ids must be between 16 and 10000") + + +def _mapping(value: Any, name: str) -> dict[str, Any]: + if value is None: + return {} + if not isinstance(value, dict): + raise ValueError(f"OpenViking {name} must be a mapping") + return dict(value) + + +def _optional_float(value: Any) -> float | None: + return None if value is None else float(value) diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/models.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/models.py new file mode 100644 index 000000000..a86fbb4b8 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/models.py @@ -0,0 +1,63 @@ +"""Small transport-neutral models used by the OpenViking adapter.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class OpenVikingIdentity: + account: str + user: str + + +@dataclass(frozen=True) +class OpenVikingMessage: + message_id: str + role: str + content: str + + def as_request(self) -> dict[str, Any]: + # OpenViking AddMessageRequest generates its own message ID and rejects + # unknown request fields, so the DeerFlow ID remains adapter-local for + # watermarking and is not sent over the wire. + return {"role": self.role, "content": self.content} + + +@dataclass(frozen=True) +class OpenVikingCommitResult: + status: str + task_id: str | None + archive_uri: str | None + archived: bool + + +@dataclass(frozen=True) +class OpenVikingSearchHit: + uri: str + context_type: str + category: str + score: float + abstract: str + overview: str | None + match_reason: str + + @classmethod + def from_response(cls, value: dict[str, Any]) -> OpenVikingSearchHit: + return cls( + uri=str(value.get("uri") or ""), + context_type=str(value.get("context_type") or "memory"), + category=str(value.get("category") or "memory"), + score=float(value.get("score") or 0.0), + abstract=str(value.get("abstract") or ""), + overview=str(value["overview"]) if value.get("overview") else None, + match_reason=str(value.get("match_reason") or ""), + ) + + +@dataclass(frozen=True) +class OpenVikingSessionContext: + latest_archive_overview: str + messages: list[dict[str, Any]] + estimated_tokens: int diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py new file mode 100644 index 000000000..e1997d80e --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py @@ -0,0 +1,494 @@ +"""OpenViking HTTP implementation of the pluggable memory contract. + +This backend deliberately contains no OpenViking extraction or vector logic. +It forwards filtered conversation turns to OpenViking Sessions and maps remote +memory search results back to DeerFlow's backend-neutral shapes. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import logging +import os +import re +import threading +import time +import weakref +from collections.abc import Callable +from pathlib import Path +from typing import Any, ClassVar, Literal + +from pydantic import PrivateAttr + +# ABC contract -- the only DeerFlow import in this backend package. +from deerflow.agents.memory.manager import MemoryManager + +from .client import OpenVikingClientError, OpenVikingHttpClient +from .config import OpenVikingConfig +from .models import OpenVikingIdentity, OpenVikingMessage, OpenVikingSearchHit + +logger = logging.getLogger(__name__) + +_DEFAULT_AGENT_SCOPE = "__default__" +_SESSION_NAMESPACE = "deerflow-openviking-v1" +_SAFE_SCOPE_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") + + +class OpenVikingMemoryManager(MemoryManager): + """Remote OpenViking memory backend for passive middleware mode.""" + + supports_search: ClassVar[bool] = True + + _config: OpenVikingConfig = PrivateAttr() + _client: OpenVikingHttpClient = PrivateAttr() + _should_keep_hidden_message: Callable[[Any], bool] | None = PrivateAttr(default=None) + _session_locks: weakref.WeakValueDictionary[str, threading.RLock] = PrivateAttr(default_factory=weakref.WeakValueDictionary) + _session_locks_guard: threading.Lock = PrivateAttr(default_factory=threading.Lock) + _lifecycle: threading.Condition = PrivateAttr(default_factory=threading.Condition) + _active_operations: int = PrivateAttr(default=0) + _closed: bool = PrivateAttr(default=False) + _client_closed: bool = PrivateAttr(default=False) + + def model_post_init(self, __context: Any) -> None: + self._config = OpenVikingConfig.from_backend_config(self.backend_config) + self._client = OpenVikingHttpClient(self._config) + + @classmethod + def from_config( + cls, + backend_config: dict[str, Any] | None = None, + *, + mode: Literal["middleware", "tool"] = "middleware", + **host_hooks: Any, + ) -> OpenVikingMemoryManager: + if mode != "middleware": + raise ValueError("The OpenViking HTTP backend currently supports memory.mode='middleware' only") + instance = cls(backend_config=backend_config, mode=mode) + hook = host_hooks.get("should_keep_hidden_message") + instance._should_keep_hidden_message = hook if callable(hook) else None + return instance + + def add( + self, + thread_id: str, + messages: list[Any], + *, + agent_name: str | None = None, + user_id: str | None = None, + trace_id: str | None = None, + ) -> None: + self._write_conversation(thread_id, messages, agent_name=agent_name, user_id=user_id) + + def add_nowait( + self, + thread_id: str, + messages: list[Any], + *, + agent_name: str | None = None, + user_id: str | None = None, + ) -> None: + self._write_conversation(thread_id, messages, agent_name=agent_name, user_id=user_id) + + async def aadd( + self, + thread_id: str, + messages: list[Any], + *, + agent_name: str | None = None, + user_id: str | None = None, + trace_id: str | None = None, + ) -> None: + await asyncio.to_thread( + self.add, + thread_id, + messages, + agent_name=agent_name, + user_id=user_id, + trace_id=trace_id, + ) + + def get_context( + self, + user_id: str | None, + *, + agent_name: str | None = None, + thread_id: str | None = None, + ) -> str: + if not self._begin_operation(): + return "" + try: + try: + hits = self._search_hits( + self._config.injection_query, + top_k=self._config.search_top_k, + user_id=user_id, + agent_name=agent_name, + category=None, + thread_id=thread_id, + ) + except OpenVikingClientError: + if self._config.read_failure_policy == "raise": + raise + logger.warning("OpenViking context retrieval failed; continuing without injected memory", exc_info=True) + return "" + return _format_context(hits, max_chars=self._config.max_injection_chars) + finally: + self._end_operation() + + async def aget_context( + self, + user_id: str | None, + *, + agent_name: str | None = None, + thread_id: str | None = None, + ) -> str: + return await asyncio.to_thread( + self.get_context, + user_id, + agent_name=agent_name, + thread_id=thread_id, + ) + + def search( + self, + query: str, + top_k: int = 5, + *, + user_id: str | None = None, + agent_name: str | None = None, + category: str | None = None, + ) -> list[dict[str, Any]]: + if not query.strip(): + return [] + if not self._begin_operation(): + return [] + try: + try: + hits = self._search_hits( + query, + top_k=top_k, + user_id=user_id, + agent_name=agent_name, + category=category, + thread_id=None, + ) + except OpenVikingClientError: + if self._config.read_failure_policy == "raise": + raise + logger.warning("OpenViking memory search failed; returning no results", exc_info=True) + return [] + return [_hit_to_fact(hit) for hit in hits] + finally: + self._end_operation() + + async def asearch( + self, + query: str, + top_k: int = 5, + *, + user_id: str | None = None, + agent_name: str | None = None, + category: str | None = None, + ) -> list[dict[str, Any]]: + return await asyncio.to_thread( + self.search, + query, + top_k, + user_id=user_id, + agent_name=agent_name, + category=category, + ) + + def warm(self) -> bool | None: + if not self._begin_operation(): + return False + try: + try: + healthy = self._client.health() + except OpenVikingClientError: + if self._config.startup_policy == "fail_fast": + raise + logger.warning("OpenViking health check failed; memory will run in degraded mode", exc_info=True) + return False + if not healthy and self._config.startup_policy == "fail_fast": + raise RuntimeError("OpenViking health check returned an unhealthy response") + if not healthy: + logger.warning("OpenViking health check returned unhealthy; memory will run in degraded mode") + return healthy + finally: + self._end_operation() + + def shutdown_flush(self, timeout: float) -> bool: + """Stop new operations, drain in-flight work, then close the HTTP pool.""" + deadline = time.monotonic() + max(0.0, timeout) + with self._lifecycle: + self._closed = True + while self._active_operations: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + self._lifecycle.wait(remaining) + if self._client_closed: + return True + self._client_closed = True + self._client.close() + return True + + def _write_conversation( + self, + thread_id: str, + messages: list[Any], + *, + agent_name: str | None, + user_id: str | None, + ) -> None: + if not self._begin_operation(): + logger.warning("OpenViking write ignored after backend shutdown") + return + try: + if not thread_id: + raise ValueError("OpenViking memory write requires thread_id") + + identity = self._identity(user_id, agent_name) + session_id = _session_id(identity, thread_id) + lock = self._session_lock(session_id) + with lock: + state = self._load_state(session_id) + submitted_ids = list(state.get("submitted_message_ids", state.get("seen_message_ids", []))) + committed_ids = list(state.get("committed_message_ids", state.get("seen_message_ids", []))) + submitted = set(submitted_ids) + converted = _convert_messages(messages, self._should_keep_hidden_message) + pending = [message for message in converted if message.message_id not in submitted] + if not pending: + return + try: + self._client.ensure_session(identity, session_id) + self._client.add_messages(identity, session_id, pending) + except OpenVikingClientError: + if self._config.write_failure_policy == "raise": + raise + logger.error( + "OpenViking memory message submission failed; dropping this update (session=%s, messages=%d)", + session_id, + len(pending), + exc_info=True, + ) + return + submitted_ids.extend(message.message_id for message in pending) + state = { + "schema_version": 2, + "session_id": session_id, + "submitted_message_ids": submitted_ids[-self._config.max_seen_message_ids :], + "committed_message_ids": committed_ids[-self._config.max_seen_message_ids :], + "last_commit_task_id": state.get("last_commit_task_id"), + "last_archive_uri": state.get("last_archive_uri"), + } + self._save_state(session_id, state) + + try: + commit = self._client.commit_session(identity, session_id) + except OpenVikingClientError: + if self._config.write_failure_policy == "raise": + raise + logger.error( + "OpenViking memory commit failed; preserving submitted watermark without retry (session=%s)", + session_id, + exc_info=True, + ) + return + + state = { + **state, + "schema_version": 2, + "committed_message_ids": state["submitted_message_ids"], + "last_commit_task_id": commit.task_id, + "last_archive_uri": commit.archive_uri, + } + self._save_state(session_id, state) + finally: + self._end_operation() + + def _search_hits( + self, + query: str, + *, + top_k: int, + user_id: str | None, + agent_name: str | None, + category: str | None, + thread_id: str | None, + ) -> list[OpenVikingSearchHit]: + identity = self._identity(user_id, agent_name) + session_id = _session_id(identity, thread_id) if thread_id else None + return self._client.search( + identity, + query, + top_k=max(1, min(top_k, 100)), + category=category, + session_id=session_id, + ) + + def _identity(self, user_id: str | None, agent_name: str | None) -> OpenVikingIdentity: + raw_user = str(user_id or "anonymous") + agent_scope = _canonical_agent_scope(agent_name) + # OpenViking trusted identity must be a safe path segment. Hashing the + # DeerFlow scope also prevents raw emails/usernames from leaving the + # Gateway and gives each agent a hard-isolated memory namespace. + digest = hashlib.sha256(f"{self._config.account}\0{raw_user}\0{agent_scope}".encode()).hexdigest() + return OpenVikingIdentity(account=self._config.account, user=f"df_{digest[:40]}") + + def _session_lock(self, session_id: str) -> threading.RLock: + with self._session_locks_guard: + return self._session_locks.setdefault(session_id, threading.RLock()) + + def _begin_operation(self) -> bool: + with self._lifecycle: + if self._closed: + return False + self._active_operations += 1 + return True + + def _end_operation(self) -> None: + with self._lifecycle: + self._active_operations -= 1 + if self._active_operations == 0: + self._lifecycle.notify_all() + + def _state_path(self, session_id: str) -> Path: + root = Path(self._config.storage_path or ".") / "openviking" / "sessions" + return root / f"{session_id}.json" + + def _load_state(self, session_id: str) -> dict[str, Any]: + path = self._state_path(session_id) + try: + value = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return {} + except (OSError, ValueError): + logger.warning("Ignoring unreadable OpenViking session watermark: %s", path, exc_info=True) + return {} + return value if isinstance(value, dict) else {} + + def _save_state(self, session_id: str, state: dict[str, Any]) -> None: + path = self._state_path(session_id) + path.parent.mkdir(parents=True, exist_ok=True) + temp_path = path.with_suffix(f".{os.getpid()}.{threading.get_ident()}.tmp") + try: + temp_path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") + os.replace(temp_path, path) + finally: + try: + temp_path.unlink(missing_ok=True) + except OSError: + logger.debug("Failed to remove OpenViking watermark temp file: %s", temp_path, exc_info=True) + + +def _canonical_agent_scope(agent_name: str | None) -> str: + if agent_name is None: + return _DEFAULT_AGENT_SCOPE + value = str(agent_name).strip().lower() + if value == _DEFAULT_AGENT_SCOPE or not _SAFE_SCOPE_RE.fullmatch(value): + raise ValueError(f"Invalid OpenViking agent scope: {agent_name!r}") + return value + + +def _session_id(identity: OpenVikingIdentity, thread_id: str) -> str: + digest = hashlib.sha256(f"{_SESSION_NAMESPACE}\0{identity.account}\0{identity.user}\0{thread_id}".encode()).hexdigest() + return f"df_{digest[:48]}" + + +def _convert_messages( + messages: list[Any], + should_keep_hidden_message: Callable[[Any], bool] | None, +) -> list[OpenVikingMessage]: + converted: list[OpenVikingMessage] = [] + for index, message in enumerate(messages): + role = _message_role(message) + if role not in {"user", "assistant"}: + continue + additional_kwargs = _message_value(message, "additional_kwargs", {}) + if not isinstance(additional_kwargs, dict): + additional_kwargs = {} + if additional_kwargs.get("hide_from_ui") and not (should_keep_hidden_message and should_keep_hidden_message(additional_kwargs)): + continue + tool_calls = _message_value(message, "tool_calls", []) + if role == "assistant" and tool_calls: + continue + content = _text_content(_message_value(message, "content", "")) + if not content.strip(): + continue + native_id = _message_value(message, "id", None) + stable_id = str(native_id) if native_id else hashlib.sha256(f"{role}\0{index}\0{content}".encode()).hexdigest() + converted.append(OpenVikingMessage(message_id=f"df_{stable_id}", role=role, content=content.strip())) + return converted + + +def _message_role(message: Any) -> str | None: + value = _message_value(message, "type", None) or _message_value(message, "role", None) + if value in {"human", "user"}: + return "user" + if value in {"ai", "assistant"}: + return "assistant" + name = type(message).__name__.lower() + if "human" in name: + return "user" + if "ai" in name: + return "assistant" + return None + + +def _message_value(message: Any, key: str, default: Any) -> Any: + return message.get(key, default) if isinstance(message, dict) else getattr(message, key, default) + + +def _text_content(content: Any) -> str: + if isinstance(content, str): + return content + if not isinstance(content, list): + return str(content) if content is not None else "" + parts: list[str] = [] + for block in content: + if isinstance(block, str): + parts.append(block) + elif isinstance(block, dict) and block.get("type") in {"text", "input_text", "output_text"}: + text = block.get("text") + if text: + parts.append(str(text)) + return "\n".join(parts) + + +def _hit_content(hit: OpenVikingSearchHit) -> str: + return (hit.overview or hit.abstract).strip() + + +def _hit_to_fact(hit: OpenVikingSearchHit) -> dict[str, Any]: + return { + "id": hit.uri, + "content": _hit_content(hit), + "category": hit.category or "memory", + "confidence": hit.score, + "source": hit.uri, + "score": hit.score, + } + + +def _format_context(hits: list[OpenVikingSearchHit], *, max_chars: int) -> str: + lines: list[str] = [] + seen: set[str] = set() + for hit in hits: + content = " ".join(_hit_content(hit).split()) + key = content.casefold() + if not content or key in seen: + continue + seen.add(key) + line = f"- [{hit.category or 'memory'}] {content}" + candidate = "\n".join([*lines, line]) + if len(candidate) > max_chars: + remaining = max_chars - len("\n".join(lines)) - (1 if lines else 0) + if remaining > 16: + lines.append(f"{line[: max(0, remaining - 1)]}…") + break + lines.append(line) + return "\n".join(lines) diff --git a/backend/tests/blocking_io/test_openviking_memory_backend.py b/backend/tests/blocking_io/test_openviking_memory_backend.py new file mode 100644 index 000000000..78ebeede1 --- /dev/null +++ b/backend/tests/blocking_io/test_openviking_memory_backend.py @@ -0,0 +1,91 @@ +"""Regression anchors: OpenViking async memory methods must not block the loop.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest +from langchain_core.messages import AIMessage, HumanMessage + +from deerflow.agents.memory.backends.openviking.models import OpenVikingCommitResult, OpenVikingSearchHit +from deerflow.agents.memory.backends.openviking.openviking_manager import OpenVikingMemoryManager + + +class _BlockingProbeClient: + """Perform real file IO so Blockbuster can detect missing async offload.""" + + def __init__(self, probe_path: Path): + self._probe_path = probe_path + + def _probe(self) -> None: + self._probe_path.write_text("probe", encoding="utf-8") + + def ensure_session(self, identity, session_id) -> None: + self._probe() + + def add_messages(self, identity, session_id, messages) -> int: + self._probe() + return len(messages) + + def commit_session(self, identity, session_id) -> OpenVikingCommitResult: + self._probe() + return OpenVikingCommitResult(status="accepted", task_id="task-1", archive_uri=None, archived=True) + + def search( + self, + identity, + query: str, + *, + top_k: int, + category: str | None = None, + session_id: str | None = None, + ) -> list[OpenVikingSearchHit]: + self._probe() + return [ + OpenVikingSearchHit( + uri="viking://user/memories/preferences/test.md", + context_type="memory", + category="preferences", + score=0.9, + abstract="Prefers concise answers.", + overview=None, + match_reason="", + ) + ] + + def close(self) -> None: + pass + + +def _manager(tmp_path: Path) -> OpenVikingMemoryManager: + manager = OpenVikingMemoryManager.from_config( + { + "base_url": "http://openviking:1933", + "storage_path": str(tmp_path), + "auth_mode": "trusted", + "account": "deerflow", + "startup_policy": "warn", + } + ) + manager._client = _BlockingProbeClient(tmp_path / "probe.txt") # type: ignore[assignment] + return manager + + +@pytest.mark.asyncio +async def test_async_openviking_operations_do_not_block_event_loop(tmp_path: Path) -> None: + manager = _manager(tmp_path) + messages: list[Any] = [HumanMessage("hello", id="h1"), AIMessage("hi", id="a1")] + + await manager.aadd("thread-1", messages, user_id="alice") + assert await manager.aget_context("alice") == "- [preferences] Prefers concise answers." + assert await manager.asearch("answer style", user_id="alice") == [ + { + "id": "viking://user/memories/preferences/test.md", + "content": "Prefers concise answers.", + "category": "preferences", + "confidence": 0.9, + "source": "viking://user/memories/preferences/test.md", + "score": 0.9, + } + ] diff --git a/backend/tests/test_openviking_memory_backend.py b/backend/tests/test_openviking_memory_backend.py new file mode 100644 index 000000000..329a77f1e --- /dev/null +++ b/backend/tests/test_openviking_memory_backend.py @@ -0,0 +1,368 @@ +from __future__ import annotations + +import gc +import json +import threading +import weakref +from pathlib import Path +from typing import Any + +import httpx +import pytest +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage + +from deerflow.agents.memory.backends.openviking.client import OpenVikingAuthenticationError, OpenVikingHttpClient, OpenVikingUnavailableError +from deerflow.agents.memory.backends.openviking.config import OpenVikingConfig +from deerflow.agents.memory.backends.openviking.models import ( + OpenVikingCommitResult, + OpenVikingIdentity, + OpenVikingMessage, + OpenVikingSearchHit, +) +from deerflow.agents.memory.backends.openviking.openviking_manager import OpenVikingMemoryManager +from deerflow.agents.memory.manager import _scan_backends, reset_memory_manager + + +def _backend_config(tmp_path: Path, **overrides: Any) -> dict[str, Any]: + config: dict[str, Any] = { + "base_url": "http://openviking:1933", + "storage_path": str(tmp_path), + "auth_mode": "trusted", + "account": "deerflow", + "startup_policy": "warn", + "retrieval": {"top_k": 4, "max_injection_chars": 1000}, + } + config.update(overrides) + return config + + +def test_config_parses_nested_fields_and_rejects_unknown(tmp_path: Path) -> None: + config = OpenVikingConfig.from_backend_config( + _backend_config( + tmp_path, + retrieval={"top_k": 7, "score_threshold": 0.4, "max_injection_chars": 2048}, + failure_policy={"read": "raise", "write": "raise"}, + ) + ) + + assert config.search_top_k == 7 + assert config.score_threshold == 0.4 + assert config.read_failure_policy == "raise" + + with pytest.raises(ValueError, match="Unknown OpenViking"): + OpenVikingConfig.from_backend_config(_backend_config(tmp_path, typo=True)) + + +def test_config_rejects_dev_auth_without_explicit_opt_in(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="allow_insecure_dev"): + OpenVikingConfig.from_backend_config(_backend_config(tmp_path, auth_mode="dev")) + + +def test_backend_is_discovered_by_registered_name() -> None: + reset_memory_manager() + assert _scan_backends()["openviking"] is OpenVikingMemoryManager + + +def test_http_client_sends_trusted_identity_and_maps_search(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OPENVIKING_API_KEY", "secret") + requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if request.url.path == "/api/v1/search/find": + return httpx.Response( + 200, + json={ + "status": "ok", + "result": { + "memories": [ + { + "uri": "viking://user/memories/preferences/editor.md", + "context_type": "memory", + "category": "preferences", + "score": 0.91, + "abstract": "Uses Vim.", + "overview": None, + } + ], + "resources": [], + "skills": [], + "total": 1, + }, + }, + ) + raise AssertionError(f"unexpected path: {request.url.path}") + + config = OpenVikingConfig.from_backend_config(_backend_config(tmp_path)) + client = OpenVikingHttpClient(config, transport=httpx.MockTransport(handler)) + identity = OpenVikingIdentity(account="deerflow", user="df_user") + + hits = client.search(identity, "editor preference", top_k=3) + + assert [hit.abstract for hit in hits] == ["Uses Vim."] + assert requests[0].headers["X-OpenViking-Account"] == "deerflow" + assert requests[0].headers["X-OpenViking-User"] == "df_user" + assert requests[0].headers["X-API-Key"] == "secret" + assert json.loads(requests[0].content)["target_uri"] == "viking://user/memories" + assert json.loads(requests[0].content)["context_type"] == "memory" + + +def test_http_client_maps_authentication_error(tmp_path: Path) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 401, + json={"status": "error", "error": {"code": "UNAUTHENTICATED", "message": "bad key"}}, + ) + + config = OpenVikingConfig.from_backend_config(_backend_config(tmp_path)) + client = OpenVikingHttpClient(config, transport=httpx.MockTransport(handler)) + + with pytest.raises(OpenVikingAuthenticationError) as exc_info: + client.ensure_session(OpenVikingIdentity(account="deerflow", user="df_user"), "session") + assert exc_info.value.code == "UNAUTHENTICATED" + + +class _FakeClient: + def __init__(self) -> None: + self.ensured: list[tuple[OpenVikingIdentity, str]] = [] + self.added: list[tuple[OpenVikingIdentity, str, list[OpenVikingMessage]]] = [] + self.committed: list[tuple[OpenVikingIdentity, str]] = [] + self.searches: list[tuple[OpenVikingIdentity, str, int, str | None]] = [] + self.closed = False + + def ensure_session(self, identity: OpenVikingIdentity, session_id: str) -> None: + self.ensured.append((identity, session_id)) + + def add_messages( + self, + identity: OpenVikingIdentity, + session_id: str, + messages: list[OpenVikingMessage], + ) -> int: + self.added.append((identity, session_id, messages)) + return len(messages) + + def commit_session(self, identity: OpenVikingIdentity, session_id: str) -> OpenVikingCommitResult: + self.committed.append((identity, session_id)) + return OpenVikingCommitResult( + status="accepted", + task_id="task-1", + archive_uri="viking://user/sessions/session/history/archive_001", + archived=True, + ) + + def search( + self, + identity: OpenVikingIdentity, + query: str, + *, + top_k: int, + category: str | None = None, + session_id: str | None = None, + ) -> list[OpenVikingSearchHit]: + self.searches.append((identity, query, top_k, category)) + return [ + OpenVikingSearchHit( + uri="viking://user/memories/preferences/editor.md", + context_type="memory", + category="preferences", + score=0.9, + abstract="User prefers concise answers.", + overview=None, + match_reason="", + ) + ] + + def health(self) -> bool: + return True + + def close(self) -> None: + self.closed = True + + +def _manager(tmp_path: Path) -> tuple[OpenVikingMemoryManager, _FakeClient]: + manager = OpenVikingMemoryManager.from_config(_backend_config(tmp_path)) + fake = _FakeClient() + manager._client = fake # type: ignore[assignment] + return manager, fake + + +def test_manager_filters_messages_commits_and_deduplicates(tmp_path: Path) -> None: + manager, client = _manager(tmp_path) + messages = [ + SystemMessage("system"), + HumanMessage("Remember that I prefer Vim.", id="human-1"), + AIMessage("", tool_calls=[{"name": "search", "args": {}, "id": "call-1", "type": "tool_call"}]), + ToolMessage("tool output", tool_call_id="call-1"), + AIMessage("I will remember that.", id="ai-1"), + ] + + manager.add("thread-1", messages, user_id="alice", agent_name="research") + manager.add("thread-1", messages, user_id="alice", agent_name="research") + + assert len(client.added) == 1 + assert [(message.role, message.content) for message in client.added[0][2]] == [ + ("user", "Remember that I prefer Vim."), + ("assistant", "I will remember that."), + ] + assert len(client.committed) == 1 + watermark = next((tmp_path / "openviking" / "sessions").glob("*.json")) + state = json.loads(watermark.read_text(encoding="utf-8")) + assert state["last_commit_task_id"] == "task-1" + assert state["submitted_message_ids"] == state["committed_message_ids"] + + +def test_manager_does_not_resubmit_messages_after_failed_commit(tmp_path: Path) -> None: + manager, client = _manager(tmp_path) + messages = [HumanMessage("hello", id="h1"), AIMessage("hi", id="a1")] + original_commit = client.commit_session + commit_attempts = 0 + + def fail_once(identity: OpenVikingIdentity, session_id: str) -> OpenVikingCommitResult: + nonlocal commit_attempts + commit_attempts += 1 + if commit_attempts == 1: + raise OpenVikingUnavailableError("session.commit", "temporary failure") + return original_commit(identity, session_id) + + client.commit_session = fail_once # type: ignore[method-assign] + + manager.add("thread-1", messages, user_id="alice", agent_name="research") + + watermark = next((tmp_path / "openviking" / "sessions").glob("*.json")) + failed_state = json.loads(watermark.read_text(encoding="utf-8")) + assert failed_state["submitted_message_ids"] == ["df_h1", "df_a1"] + assert failed_state["committed_message_ids"] == [] + + manager.add("thread-1", messages, user_id="alice", agent_name="research") + + assert len(client.added) == 1 + assert commit_attempts == 1 + + messages.append(HumanMessage("new information", id="h2")) + manager.add("thread-1", messages, user_id="alice", agent_name="research") + + assert len(client.added) == 2 + assert [message.message_id for message in client.added[1][2]] == ["df_h2"] + assert commit_attempts == 2 + recovered_state = json.loads(watermark.read_text(encoding="utf-8")) + assert recovered_state["submitted_message_ids"] == recovered_state["committed_message_ids"] + assert recovered_state["last_commit_task_id"] == "task-1" + + +def test_manager_identity_is_stable_and_agent_isolated(tmp_path: Path) -> None: + manager, client = _manager(tmp_path) + messages = [HumanMessage("hello", id="h1"), AIMessage("hi", id="a1")] + + manager.add("thread-1", messages, user_id="alice", agent_name="research") + manager.add("thread-1", messages, user_id="alice", agent_name="coding") + + identities = [entry[0].user for entry in client.added] + session_ids = [entry[1] for entry in client.added] + assert identities[0] != identities[1] + assert session_ids[0] != session_ids[1] + assert all(value.startswith("df_") for value in identities) + + +def test_manager_search_and_context_map_remote_results(tmp_path: Path) -> None: + manager, client = _manager(tmp_path) + + results = manager.search("answer style", user_id="alice", agent_name="research") + context = manager.get_context("alice", agent_name="research") + + assert results == [ + { + "id": "viking://user/memories/preferences/editor.md", + "content": "User prefers concise answers.", + "category": "preferences", + "confidence": 0.9, + "source": "viking://user/memories/preferences/editor.md", + "score": 0.9, + } + ] + assert context == "- [preferences] User prefers concise answers." + assert client.searches[1][1] == manager._config.injection_query + + +def test_manager_rejects_tool_mode(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="middleware"): + OpenVikingMemoryManager.from_config(_backend_config(tmp_path), mode="tool") + + +def test_manager_session_locks_do_not_accumulate(tmp_path: Path) -> None: + manager, _ = _manager(tmp_path) + lock = manager._session_lock("session-1") + lock_ref = weakref.ref(lock) + + assert len(manager._session_locks) == 1 + del lock + gc.collect() + + assert lock_ref() is None + assert len(manager._session_locks) == 0 + + +@pytest.mark.asyncio +async def test_manager_async_methods_offload_sync_operations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + manager, _ = _manager(tmp_path) + event_loop_thread = threading.get_ident() + worker_threads: list[int] = [] + + def fake_add(self: OpenVikingMemoryManager, *args: Any, **kwargs: Any) -> None: + worker_threads.append(threading.get_ident()) + + def fake_get_context(self: OpenVikingMemoryManager, *args: Any, **kwargs: Any) -> str: + worker_threads.append(threading.get_ident()) + return "context" + + def fake_search(self: OpenVikingMemoryManager, *args: Any, **kwargs: Any) -> list[dict[str, Any]]: + worker_threads.append(threading.get_ident()) + return [{"id": "memory-1"}] + + monkeypatch.setattr(OpenVikingMemoryManager, "add", fake_add) + monkeypatch.setattr(OpenVikingMemoryManager, "get_context", fake_get_context) + monkeypatch.setattr(OpenVikingMemoryManager, "search", fake_search) + + await manager.aadd("thread-1", [], user_id="alice") + assert await manager.aget_context("alice") == "context" + assert await manager.asearch("query", user_id="alice") == [{"id": "memory-1"}] + + assert len(worker_threads) == 3 + assert all(thread_id != event_loop_thread for thread_id in worker_threads) + + +def test_manager_shutdown_waits_for_in_flight_write(tmp_path: Path) -> None: + manager, client = _manager(tmp_path) + write_started = threading.Event() + release_write = threading.Event() + original_add = client.add_messages + + def blocking_add( + identity: OpenVikingIdentity, + session_id: str, + messages: list[OpenVikingMessage], + ) -> int: + write_started.set() + assert release_write.wait(2) + return original_add(identity, session_id, messages) + + client.add_messages = blocking_add # type: ignore[method-assign] + write_thread = threading.Thread( + target=manager.add, + args=("thread-1", [HumanMessage("hello", id="h1")]), + kwargs={"user_id": "alice"}, + ) + write_thread.start() + assert write_started.wait(1) + + assert manager.shutdown_flush(0.01) is False + assert client.closed is False + manager.add("thread-2", [HumanMessage("ignored", id="h2")], user_id="alice") + assert len(client.ensured) == 1 + + release_write.set() + write_thread.join(2) + assert not write_thread.is_alive() + + assert manager.shutdown_flush(1) is True + assert client.closed is True diff --git a/config.example.yaml b/config.example.yaml index 1dd9ac35d..18b9a97a5 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1618,7 +1618,7 @@ summarization: # enabled - Master switch for the memory mechanism (call-site gate) # injection_enabled - Whether to inject memory into the system prompt (call-site gate) # shutdown_flush_timeout_seconds - Hard budget (s) to drain pending updates on Gateway graceful shutdown (default: 30) -# manager_class - Backend selector: registered name (deermem/noop) or dotted path +# manager_class - Backend selector: registered name (deermem/noop/openviking) or dotted path # backend_config - Backend-private config dict (passthrough; each backend self-interprets) # # DeerMem-private fields live under ``backend_config`` (NOT at the memory: top level): @@ -1671,6 +1671,27 @@ memory: mode: middleware # Backend-private config (a dict), passed verbatim to the backend __init__. # Each backend self-interprets it (DeerMem parses it into DeerMemConfig). + # + # OpenViking HTTP example (replace this DeerMem backend_config block when + # manager_class is openviking; OpenViking currently supports middleware mode): + # + # backend_config: + # base_url: http://openviking:1933 + # auth_mode: trusted + # account: deerflow + # api_key_env: OPENVIKING_API_KEY + # startup_policy: fail_fast + # failure_policy: + # read: fail_open + # write: log_and_drop + # retrieval: + # top_k: 8 + # score_threshold: 0.25 + # max_injection_chars: 12000 + # + # For a host-installed OpenViking used by Docker DeerFlow, set base_url to + # http://host.docker.internal:1933 and allow_insecure_http: true. The bundled + # optional Compose overlay uses the internal http://openviking:1933 address. backend_config: storage_path: "" # empty = deer-flow base_dir (factory injects absolute runtime_home); a non-empty path is the root DIRECTORY (per-user memory under {storage_path}/users/{uid}/memory.json) storage_class: file # file (default) or a dotted MemoryStorage class path; invalid persistent backends fail fast diff --git a/docker/docker-compose-dev.yaml b/docker/docker-compose-dev.yaml index 3713f1195..a51f432c9 100644 --- a/docker/docker-compose-dev.yaml +++ b/docker/docker-compose-dev.yaml @@ -208,8 +208,8 @@ services: - PROVISIONER_API_KEY=${PROVISIONER_API_KEY:-} # Proxy values (HTTP_PROXY/HTTPS_PROXY/ALL_PROXY) are inherited from ../.env via env_file. # Only NO_PROXY is declared here so internal service hostnames are always exempt from the proxy. - - NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal - - no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal + - NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal + - no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal env_file: - ../.env extra_hosts: diff --git a/docker/docker-compose.openviking.yaml b/docker/docker-compose.openviking.yaml new file mode 100644 index 000000000..b408f2f4a --- /dev/null +++ b/docker/docker-compose.openviking.yaml @@ -0,0 +1,42 @@ +# Optional OpenViking memory service overlay. +# +# First-time initialization: +# docker compose -f docker/docker-compose.yaml \ +# -f docker/docker-compose.openviking.yaml up -d openviking +# docker exec -it deer-flow-openviking openviking-server init +# +# Full stack: +# docker compose -f docker/docker-compose.yaml \ +# -f docker/docker-compose.openviking.yaml up -d --build + +services: + openviking: + image: ${OPENVIKING_IMAGE:-ghcr.io/volcengine/openviking:latest} + container_name: deer-flow-openviking + command: ["--without-bot"] + ports: + - "${OPENVIKING_PORT:-1933}:1933" + volumes: + - openviking-data:/app/.openviking + networks: + - deer-flow + restart: unless-stopped + healthcheck: + test: + [ + "CMD", + "python", + "-c", + "import urllib.request; urllib.request.urlopen('http://127.0.0.1:1933/health', timeout=3)", + ] + interval: 10s + timeout: 5s + retries: 12 + + gateway: + depends_on: + openviking: + condition: service_healthy + +volumes: + openviking-data: diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index a64c13398..edc7e353c 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -131,8 +131,8 @@ services: - DEER_FLOW_SANDBOX_HOST=host.docker.internal # Proxy values (HTTP_PROXY/HTTPS_PROXY/ALL_PROXY) are inherited from ../.env via env_file. # Only NO_PROXY is declared here so internal service hostnames are always exempt from the proxy. - - NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal - - no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal + - NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal + - no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal env_file: - ../.env extra_hosts: diff --git a/docs/OPENVIKING.md b/docs/OPENVIKING.md new file mode 100644 index 000000000..2412d02f0 --- /dev/null +++ b/docs/OPENVIKING.md @@ -0,0 +1,212 @@ +# OpenViking memory backend + +DeerFlow can use a remote OpenViking server as an optional long-term memory +backend. The integration is a pluggable `MemoryManager`; DeerMem remains the +default and the agent/Gateway runtime does not import the OpenViking Python +runtime. + +## Supported behavior + +- HTTP connection to an independent OpenViking server. +- Passive `memory.mode: middleware` capture after each completed turn. +- OpenViking Session commit and asynchronous memory extraction. +- Automatic prompt injection from OpenViking memory search. +- Explicit search through the backend-neutral `MemoryManager.search` API. +- Hard isolation by hashing each DeerFlow `(user_id, agent_name)` scope into a + separate OpenViking trusted user identity. +- Local message watermarks under DeerFlow's runtime home to avoid resubmitting + the same conversation history in a single-Gateway deployment. + +The current backend does not implement DeerMem fact CRUD, import/export, or the +Settings memory document. Keep `mode: middleware`; tool mode is rejected +because its add/update/delete tools require fact CRUD. + +## OpenViking requirements + +OpenViking must be configured with: + +- a VLM provider; +- an embedding provider; +- persistent workspace storage; +- `server.auth_mode: trusted`; +- a non-empty `server.root_api_key` when exposed beyond localhost. + +DeerFlow passes trusted `X-OpenViking-Account` and +`X-OpenViking-User` headers. Do not expose a trusted-mode OpenViking endpoint +directly to untrusted clients. + +## Configure DeerFlow + +Put the trusted OpenViking key in the repository root `.env`: + +```dotenv +OPENVIKING_API_KEY=replace-with-the-same-root-api-key +``` + +Replace the `memory` section in `config.yaml` with: + +```yaml +memory: + enabled: true + injection_enabled: true + shutdown_flush_timeout_seconds: 30 + manager_class: openviking + mode: middleware + backend_config: + base_url: http://openviking:1933 + auth_mode: trusted + account: deerflow + api_key_env: OPENVIKING_API_KEY + startup_policy: fail_fast + failure_policy: + read: fail_open + write: log_and_drop + retrieval: + top_k: 8 + score_threshold: 0.25 + max_injection_chars: 12000 +``` + +For a locally installed DeerFlow process, use +`http://127.0.0.1:1933`. For a DeerFlow container connecting to OpenViking on +the host, use `http://host.docker.internal:1933` and set +`allow_insecure_http: true`. + +## Docker first-time startup + +Create the standard DeerFlow local files if they no longer exist: + +```bash +make config +cp .env.example .env +cp frontend/.env.example frontend/.env +``` + +Set at least the normal DeerFlow secrets in `.env`, including +`BETTER_AUTH_SECRET`, model provider credentials, and +`OPENVIKING_API_KEY`. + +The production Compose file expects the same path variables normally exported +by `scripts/deploy.sh`. Export them before using the OpenViking overlay +directly: + +```bash +export DEER_FLOW_CONFIG_PATH="$PWD/config.yaml" +export DEER_FLOW_EXTENSIONS_CONFIG_PATH="$PWD/extensions_config.json" +export DEER_FLOW_HOME="$PWD/backend/.deer-flow" +export DEER_FLOW_REPO_ROOT="$PWD" +``` + +Start only OpenViking: + +```bash +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + up -d openviking +``` + +Initialize it interactively: + +```bash +docker exec -it deer-flow-openviking openviking-server init +``` + +Choose trusted authentication, configure the same root API key stored in +`OPENVIKING_API_KEY`, and configure VLM and embedding providers. Validate the +configuration: + +```bash +docker exec -it deer-flow-openviking openviking-server doctor +docker restart deer-flow-openviking +curl http://localhost:1933/health +``` + +Then start DeerFlow with the same overlay: + +```bash +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + up -d --build +``` + +Open DeerFlow at and OpenViking Studio at +. + +## Routine Docker operations + +```bash +# Logs +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + logs -f gateway openviking + +# Stop containers but retain data +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + down + +# Restart +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + up -d + +# Pull a newer OpenViking image and recreate +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + pull openviking +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + up -d openviking +``` + +Do not add `-v` to `docker compose down` unless you intentionally want to +delete Redis and OpenViking persistent volumes. + +## Local-process startup + +Run OpenViking separately and verify: + +```bash +openviking-server doctor +openviking-server +curl http://127.0.0.1:1933/health +``` + +Set `base_url: http://127.0.0.1:1933`, then start DeerFlow normally: + +```bash +make doctor +make dev +``` + +The DeerFlow entrypoint is . + +## Failure behavior + +- Invalid backend configuration fails loudly; DeerFlow never silently writes + to DeerMem instead. +- With `read: fail_open`, retrieval failures produce no injected memory and + the main agent continues. +- With `write: log_and_drop`, a failed OpenViking commit is logged without + failing an already generated assistant response. +- Once a message batch is accepted, DeerFlow persists a submitted-message + watermark before committing the Session. If commit then fails, later updates + do not resubmit those messages or retry the ambiguous commit; a future batch + can commit the still-open Session together with new messages. +- OpenViking commit is eventually consistent: accepting a commit archives the + messages immediately, while summary and memory extraction finish in a + background task. +- Graceful shutdown stops admitting new memory operations, waits up to + `shutdown_flush_timeout_seconds` for active reads and writes, and closes the + shared HTTP client only after they drain. + +For deployments where a lost memory update is unacceptable, a durable outbox +is still required; the initial plugin intentionally does not claim +at-least-once delivery. diff --git a/docs/plans/OPENVIKING_HTTP_MEMORY_INTEGRATION.md b/docs/plans/OPENVIKING_HTTP_MEMORY_INTEGRATION.md new file mode 100644 index 000000000..aceaf1ec4 --- /dev/null +++ b/docs/plans/OPENVIKING_HTTP_MEMORY_INTEGRATION.md @@ -0,0 +1,1037 @@ +# OpenViking HTTP Memory Backend 接入实现方案 + +> 状态：设计稿 +> 范围：通过 HTTP 将 OpenViking 接入为 DeerFlow 的可选 `MemoryManager` 后端 +> 默认行为：DeerMem 继续作为默认后端，OpenViking 需要显式启用 +> 非目标：本文不包含代码实现、DeerMem 存量数据迁移、OpenViking 服务端开发 + +## 1. 背景 + +DeerFlow 已提供可插拔的 `MemoryManager` 接口。内置的 DeerMem 负责从对话中提取事实、持久化事实并向后续对话注入记忆；`noop` 则提供最小空实现。后端发现器会扫描： + +```text +backend/packages/harness/deerflow/agents/memory/backends// +``` + +只要子包导出 `MANAGER_CLASS`，便可通过以下配置选择： + +```yaml +memory: + manager_class: + backend_config: {} +``` + +OpenViking 本身提供 Session、Memory Extraction、语义检索、多租户和后台任务能力。它的 Session 生命周期为： + +```text +创建 Session + -> 添加 user/assistant 消息 + -> commit + -> 同步归档消息 + -> 后台生成摘要并提取长期记忆 + -> 后续 search/find 检索记忆 +``` + +因此，本方案不在 DeerFlow 内重复实现 OpenViking 已具备的抽取、去重、合并和向量索引逻辑，而是在 DeerFlow 的 `MemoryManager` 边界增加一个 HTTP 适配器。 + +## 2. 目标 + +首版实现以下闭环： + +1. DeerFlow 能通过配置选择 `openviking` 记忆后端。 +2. 每轮对话完成后，DeerFlow 将本轮有效消息提交到对应的 OpenViking Session。 +3. OpenViking 完成归档及异步长期记忆提取。 +4. 新一轮对话开始时，DeerFlow 按当前用户、Agent 和线程范围检索相关记忆。 +5. 检索结果被格式化为有长度上限的纯文本，通过现有 prompt 注入路径提供给 Agent。 +6. OpenViking 短暂不可用时，DeerFlow 主对话链路可以按配置降级，并留下可观测错误。 +7. 不同 DeerFlow 用户之间保持强隔离；不同 Agent 的专属记忆不会互相污染。 + +## 3. 非目标 + +以下内容不进入首版： + +- 不替换 DeerFlow 的 thread、checkpoint、run event 等持久化系统。 +- 不把 OpenViking 嵌入 DeerFlow Gateway 进程。 +- 不修改 OpenViking 服务端。 +- 不迁移现有 DeerMem Markdown/JSON 数据。 +- 不接管 DeerFlow 的资源库、上传文件或 Skill 存储。 +- 不保证现有 Memory 页面上的 fact 新增、编辑、删除按钮可用于 OpenViking。 +- 不将 OpenViking 原生多类型记忆强行压平成 DeerMem 的事实模型。 +- 不在首版启用 `memory.mode: tool`。 +- 不在 DeerFlow 内再次调用 LLM 进行事实抽取、去重或合并。 + +## 4. 总体架构 + +```text +┌──────────────────────── DeerFlow Gateway ────────────────────────┐ +│ │ +│ MemoryMiddleware / summarization hook / prompt injection │ +│ │ │ +│ ▼ │ +│ MemoryManager contract │ +│ │ │ +│ ▼ │ +│ OpenVikingMemoryManager adapter │ +│ │ │ │ +│ ▼ ▼ │ +│ scope/message mapping result formatting │ +│ │ ▲ │ +│ └──────── OpenVikingHttpClient ────────────────┐│ +└─────────────────────────────────────────────────────────────────┼┘ + │ HTTP + ▼ +┌──────────────────────── OpenViking Server ───────────────────────┐ +│ Sessions │ Memory Extraction │ Search │ Tasks │ Tenant Storage │ +└──────────────────────────────────────────────────────────────────┘ +``` + +OpenViking 作为独立服务运行。DeerFlow 只依赖其 HTTP API，不依赖 OpenViking 的 Python 嵌入式运行时。 + +选择 HTTP 模式的原因： + +- 避免把 OpenViking 的模型、向量库、Rust/C++ 扩展和后台任务依赖带入 Gateway。 +- OpenViking 可独立升级、扩容、监控和持久化。 +- 一个 OpenViking 服务可以服务多个 DeerFlow Gateway 实例。 +- HTTP 边界更容易进行超时、熔断、认证和契约测试。 +- 降低 DeerFlow 与 OpenViking Python SDK 版本的耦合。 + +## 5. 代码布局 + +新增： + +```text +backend/packages/harness/deerflow/agents/memory/backends/openviking/ +├── __init__.py +├── config.py +├── client.py +├── models.py +└── openviking_manager.py +``` + +建议职责如下。 + +### 5.1 `__init__.py` + +仅负责注册后端： + +```python +from .openviking_manager import OpenVikingMemoryManager + +MANAGER_CLASS = OpenVikingMemoryManager +``` + +目录名、注册名和配置值统一使用 `openviking`。 + +### 5.2 `config.py` + +定义并验证 OpenViking 私有配置，禁止读取 DeerFlow 全局配置单例。所有值均从 `backend_config` 或明确的环境变量引用取得。 + +职责包括： + +- 校验 `base_url`； +- 校验认证模式； +- 解析 API key 环境变量名； +- 校验 timeout、重试次数、检索数量和注入长度； +- 为日志生成脱敏后的配置摘要； +- 拒绝未知字段，避免拼写错误静默失效。 + +### 5.3 `models.py` + +定义适配器内部数据模型，隔离 OpenViking HTTP 响应格式变化，例如： + +```text +OpenVikingMessage +OpenVikingSearchHit +OpenVikingCommitResult +OpenVikingTaskStatus +OpenVikingErrorBody +``` + +这些模型不暴露到 DeerFlow 公共 API，也不导入 OpenViking SDK 类型。 + +### 5.4 `client.py` + +实现一个薄 HTTP Client，只封装接入所需的稳定 API： + +- `GET /health` +- 创建或获取 Session +- 批量添加 Session 消息 +- commit Session +- 查询后台 Task +- `search`/`find` 检索 + +HTTP URL、请求头、响应 envelope、timeout、重试和错误转换全部集中在此文件。`OpenVikingMemoryManager` 不直接拼接 URL。 + +### 5.5 `openviking_manager.py` + +实现 `MemoryManager`： + +- `from_config` +- `add` +- `add_nowait` +- `get_context` +- `search` +- `shutdown_flush` +- 可选 `warm` + +首版不实现的管理方法继承基类的 `NotImplementedError`。 + +## 6. DeerFlow 与 OpenViking 的概念映射 + +### 6.1 租户与用户 + +推荐映射： + +| DeerFlow | OpenViking | 说明 | +|---|---|---| +| 一个 DeerFlow 部署或业务工作区 | `account` | 外层租户边界 | +| `user_id` | `user` | 用户记忆和 Session 的隔离边界 | +| `agent_name` | Agent scope/tag/安全 URI 段 | 同一用户下的 Agent 专属范围 | +| `thread_id` | Session 稳定键的一部分 | 对话来源，不单独作为全局 Session ID | + +OpenViking 的 `user` 是强数据边界。每个请求必须显式带上当前 DeerFlow 用户身份，不能依赖 OpenViking 的 `default/default` 开发身份。 + +首版推荐使用 OpenViking `trusted` 认证模式，由 DeerFlow Gateway 向内部 OpenViking 服务传递： + +```text +X-OpenViking-Account: +X-OpenViking-User: +X-API-Key: +``` + +部署必须满足以下安全条件： + +- OpenViking 仅暴露在可信内网； +- 外部客户端不能绕过 DeerFlow 直接伪造 account/user header； +- DeerFlow 不接受客户端传入的 OpenViking account/user 值； +- user 值只来自 DeerFlow 已认证的 runtime user context； +- API key 只从服务端环境变量或 secrets provider 读取。 + +如后续选择 OpenViking `api_key` 模式，需要增加用户注册、用户 key 保存、轮换和吊销机制，不作为首版默认方案。 + +### 6.2 Agent 范围 + +DeerFlow 当前记忆作用域是： + +```text +(user_id, agent_name) +``` + +其中 `agent_name=None` 表示默认/共享桶。OpenViking 的用户记忆默认属于整个用户，因此必须额外表达 Agent 范围。 + +首版采用以下逻辑命名： + +```text +default Agent: __default__ +custom Agent: canonicalize(agent_name) +``` + +`canonicalize` 必须与 DeerFlow Agent 命名规则一致： + +- 转为小写； +- 只允许安全字符； +- 拒绝路径分隔符、`.`、`..` 和控制字符； +- 显式 Agent 不允许占用 `__default__`。 + +首版采用独立 OpenViking trusted-user 映射作为硬隔离边界： + +```text +openviking_user = "df_" + sha256(account + user_id + agent_scope)[:40] +``` + +这样不依赖 OpenViking 的实验性 Agent tag/URI 过滤语义，不同 DeerFlow +Agent 不可能通过一次普通用户级检索读到彼此的记忆。代价是同一 DeerFlow +用户的 profile/preferences 不会自动跨 Agent 共享；这是首版为强隔离选择的 +明确权衡。后续只有在 OpenViking 提供稳定、服务端强制的 Agent 过滤后，才 +考虑改为同一 OpenViking user 下的子范围，并且需要数据迁移方案。 + +### 6.3 Thread 与 Session + +Session ID 必须稳定、不可碰撞且不暴露原始标识。建议： + +```text +session_id = "df_" + base32( + sha256( + integration_namespace + + "\0" + account + + "\0" + user_id + + "\0" + canonical_agent_scope + + "\0" + thread_id + ) +) +``` + +要求： + +- 同一用户、Agent、线程始终映射到同一 Session； +- 任意一个作用域字段不同，Session ID 必须不同； +- 不直接把邮箱、用户名或 thread ID 放进 OpenViking Session ID； +- 算法一旦发布不得随意修改； +- `integration_namespace` 使用固定版本值，例如 `deerflow-openviking-v1`。 + +## 7. 写入流程 + +### 7.1 正常写入 + +现有 `MemoryMiddleware` 在 Agent 一轮完成后调用： + +```python +manager.add( + thread_id, + messages, + agent_name=..., + user_id=..., + trace_id=..., +) +``` + +OpenViking 后端处理流程： + +```text +接收 DeerFlow messages + -> 验证 user_id/thread_id + -> 计算 Agent scope 与 Session ID + -> 过滤框架内部消息 + -> 转换 user/assistant 文本消息 + -> 根据同步水位去除已提交到 Session 的消息 + -> 批量写入 OpenViking Session + -> 原子记录 submitted 水位 + -> commit Session + -> 推进 committed 水位并记录 commit task_id + -> 返回，不等待后台提取完成 +``` + +首版应使用 OpenViking 的批量消息接口；若锁定版本不支持批量接口，Client 可内部降级为逐条提交，但 Manager 接口保持批量语义。 + +### 7.2 消息过滤 + +首版只同步： + +- 用户的真实输入； +- Agent 最终可见的 assistant 回复。 + +首版不上传： + +- system prompt； +- middleware 内部提醒； +- `hide_from_ui` 且不含真实用户澄清内容的消息； +- tool call 的原始参数与完整输出； +- 二进制或 base64 payload； +- 仅用于流式拼接的中间 assistant chunk； +- subagent 内部推理和不可见消息。 + +消息过滤应由适配器自己的纯函数完成并单独测试。后端可以消费 DeerFlow 工厂传入的 `should_keep_hidden_message` hook，但不能直接导入 DeerFlow 的消息过滤内部模块。 + +### 7.3 幂等与同步水位 + +`add`、总结前的 `add_nowait`、请求重试和 Gateway 关闭 flush 可能看到重叠消息。仅依赖 OpenViking 去重不足以保证不会重复归档。 + +适配器需要保存每个 Session 的同步水位： + +```json +{ + "schema_version": 2, + "session_id": "df_...", + "submitted_message_ids": ["..."], + "committed_message_ids": ["..."], + "last_commit_task_id": "...", + "last_archive_uri": "viking://..." +} +``` + +水位文件使用工厂提供的 `backend_config.storage_path`，建议位置： + +```text +/openviking/sessions/.json +``` + +要求： + +- 原子写入； +- 不保存消息正文； +- 记录一个有界的近期 message ID 窗口，而不是无限增长； +- 批量添加成功后立即推进 `submitted_message_ids`，避免 commit 结果未知时 + 在下一轮重复添加消息； +- commit 结果未知时不自动重试该 commit；后续相同历史直接跳过，出现新消息时 + 只添加新消息，并由新的 commit 一并归档仍留在 Session 中的旧消息； +- commit 成功后将 `committed_message_ids` 推进到 submitted 水位并保存 + task/archive 元数据； +- 网络超时后需要区分“请求未到达”和“服务端可能已处理”； +- 若 OpenViking API 支持客户端 message ID，应始终发送 DeerFlow 的稳定 message ID。 + +这里的 submitted 水位只解决“批量添加已明确成功、随后 commit 失败”的确定性 +重复路径。如果批量添加在服务端成功、但客户端在收到响应或保存水位前中断， +没有服务端幂等键仍无法做到 exactly-once；多 Gateway 部署同样需要共享水位或 +OpenViking 原生幂等支持。 + +如果 DeerFlow 消息缺少稳定 ID，应基于角色、规范化内容和在本轮中的稳定位置生成适配器 ID，但该方案只作为兼容路径。 + +### 7.4 `add` 与 `add_nowait` + +OpenViking 自己在 commit 后异步提取，不需要复制 DeerMem 的 debounce queue。 + +- `add`：完成消息上传和 commit 接受确认后返回。 +- `add_nowait`：语义为“不要因 DeerFlow 总结丢失消息”，仍应立即完成上传和 commit 接受确认；它不是 fire-and-forget。 + +两者均不默认等待 OpenViking 后台 task 完成。 + +### 7.5 后台提取 + +OpenViking commit 通常先返回 `task_id`，摘要与长期记忆提取随后在后台执行。因此系统是最终一致的： + +```text +commit accepted != memory immediately searchable +``` + +首版策略： + +- 正常轮次不轮询 task； +- 保存最近一次 task ID 用于日志与诊断； +- task 失败不回滚 DeerFlow 主回复； +- `shutdown_flush` 只保证待上传消息已获得 commit 接受确认，不保证所有 OpenViking 提取任务完成； +- 集成测试可轮询 task 以验证完整闭环。 + +## 8. 检索与注入流程 + +### 8.1 `get_context` + +`get_context` 没有显式 query 参数，而且当前 prompt 调用点也不传 +`thread_id`。为保持主框架和共享接口不变，首版使用可配置的受控通用查询： + +```text +user profile preferences important entities events ongoing goals constraints and prior decisions +``` + +该值由 `retrieval.injection_query` 配置。请求固定限制到当前 hashed trusted +user 的 `viking://user/memories`，并设置 `context_type=memory`。显式 +`MemoryManager.search(query)` 仍使用调用者提供的语义查询。后续如果需要 +基于当前用户输入进行更精准的自动召回，应单独为共享接口设计可选 query +参数并更新所有 backend，不能通过线程局部变量隐式传递。 + +### 8.2 `search` + +映射到 OpenViking 检索接口： + +```text +query -> query +top_k -> node_limit +user_id -> OpenViking user identity +agent_name -> Agent scope filter +context_type -> memory +score_threshold -> backend_config.retrieval.score_threshold +``` + +返回值转换为 DeerFlow `memory_search` 可消费的字典： + +```json +{ + "id": "stable-result-id-or-uri", + "content": "retrieved memory text", + "category": "openviking memory type", + "confidence": 0.82, + "source": "viking://...", + "score": 0.82 +} +``` + +字段规则： + +- `id` 优先使用 OpenViking 稳定 URI； +- `content` 使用可读正文或 L1/L2 摘要，不返回内部原始 envelope； +- `category` 使用 OpenViking memory type； +- `confidence` 若 OpenViking 未提供事实置信度，可使用归一化检索 score，但必须在代码中明确这是兼容映射； +- `source` 只返回非敏感 URI； +- 不向模型返回 account、API key、内部文件系统路径或服务端堆栈。 + +首版 `category` 过滤应映射到 OpenViking memory type。若无法服务端过滤，可以在适配器中先过滤再截取 `top_k`，不能先截断再过滤。 + +### 8.3 注入文本 + +`get_context` 返回纯文本，由 DeerFlow 现有调用点包裹进 ``。适配器自身不得再添加 `` 标签。 + +建议格式： + +```markdown +## Relevant long-term memory + +- [preferences] User prefers concise technical explanations. +- [entities] Project Alpha uses Python 3.12. +- [experiences] Previous deployment succeeded after enabling trusted auth. +``` + +要求： + +- 按相关度排序； +- 去除重复 URI 和重复内容； +- 限制单条长度； +- 限制总字符数或估算 token 数； +- 结果为空时返回空字符串； +- OpenViking 读取失败且配置为 fail-open 时返回空字符串； +- 不把 HTTP 错误文本注入 prompt。 + +## 9. `MemoryManager` 方法支持矩阵 + +| 方法 | 首版 | 实现策略 | +|---|---:|---| +| `from_config` | 支持 | 解析配置并创建 Client | +| `add` | 支持 | 批量添加消息并 commit | +| `add_nowait` | 支持 | 立即添加并 commit | +| `get_context` | 支持 | 检索并格式化注入文本 | +| `search` | 支持 | OpenViking memory search | +| `shutdown_flush` | 支持 | 有界等待本地待提交操作 | +| `warm` | 支持 | `/health`，不改变数据 | +| `get_memory` | 暂不支持 | 继承 `NotImplementedError` | +| `clear_memory` | 暂不支持 | 继承 `NotImplementedError` | +| `import_memory` | 暂不支持 | 继承 `NotImplementedError` | +| `reload_memory` | 暂不支持 | 继承默认行为 | +| `create_fact` | 暂不支持 | OpenViking 不是 DeerMem fact 模型 | +| `update_fact` | 暂不支持 | 同上 | +| `delete_fact` | 暂不支持 | 同上 | + +由于 `supports_search=True`，技术上可通过 `MemoryManager` 的 tool-mode invariant，但首版配置校验应明确拒绝 `mode: tool`。原因是 DeerFlow tool 模式不仅需要 `search`，还暴露 `memory_add/update/delete`，而首版不实现 fact CRUD。 + +## 10. HTTP Client 设计 + +### 10.1 Client 接口 + +建议内部接口： + +```python +class OpenVikingHttpClient: + def health(self) -> bool: ... + + def ensure_session( + self, + *, + identity: OpenVikingIdentity, + session_id: str, + ) -> None: ... + + def add_messages( + self, + *, + identity: OpenVikingIdentity, + session_id: str, + messages: list[OpenVikingMessage], + ) -> None: ... + + def commit_session( + self, + *, + identity: OpenVikingIdentity, + session_id: str, + ) -> OpenVikingCommitResult: ... + + def search( + self, + *, + identity: OpenVikingIdentity, + query: str, + session_id: str | None, + agent_scope: str, + top_k: int, + category: str | None, + ) -> list[OpenVikingSearchHit]: ... + + def get_task( + self, + *, + identity: OpenVikingIdentity, + task_id: str, + ) -> OpenVikingTaskStatus: ... +``` + +### 10.2 HTTP 库 + +优先复用 DeerFlow 已声明的 HTTP 客户端库。若使用 `httpx`： + +- 同步 `MemoryManager` 路径使用有连接池的 `httpx.Client`； +- Client 生命周期跟随 `OpenVikingMemoryManager` 单例； +- 禁止每次请求创建新 Client； +- 连接池上限配置化； +- 关闭时显式释放 Client。 + +不能直接使用 OpenViking 嵌入式 Python Client。若官方 HTTP SDK 只带来轻量依赖且能够稳定锁版本，可以在后续替换，但适配器内部 Client 接口不变。 + +### 10.3 Timeout + +必须分别设置： + +- connect timeout； +- read timeout； +- write timeout； +- pool timeout； +- shutdown 总预算。 + +任何请求不得无限等待。推荐初始值： + +```yaml +connect_timeout_seconds: 2 +read_timeout_seconds: 10 +write_timeout_seconds: 10 +pool_timeout_seconds: 2 +shutdown_flush_timeout_seconds: 20 +``` + +### 10.4 重试 + +只自动重试满足幂等条件的操作： + +| 操作 | 自动重试 | +|---|---| +| health | 可以 | +| search/find | 可以 | +| get task | 可以 | +| ensure/get session | 可以 | +| add messages | 仅在稳定 message ID 被服务端幂等处理时 | +| commit | 仅在服务端提供幂等键或可确认当前 archive 状态时 | + +重试采用有上限的指数退避和 jitter。HTTP 400/401/403/404 等确定性错误不重试；429、502、503、504 和连接建立失败可按策略重试。 + +### 10.5 错误类型 + +Client 将底层异常转换为适配器私有错误： + +```text +OpenVikingConnectionError +OpenVikingTimeoutError +OpenVikingAuthenticationError +OpenVikingProtocolError +OpenVikingRateLimitError +OpenVikingUnavailableError +OpenVikingScopeError +``` + +错误对象可包含： + +- 操作名； +- HTTP status； +- 可公开的错误码； +- request ID； +- 是否可重试。 + +错误对象不得包含： + +- API key； +-完整认证 header； +- 用户消息正文； +- OpenViking 内部堆栈； +- 未脱敏的响应 body。 + +## 11. 失败与降级策略 + +### 11.1 启动 + +配置语法错误、认证模式矛盾或 URL 非法时启动失败，不能静默切换回 DeerMem。错误的持久化后端若被静默替换，会把记忆写入错误存储。 + +OpenViking 健康检查失败有两种可配置策略： + +- `startup_policy: fail_fast`：生产推荐，Gateway 启动失败； +- `startup_policy: warn`：本地开发可用，Gateway 启动但记忆暂时降级。 + +### 11.2 读取 + +默认： + +```yaml +failure_policy: + read: fail_open +``` + +search/get_context 超时或服务不可用时： + +- 记录结构化 warning/error； +- 返回空结果或空字符串； +- 不阻塞主 Agent 回复； +- 认证失败和 scope 错误使用更高等级日志，不视为普通临时故障。 + +### 11.3 写入 + +首版不实现无限持久化重试队列。默认： + +```yaml +failure_policy: + write: log_and_drop +``` + +含义： + +- 写入失败不使已经生成的主回复失败； +- 记录 thread/session、操作类型和错误码； +- 不记录消息正文； +- 暴露指标以便告警； +- 明确承认本轮记忆可能未被 OpenViking 捕获。 + +后续如要求“至少一次”交付，应增加独立、持久化的 outbox，而不是简单扩大内存重试： + +```text +DeerFlow commit outbox + -> durable enqueue + -> background delivery + -> OpenViking idempotency + -> acknowledge/delete +``` + +该能力不属于首版。 + +## 12. 配置草案 + +`config.example.yaml` 增加注释示例，默认仍为 `deermem`： + +```yaml +memory: + enabled: true + mode: middleware + injection_enabled: true + manager_class: deermem + + # To use OpenViking over HTTP: + # manager_class: openviking + # backend_config: + # base_url: http://openviking:1933 + # auth_mode: trusted + # account: deerflow + # api_key_env: OPENVIKING_API_KEY + # + # connect_timeout_seconds: 2 + # read_timeout_seconds: 10 + # write_timeout_seconds: 10 + # pool_timeout_seconds: 2 + # + # startup_policy: fail_fast + # failure_policy: + # read: fail_open + # write: log_and_drop + # + # retrieval: + # top_k: 8 + # score_threshold: 0.25 + # max_injection_chars: 12000 + # + # session: + # wait_for_extraction: false +``` + +配置要求： + +- `api_key_env` 保存环境变量名，不保存密钥值； +- `auth_mode=trusted` 时必须配置 `account`； +- 非 localhost 的 OpenViking URL 默认要求 HTTPS，或显式允许内部明文 HTTP； +- `top_k`、timeout 和注入长度设置合理上下限； +- `mode != middleware` 时首版拒绝启动； +- `/memory/config` 返回配置时只显示 `api_key_env`，不解析或回显密钥。 + +## 13. 并发与生命周期 + +`MemoryManager` 是进程级单例，可能从 Agent 请求线程、总结路径和 Gateway 关闭路径并发访问。 + +实现必须保证： + +- HTTP Client 可跨线程安全使用，或使用明确的线程本地 Client； +- Session 水位更新有进程内锁； +- 多 Gateway 进程共享同一 `storage_path` 时使用跨进程锁或不依赖本地水位作为唯一真相； +- 同一 Session 的 add + commit 串行化； +- 不同 Session 可以并行； +- `shutdown_flush(timeout)` 遵守硬超时； +- Gateway 关闭后拒绝接受新的写入； +- Client close 与在途请求之间不存在竞态。 + +如果部署允许多个 Gateway 副本同时处理同一 thread，本地水位文件不足以提供全局一致性。此时必须依赖 OpenViking 稳定 message ID/幂等键，或把水位迁移到 DeerFlow 的共享数据库。首版发布前必须明确支持的部署拓扑。 + +## 14. 可观测性 + +至少记录以下结构化事件： + +```text +openviking.health +openviking.session.ensure +openviking.messages.add +openviking.session.commit +openviking.search +openviking.task.status +openviking.degraded +``` + +推荐字段： + +- operation； +- duration_ms； +- success； +- status_code； +- error_type； +- retry_count； +- result_count； +- account 的非敏感别名； +- user/session 的不可逆 hash； +- DeerFlow trace ID； +- OpenViking request ID/task ID。 + +禁止记录： + +- API key； +-完整用户消息； +-完整记忆正文； +-认证 header； +-未经处理的 HTTP response body。 + +推荐指标： + +```text +openviking_requests_total{operation,status} +openviking_request_duration_seconds{operation} +openviking_commit_total{status} +openviking_search_results{operation} +openviking_degraded_total{reason} +openviking_pending_writes +``` + +`MemoryCallbacks.on_memory_llm_call` 不适用于远程 OpenViking 内部的 LLM 调用，因为 DeerFlow 看不到该 LLM 边界。DeerFlow 应记录 HTTP 调用 span，并由 OpenViking 自己提供服务端模型调用可观测性。 + +## 15. 测试方案 + +### 15.1 单元测试 + +新增建议： + +```text +backend/tests/test_openviking_memory_config.py +backend/tests/test_openviking_memory_client.py +backend/tests/test_openviking_memory_manager.py +backend/tests/test_openviking_memory_scope.py +``` + +配置测试： + +- 合法配置； +- 未知字段； +- URL 和 timeout 边界； +- 环境变量密钥解析； +- 配置序列化不泄漏密钥； +- tool mode 被首版拒绝。 + +Client 测试： + +- 正确 endpoint、method、header 和 body； +- trusted identity header； +- response envelope 解析； +- timeout； +- 401/403/429/5xx 映射； +- 只对允许操作重试； +- 日志脱敏。 + +Manager 测试： + +- backend 注册发现； +- `from_config`； +- DeerFlow 消息过滤与转换； +- Session ID 稳定性和隔离性； +- add -> batch messages -> commit； +- 重复 message ID 不重复发送； +- search 结果转换； +- context 排序、去重和长度上限； +- read fail-open； +- write failure 不抛到主链路； +- shutdown 硬超时。 + +范围测试： + +- 不同 user 的请求 header 和 Session 不同； +- 同一 user 不同 Agent 的 scope 不同； +- `agent_name=None` 稳定映射为默认范围； +- 非法 Agent 名称不能进入 URI/header； +- category 在 top-k 之前过滤； +- 检索请求始终限制为 `context_type=memory`。 + +### 15.2 契约测试 + +CI 默认使用 mock HTTP server，不要求安装或启动真实 OpenViking。契约测试应固定本项目所支持 OpenViking 版本的： + +- endpoint； +-请求字段； +-响应 envelope； +-错误格式； +-认证 header； +-commit/task 状态枚举； +-search hit 字段。 + +OpenViking 升级时先更新契约 fixture，再修改 Client。 + +### 15.3 真实集成测试 + +提供可选测试标记，例如： + +```text +pytest -m openviking_integration +``` + +真实测试至少验证： + +```text +创建测试身份 + -> 写入一轮明确偏好 + -> commit + -> 轮询 task 完成 + -> 新 Session 搜索该偏好 + -> 结果只在当前 user/Agent 范围可见 +``` + +测试数据必须使用独立 account/user 前缀，并在测试后清理。 + +### 15.4 阻塞 IO 测试 + +本接入会引入网络 IO，必须纳入仓库 blocking-IO 检查： + +- 确认所有同步 HTTP 调用都运行在允许的 worker/thread 路径； +- 不允许在 ASGI event loop 上直接执行同步 `httpx.Client` 请求； +- 为关键调用路径增加 `tests/blocking_io/` 回归锚点； +- 若现有调用点不能保证 offload，应优先改为调用 `aadd`、`aget_context`、`asearch` 的 async 实现，而不是在 backend 内创建隐藏 event loop。 + +## 16. 文档与部署 + +实现完成时需要同步更新： + +- 根 `README.md`：列出 OpenViking 可选记忆后端； +- `backend/AGENTS.md`：记录架构、配置与测试边界； +- `config.example.yaml`：完整注释示例； +- 前端 application configuration 文档； +- harness memory 文档； +- memory backends README； +- OpenViking 服务部署与认证说明。 + +首版不要求修改 Docker Compose。部署文档先要求操作者提供： + +- 可访问的 OpenViking Server URL； +- 已配置的 embedding/VLM； +- trusted auth 或用户 key； +-持久化存储； +-健康检查； +-支持版本。 + +后续可增加可选 Compose profile，而不把 OpenViking 变成 DeerFlow 的强制服务。 + +## 17. 实施阶段 + +### 阶段 0：版本与范围验证 + +- 锁定一个 OpenViking 支持版本； +- 验证 Session batch messages、commit、task、search API； +- 固定 hashed trusted-user Agent scope 映射； +- 固定 `retrieval.injection_query` 自动召回策略； +- 确认 OpenViking HTTP 错误 envelope； +- 产出固定的契约 fixture。 + +退出条件：身份隔离、Agent 隔离和 query 数据流没有未决设计。 + +### 阶段 1：HTTP Client + +- 配置模型； +-内部数据模型； +- HTTP Client； +-认证、timeout、重试、错误映射； +- Client 单元测试。 + +退出条件：Client 可以在 mock server 上完成 health、消息提交、commit、task 和 search。 + +### 阶段 2：MemoryManager 适配 + +- backend 注册； +-作用域映射； +-消息转换； +-同步水位； +- `add`、`add_nowait`； +- `get_context`、`search`； +- `warm`、`shutdown_flush`； +- Manager 和隔离测试。 + +退出条件：mock 环境跑通 DeerFlow 写入与检索注入闭环。 + +### 阶段 3：真实联调 + +-连接真实 OpenViking； +-完成 commit 后轮询提取； +-跨 Session 召回； +-用户和 Agent 隔离； +-故障、超时和重启测试； +- blocking-IO 验证。 + +退出条件：验收用例全部通过，且 OpenViking 故障不会拖垮主对话链路。 + +### 阶段 4：文档与发布 + +-补充配置与部署文档； +-锁定兼容版本； +-记录已知限制； +-保留 DeerMem 默认值； +-增加发布说明和回滚方法。 + +## 18. 验收标准 + +功能： + +- `manager_class: openviking` 可被后端扫描器发现并实例化； +- middleware 模式下每轮有效消息只提交一次； +- commit 获得接受确认并记录 task ID； +-后续对话能召回已完成提取的长期记忆； +-注入文本有明确长度上限； +-搜索结果可被 DeerFlow `memory_search` 兼容消费； +- OpenViking 内部完成抽取，DeerFlow 不执行第二套抽取。 + +隔离： + +-不同 DeerFlow user 无法互相读写 Session 或记忆； +-同一 user 的不同 Agent 专属记忆按既定策略隔离； +-默认 Agent 与显式 Agent 不冲突； +-非法 scope 输入在发送 HTTP 请求前被拒绝。 + +可靠性： + +-所有 HTTP 请求有硬 timeout； +-读取故障按配置 fail-open； +-写入故障不会导致已生成的主回复失败； +-不存在 API key 日志泄漏； +- shutdown 遵守总预算； +-同步网络 IO 不阻塞 ASGI event loop。 + +兼容： + +- DeerMem 继续为默认 backend； +-未配置 OpenViking 的部署不引入额外运行时依赖或启动开销； +- OpenViking backend 目录除 `MemoryManager` 契约外不导入 DeerFlow 内部模块； +-现有 DeerMem/noop 测试全部通过。 + +## 19. 回滚 + +回滚只修改配置并重启 DeerFlow： + +```yaml +memory: + manager_class: deermem +``` + +注意： + +-回滚不会自动把 OpenViking 记忆迁回 DeerMem； +- OpenViking 中已有数据保持不变； +-切回 DeerMem 后只会使用 DeerMem 自己的历史数据； +-禁止在 OpenViking 加载失败时运行时静默回退到 DeerMem，因为这会把新写入路由到不同存储并制造分叉。 + +## 20. 待确认事项 + +进入实现前必须关闭以下问题： + +1. 首个支持的 OpenViking 版本及升级策略是什么？ +2. DeerFlow 的目标部署是否允许多个 Gateway 副本并发处理同一 thread？ +3. 写入失败的首版策略是否长期接受 `log_and_drop`，还是后续必须实现 durable outbox？ +4. 是否需要在后续版本对 `/memory` 管理页面隐藏不支持的 CRUD 操作？ + +这些问题涉及数据隔离、幂等性或用户可见行为，不应由实现代码中的隐式默认值代替。 + +## 21. OpenViking 参考 + +- API Overview: +- Sessions API: +- Retrieval API: +- Session Management: +- Multi-Tenant: +- Authentication: +- OpenViking repository: diff --git a/frontend/src/content/en/application/configuration.mdx b/frontend/src/content/en/application/configuration.mdx index 671f10c9f..a79c7e8d2 100644 --- a/frontend/src/content/en/application/configuration.mdx +++ b/frontend/src/content/en/application/configuration.mdx @@ -223,7 +223,7 @@ For Docker or scripted starts, set `UV_EXTRAS=postgres` before installing or bui memory: enabled: true injection_enabled: true - manager_class: deermem # backend selector: deermem | noop | dotted path + manager_class: deermem # backend selector: deermem | noop | openviking | dotted path mode: middleware # middleware (default) | tool (experimental) backend_config: # DeerMem-private knobs (the backend self-interprets these) storage_path: "" # empty = deer-flow base_dir (per-user memory under it) @@ -237,6 +237,13 @@ memory: # api_key: $OPENAI_API_KEY ``` +The optional `openviking` backend connects to an independent OpenViking HTTP +server and currently supports `mode: middleware`. Its private configuration +uses `base_url`, `auth_mode`, `account`, `api_key_env`, `failure_policy`, and +`retrieval` instead of the DeerMem fields shown above. See +`docs/OPENVIKING.md` in the repository for the complete configuration and +Docker startup sequence. + ## Frontend environment variables Set these before running `pnpm build` or starting the frontend in production: diff --git a/frontend/src/content/en/harness/memory.mdx b/frontend/src/content/en/harness/memory.mdx index ffbdb3638..19d70a46d 100644 --- a/frontend/src/content/en/harness/memory.mdx +++ b/frontend/src/content/en/harness/memory.mdx @@ -45,7 +45,7 @@ memory: enabled: true injection_enabled: true - # Backend selector: deermem (default) | noop | a dotted path to a custom + # Backend selector: deermem (default) | noop | openviking | a dotted path to a custom # MemoryManager subclass. Swap backend = drop a backends// folder and # set this (see backend/.../agents/memory/backends/). manager_class: deermem @@ -89,6 +89,36 @@ memory: max_injection_tokens: 2000 ``` +## OpenViking HTTP backend + +Set `manager_class: openviking` to send completed turns to an independent +OpenViking server and recall its memory over HTTP. This backend currently +supports middleware mode only; DeerMem remains the default. + +```yaml +memory: + enabled: true + injection_enabled: true + manager_class: openviking + mode: middleware + backend_config: + base_url: http://openviking:1933 + auth_mode: trusted + account: deerflow + api_key_env: OPENVIKING_API_KEY + failure_policy: + read: fail_open + write: log_and_drop + retrieval: + top_k: 8 + score_threshold: 0.25 + max_injection_chars: 12000 +``` + +The backend hashes each DeerFlow user/agent scope into a distinct OpenViking +trusted identity. Keep trusted-mode OpenViking on an internal network and put +the API key in the server environment, not directly in `config.yaml`. + ## Global vs per-agent memory DeerFlow supports two levels of memory: