mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-07 21:38:53 +00:00
refactor(memory): use official OpenViking adapter (#4707)
* refactor(memory): use official OpenViking adapter * fix(memory): preserve OpenViking recall behavior * fix(memory): ignore ambient OpenViking headers
This commit is contained in:
Hao Zhe
GitHub
parent
99c926b7bb
commit
6556d09d7f
@ -25,7 +25,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
|
||||
# OPENVIKING_API_KEY=your-openviking-user-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
|
||||
|
||||
18
README.md
18
README.md
@ -1028,16 +1028,14 @@ request the binary capability retain the legacy JSON/base64 frame protocol.
|
||||
|
||||
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`. Bounded submitted-message watermarks cover long and
|
||||
compacted histories and prevent a failed Session commit from duplicating
|
||||
already accepted messages on retry; the shared HTTP client also has explicit
|
||||
connection limits and jittered retries. See
|
||||
[OpenViking memory backend](docs/OPENVIKING.md) for configuration and Docker
|
||||
startup.
|
||||
DeerFlow also includes an optional `openviking` memory backend. It uses the
|
||||
official `langchain-openviking` package to capture completed turns into stable
|
||||
OpenViking Sessions and recall memory for prompt injection while leaving
|
||||
DeerMem as the default. The initial integration supports one DeerFlow user with
|
||||
one credential-bound OpenViking USER API key in `memory.mode: middleware` and
|
||||
does not inherit arbitrary HTTP headers from `ovcli.conf`.
|
||||
See [OpenViking memory backend](docs/OPENVIKING.md) for its configuration,
|
||||
behavior, and current boundaries.
|
||||
|
||||
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.
|
||||
|
||||
|
||||
@ -271,8 +271,8 @@ Blocking-IO runtime gate (`tests/blocking_io/`):
|
||||
skips redundant sandbox sync when thread data is already mounted);
|
||||
`test_channel_outbound_files.py` (locks Feishu, Telegram, and WeCom outbound
|
||||
attachment open/read/hash work off the event loop);
|
||||
`test_openviking_memory_backend.py` (locks the OpenViking backend's async
|
||||
add/context/search entrypoints offloading synchronous HTTP and watermark
|
||||
`test_openviking_memory_backend.py` (locks the official OpenViking backend's
|
||||
async add/context/search entrypoints offloading synchronous SDK and cursor
|
||||
filesystem IO); and
|
||||
`test_workspace_changes_recorder.py` (locks the offload around the snapshot
|
||||
text cache lifecycle — roots resolution, `mkdtemp`, and the `shutil.rmtree`
|
||||
@ -1014,26 +1014,26 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
|
||||
- `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
|
||||
remote-only adapter built on the maintained `langchain-openviking` package.
|
||||
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 combines a constant-size
|
||||
ordered-prefix digest for append-only histories with a bounded recent-ID
|
||||
fallback for compaction and separately records submitted and committed
|
||||
progress. 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. Schema-v2 recent-ID
|
||||
watermarks migrate without duplicating their anchored history. The shared
|
||||
HTTP client has explicit total/keep-alive connection limits and jittered
|
||||
exponential retry delays, and its configuration representation omits the API
|
||||
key. 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.
|
||||
uses one OpenViking USER API key bound to the configured DeerFlow
|
||||
`owner_user_id`; another DeerFlow user is rejected before remote access.
|
||||
DeerFlow owns the existing recall/capture timing, fixed injection query and
|
||||
full-transcript suffix cursor. `langchain-openviking` owns SDK transport,
|
||||
message conversion, tool-call preservation, batching, partial-write progress
|
||||
and Session commits. One DeerFlow thread maps to one stable OpenViking
|
||||
Session, with the default or named agent represented as its actor peer.
|
||||
Bounded hash-only cursors live below `{storage_path}/openviking/sessions/`;
|
||||
session locks are weakly cached, async entrypoints offload synchronous SDK
|
||||
and file IO, and graceful shutdown drains active operations before closing
|
||||
the recorder-owned client. The recorder receives an explicit empty
|
||||
`extra_headers` mapping so `ovcli.conf` cannot add arbitrary transport
|
||||
headers. Do not reintroduce a backend-local HTTP client,
|
||||
explicitly configured trusted identity headers, root-key data access, or
|
||||
imports of the OpenViking embedded runtime. Multi-user provisioning,
|
||||
query-aware refresh policy and new lifecycle scheduling are separate changes,
|
||||
not part of this backend.
|
||||
- `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, manual CRUD primitives, and the updater backend. Injection is mode-aware: middleware mode injects global `user`/`history` summaries plus the selected agent's facts, while tool mode injects only the global summaries and leaves every agent fact behind `memory_search` to avoid duplicating automatically injected and retrieval-returned context. `memory.injection_enabled: false` suppresses the complete block in either mode.
|
||||
- Middleware extraction classifies proposed facts with extraction-only `scope`/`durability`/`authority` labels. `_apply_updates` accepts only `user` + `durable` + `descriptive` new/consolidated facts, accepts only wholly user-scoped summary prose with `authority=descriptive`, and rejects missing labels per item without aborting unrelated updates. Contradiction removals use object entries with `id`, `scope`, `reason`, and optional zero-based `replacementFactIndex`; task/project removals fail closed, and a paired removal runs only when the referenced replacement survives the scope/confidence gates, deduplication, and max-fact trim under another fact ID. The labels are not persisted, so no storage migration is required. Staleness removals retain their independent candidate/cap guardrails, while tool-mode CRUD remains outside this extraction gate. Custom `memory.backend_config.prompts_dir` templates (including per-agent overrides) must carry the same classification fields; an un-migrated template makes the fail-closed gate reject every extraction-driven write, observable only through `rejected_by_scope_gate` and the >60% fact-rejection warning.
|
||||
|
||||
@ -4,7 +4,8 @@ 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).
|
||||
- `openviking/` - optional remote backend using the official
|
||||
`langchain-openviking` package (single-user 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.
|
||||
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
"""OpenViking HTTP memory backend."""
|
||||
"""OpenViking memory backend using the official LangChain integration."""
|
||||
|
||||
from .openviking_manager import OpenVikingMemoryManager
|
||||
|
||||
|
||||
@ -1,247 +0,0 @@
|
||||
"""Thin synchronous HTTP client for the OpenViking server API."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
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,
|
||||
)
|
||||
limits = httpx.Limits(
|
||||
max_connections=config.max_connections,
|
||||
max_keepalive_connections=config.max_keepalive_connections,
|
||||
)
|
||||
self._client = httpx.Client(
|
||||
base_url=config.base_url,
|
||||
timeout=timeout,
|
||||
limits=limits,
|
||||
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(_retry_delay(attempt))
|
||||
continue
|
||||
raise OpenVikingTimeoutError(operation, "OpenViking request timed out") from exc
|
||||
except httpx.TransportError as exc:
|
||||
if attempt + 1 < attempts:
|
||||
time.sleep(_retry_delay(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(_retry_delay(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")
|
||||
|
||||
|
||||
def _retry_delay(attempt: int) -> float:
|
||||
base_delay = 0.05 * (2**attempt)
|
||||
return base_delay + random.uniform(0.0, base_delay)
|
||||
@ -1,69 +1,84 @@
|
||||
"""Configuration for the OpenViking HTTP memory backend."""
|
||||
"""Validated configuration for the official OpenViking memory adapter."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from math import isfinite
|
||||
from typing import Any, Literal
|
||||
from urllib.parse import urlparse
|
||||
|
||||
_SAFE_PEER_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
|
||||
GENERATED_PEER_PREFIX = "df-agent-"
|
||||
_REMOVED_CUSTOM_HTTP_FIELDS = frozenset(
|
||||
{
|
||||
"connect_timeout_seconds",
|
||||
"max_connections",
|
||||
"max_keepalive_connections",
|
||||
"max_retries",
|
||||
"pool_timeout_seconds",
|
||||
"read_timeout_seconds",
|
||||
"write_timeout_seconds",
|
||||
}
|
||||
)
|
||||
|
||||
@dataclass(frozen=True)
|
||||
|
||||
@dataclass(frozen=True, slots=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.
|
||||
"""
|
||||
"""Credential-bound connection settings and existing DeerFlow policy."""
|
||||
|
||||
base_url: str
|
||||
storage_path: str
|
||||
auth_mode: Literal["trusted", "dev"]
|
||||
account: str
|
||||
api_key: str | None = field(repr=False)
|
||||
connect_timeout_seconds: float
|
||||
read_timeout_seconds: float
|
||||
write_timeout_seconds: float
|
||||
pool_timeout_seconds: float
|
||||
max_connections: int
|
||||
max_keepalive_connections: int
|
||||
max_retries: int
|
||||
owner_user_id: str
|
||||
api_key: str = field(repr=False)
|
||||
api_key_env: str
|
||||
default_peer_id: str
|
||||
timeout_seconds: float
|
||||
search_top_k: int
|
||||
score_threshold: float | None
|
||||
max_injection_chars: int
|
||||
content_mode: Literal["auto", "abstract", "overview", "read"]
|
||||
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:
|
||||
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")
|
||||
if "auth_mode" in cfg or "account" in cfg:
|
||||
raise ValueError("OpenViking trusted mode is no longer supported by this backend; use a USER API key and configure owner_user_id")
|
||||
removed_fields = sorted(_REMOVED_CUSTOM_HTTP_FIELDS.intersection(cfg))
|
||||
if removed_fields:
|
||||
raise ValueError("OpenViking custom HTTP client fields are no longer supported: " + ", ".join(removed_fields))
|
||||
|
||||
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
|
||||
if not api_key_env:
|
||||
raise ValueError("OpenViking api_key_env must not be empty")
|
||||
|
||||
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_connections=int(cfg.pop("max_connections", 100)),
|
||||
max_keepalive_connections=int(cfg.pop("max_keepalive_connections", 20)),
|
||||
max_retries=int(cfg.pop("max_retries", 1)),
|
||||
owner_user_id=str(cfg.pop("owner_user_id", "")).strip(),
|
||||
api_key=os.environ.get(api_key_env, "").strip(),
|
||||
api_key_env=api_key_env,
|
||||
default_peer_id=str(cfg.pop("default_peer_id", "deerflow")).strip(),
|
||||
timeout_seconds=float(cfg.pop("timeout_seconds", 30.0)),
|
||||
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)),
|
||||
content_mode=str(retrieval.pop("content_mode", "overview")).lower(), # type: ignore[arg-type]
|
||||
injection_query=str(
|
||||
retrieval.pop(
|
||||
"injection_query",
|
||||
@ -73,14 +88,22 @@ class OpenVikingConfig:
|
||||
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)),
|
||||
allow_insecure_http=_boolean(
|
||||
cfg.pop("allow_insecure_http", False),
|
||||
"allow_insecure_http",
|
||||
),
|
||||
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)])
|
||||
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)}")
|
||||
raise ValueError("Unknown OpenViking backend_config fields: " + ", ".join(unknown))
|
||||
result._validate()
|
||||
return result
|
||||
|
||||
@ -89,28 +112,25 @@ class OpenVikingConfig:
|
||||
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 1 <= self.max_connections <= 1000:
|
||||
raise ValueError("OpenViking max_connections must be between 1 and 1000")
|
||||
if not 0 <= self.max_keepalive_connections <= self.max_connections:
|
||||
raise ValueError("OpenViking max_keepalive_connections must be between 0 and max_connections")
|
||||
if not 0 <= self.max_retries <= 5:
|
||||
raise ValueError("OpenViking max_retries must be between 0 and 5")
|
||||
raise ValueError("OpenViking plain HTTP is allowed only for localhost/openviking; set allow_insecure_http=true only for a trusted internal network")
|
||||
if not self.owner_user_id:
|
||||
raise ValueError("OpenViking owner_user_id must not be empty")
|
||||
if not self.api_key:
|
||||
raise ValueError(f"OpenViking USER API key is missing; set {self.api_key_env}")
|
||||
if not is_safe_peer_id(self.default_peer_id):
|
||||
raise ValueError("OpenViking default_peer_id must start with a lowercase letter or digit and contain at most 64 lowercase letters, digits, '_' or '-'")
|
||||
if self.default_peer_id.startswith(GENERATED_PEER_PREFIX):
|
||||
raise ValueError(f"OpenViking default_peer_id must not start with the reserved prefix {GENERATED_PEER_PREFIX!r}")
|
||||
if not isfinite(self.timeout_seconds) or self.timeout_seconds <= 0:
|
||||
raise ValueError("OpenViking timeout_seconds must be a finite value > 0")
|
||||
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 self.score_threshold is not None and (not isfinite(self.score_threshold) or not 0 <= self.score_threshold <= 1):
|
||||
raise ValueError("OpenViking retrieval.score_threshold must be a finite value 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 self.content_mode not in {"auto", "abstract", "overview", "read"}:
|
||||
raise ValueError("OpenViking retrieval.content_mode must be auto, abstract, overview, or read")
|
||||
if not self.injection_query:
|
||||
raise ValueError("OpenViking retrieval.injection_query must not be empty")
|
||||
if self.startup_policy not in {"fail_fast", "warn"}:
|
||||
@ -123,6 +143,12 @@ class OpenVikingConfig:
|
||||
raise ValueError("OpenViking max_seen_message_ids must be between 16 and 10000")
|
||||
|
||||
|
||||
def is_safe_peer_id(value: str) -> bool:
|
||||
"""Return whether *value* is valid for an OpenViking actor peer."""
|
||||
|
||||
return _SAFE_PEER_RE.fullmatch(value) is not None
|
||||
|
||||
|
||||
def _mapping(value: Any, name: str) -> dict[str, Any]:
|
||||
if value is None:
|
||||
return {}
|
||||
@ -133,3 +159,15 @@ def _mapping(value: Any, name: str) -> dict[str, Any]:
|
||||
|
||||
def _optional_float(value: Any) -> float | None:
|
||||
return None if value is None else float(value)
|
||||
|
||||
|
||||
def _boolean(value: Any, name: str) -> bool:
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
normalized = value.strip().lower()
|
||||
if normalized in {"true", "1", "yes", "on"}:
|
||||
return True
|
||||
if normalized in {"false", "0", "no", "off"}:
|
||||
return False
|
||||
raise ValueError(f"OpenViking {name} must be a boolean")
|
||||
|
||||
@ -1,63 +0,0 @@
|
||||
"""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
|
||||
@ -1,59 +1,90 @@
|
||||
"""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.
|
||||
"""
|
||||
"""OpenViking memory backend built on the maintained LangChain adapters."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
import copy
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
import weakref
|
||||
from collections.abc import Callable
|
||||
from contextlib import AbstractContextManager
|
||||
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 deerflow.agents.memory.manager import MemoryManager, MemoryManagerError
|
||||
|
||||
from .client import OpenVikingClientError, OpenVikingHttpClient
|
||||
from .config import OpenVikingConfig
|
||||
from .models import OpenVikingIdentity, OpenVikingMessage, OpenVikingSearchHit
|
||||
from .session import (
|
||||
_advanced_cursor,
|
||||
_canonical_peer_id,
|
||||
_captureable_messages,
|
||||
_matching_prefix_count,
|
||||
_memory_target_uris,
|
||||
_message_signature,
|
||||
_session_id,
|
||||
_string_list,
|
||||
)
|
||||
|
||||
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."""
|
||||
"""Single-user OpenViking backend using the official integration package.
|
||||
|
||||
DeerFlow continues to choose when capture and recall occur. The official
|
||||
adapter owns SDK transport, message conversion, batching, commit retries,
|
||||
and retrieval behavior.
|
||||
"""
|
||||
|
||||
supports_search: ClassVar[bool] = True
|
||||
|
||||
_config: OpenVikingConfig = PrivateAttr()
|
||||
_client: OpenVikingHttpClient = PrivateAttr()
|
||||
_should_keep_hidden_message: Callable[[Any], bool] | None = PrivateAttr(default=None)
|
||||
_client: Any = PrivateAttr()
|
||||
_recorder: Any = PrivateAttr()
|
||||
_retriever: Any = PrivateAttr()
|
||||
_use_actor_peer: Any = PrivateAttr()
|
||||
_partial_write_error: type[Exception] = PrivateAttr()
|
||||
_should_keep_hidden_message: Any = 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)
|
||||
_close_requested: bool = PrivateAttr(default=False)
|
||||
_resources_closed: bool = PrivateAttr(default=False)
|
||||
_resource_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)
|
||||
|
||||
def model_post_init(self, __context: Any) -> None:
|
||||
self._config = OpenVikingConfig.from_backend_config(self.backend_config)
|
||||
self._client = OpenVikingHttpClient(self._config)
|
||||
integration = _load_official_integration()
|
||||
commit_policy = integration["OpenVikingCommitPolicy"](mode="always")
|
||||
self._recorder = integration["OpenVikingSessionRecorder"](
|
||||
url=self._config.base_url,
|
||||
api_key=self._config.api_key,
|
||||
timeout=self._config.timeout_seconds,
|
||||
# An explicit mapping disables the SDK's ovcli.conf header fallback.
|
||||
extra_headers={},
|
||||
commit_policy=commit_policy,
|
||||
)
|
||||
# The recorder owns one recovery-aware SDK client. Retrieval borrows the
|
||||
# same handle, so this backend has one connection pool and one owner.
|
||||
self._client = self._recorder.client
|
||||
self._retriever = integration["OpenVikingRetriever"](
|
||||
client=self._client,
|
||||
search_mode="find",
|
||||
limit=self._config.search_top_k,
|
||||
score_threshold=self._config.score_threshold,
|
||||
context_types=("memory",),
|
||||
content_mode=self._config.content_mode,
|
||||
max_content_chars=self._config.max_injection_chars,
|
||||
)
|
||||
self._use_actor_peer = integration["use_actor_peer"]
|
||||
self._partial_write_error = integration["OpenVikingPartialWriteError"]
|
||||
|
||||
@classmethod
|
||||
def from_config(
|
||||
@ -64,10 +95,10 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
**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
|
||||
raise ValueError("The OpenViking automatic-memory backend supports memory.mode='middleware' only; use OpenViking MCP for explicit model tools")
|
||||
instance = cls(backend_config=backend_config or {}, mode=mode)
|
||||
hidden_filter = host_hooks.get("should_keep_hidden_message")
|
||||
instance._should_keep_hidden_message = hidden_filter if callable(hidden_filter) else None
|
||||
return instance
|
||||
|
||||
def add(
|
||||
@ -79,7 +110,13 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
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)
|
||||
del trace_id
|
||||
self._write_conversation(
|
||||
thread_id,
|
||||
messages,
|
||||
agent_name=agent_name,
|
||||
user_id=user_id,
|
||||
)
|
||||
|
||||
def add_nowait(
|
||||
self,
|
||||
@ -89,7 +126,14 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
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)
|
||||
# Preserve DeerFlow's existing pre-compaction behavior. This backend
|
||||
# commits every accepted capture, so add_nowait needs no separate mode.
|
||||
self._write_conversation(
|
||||
thread_id,
|
||||
messages,
|
||||
agent_name=agent_name,
|
||||
user_id=user_id,
|
||||
)
|
||||
|
||||
async def aadd(
|
||||
self,
|
||||
@ -119,21 +163,31 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
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,
|
||||
peer_id = self._resolve_scope(user_id, agent_name)
|
||||
retriever = copy.copy(self._retriever)
|
||||
retriever.target_uri = _memory_target_uris(peer_id)
|
||||
if thread_id:
|
||||
retriever.search_mode = "search"
|
||||
retriever.session_id = _session_id(
|
||||
self._config.owner_user_id,
|
||||
peer_id,
|
||||
thread_id,
|
||||
)
|
||||
except OpenVikingClientError:
|
||||
try:
|
||||
with self._actor_peer_scope(peer_id):
|
||||
documents = retriever.invoke(self._config.injection_query)
|
||||
except Exception:
|
||||
if self._config.read_failure_policy == "raise":
|
||||
raise
|
||||
logger.warning("OpenViking context retrieval failed; continuing without injected memory", exc_info=True)
|
||||
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)
|
||||
return _format_documents(
|
||||
documents,
|
||||
max_chars=self._config.max_injection_chars,
|
||||
)
|
||||
finally:
|
||||
self._end_operation()
|
||||
|
||||
@ -160,26 +214,33 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
agent_name: str | None = None,
|
||||
category: str | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
if not query.strip():
|
||||
return []
|
||||
if not self._begin_operation():
|
||||
if not query.strip() or not self._begin_operation():
|
||||
return []
|
||||
try:
|
||||
peer_id = self._resolve_scope(user_id, agent_name)
|
||||
retriever = copy.copy(self._retriever)
|
||||
retriever.target_uri = _memory_target_uris(peer_id)
|
||||
retriever.search_mode = "find"
|
||||
retriever.session_id = None
|
||||
retriever.limit = max(1, min(int(top_k), 100))
|
||||
if category:
|
||||
retriever.filter = {
|
||||
"op": "must",
|
||||
"field": "category",
|
||||
"conds": [category],
|
||||
}
|
||||
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:
|
||||
with self._actor_peer_scope(peer_id):
|
||||
documents = retriever.invoke(query.strip())
|
||||
except Exception:
|
||||
if self._config.read_failure_policy == "raise":
|
||||
raise
|
||||
logger.warning("OpenViking memory search failed; returning no results", exc_info=True)
|
||||
logger.warning(
|
||||
"OpenViking memory search failed; returning no results",
|
||||
exc_info=True,
|
||||
)
|
||||
return []
|
||||
return [_hit_to_fact(hit) for hit in hits]
|
||||
return [_document_to_fact(document) for document in documents]
|
||||
finally:
|
||||
self._end_operation()
|
||||
|
||||
@ -206,14 +267,18 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
return False
|
||||
try:
|
||||
try:
|
||||
healthy = self._client.health()
|
||||
except OpenVikingClientError:
|
||||
health = getattr(self._client, "health", None)
|
||||
healthy = bool(health()) if callable(health) else True
|
||||
except Exception:
|
||||
if self._config.startup_policy == "fail_fast":
|
||||
raise
|
||||
logger.warning("OpenViking health check failed; memory will run in degraded mode", exc_info=True)
|
||||
logger.warning(
|
||||
"OpenViking startup validation 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")
|
||||
raise MemoryManagerError("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
|
||||
@ -221,21 +286,30 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
self._end_operation()
|
||||
|
||||
def shutdown_flush(self, timeout: float) -> bool:
|
||||
"""Stop new operations, drain in-flight work, then close the HTTP pool."""
|
||||
"""Stop new work, drain accepted calls, and close owned resources."""
|
||||
|
||||
deadline = time.monotonic() + max(0.0, timeout)
|
||||
with self._lifecycle:
|
||||
self._closed = True
|
||||
self._close_requested = 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()
|
||||
self._close_resources()
|
||||
return True
|
||||
|
||||
def close(self) -> None:
|
||||
"""Request idempotent closure after any active operation finishes."""
|
||||
|
||||
with self._lifecycle:
|
||||
self._closed = True
|
||||
self._close_requested = True
|
||||
can_close = self._active_operations == 0
|
||||
if can_close:
|
||||
self._close_resources()
|
||||
|
||||
def _write_conversation(
|
||||
self,
|
||||
thread_id: str,
|
||||
@ -250,116 +324,147 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
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", [])))
|
||||
converted = _convert_messages(messages, self._should_keep_hidden_message)
|
||||
prefix_count = _matching_submitted_prefix_count(state, submitted_ids, converted)
|
||||
if prefix_count is not None:
|
||||
pending = converted[prefix_count:]
|
||||
else:
|
||||
submitted = set(submitted_ids)
|
||||
pending = [message for message in converted if message.message_id not in submitted]
|
||||
if not pending:
|
||||
if prefix_count is None and converted:
|
||||
state = {
|
||||
**state,
|
||||
"schema_version": 3,
|
||||
"submitted_prefix_count": len(converted),
|
||||
"submitted_prefix_digest": _message_sequence_digest(converted),
|
||||
}
|
||||
self._save_state(session_id, state)
|
||||
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": 3,
|
||||
"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 :],
|
||||
"submitted_prefix_count": len(converted),
|
||||
"submitted_prefix_digest": _message_sequence_digest(converted),
|
||||
"committed_prefix_count": state.get("committed_prefix_count"),
|
||||
"committed_prefix_digest": state.get("committed_prefix_digest"),
|
||||
"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": 3,
|
||||
"committed_message_ids": state["submitted_message_ids"],
|
||||
"committed_prefix_count": state["submitted_prefix_count"],
|
||||
"committed_prefix_digest": state["submitted_prefix_digest"],
|
||||
"last_commit_task_id": commit.task_id,
|
||||
"last_archive_uri": commit.archive_uri,
|
||||
}
|
||||
self._save_state(session_id, state)
|
||||
peer_id = self._resolve_scope(user_id, agent_name)
|
||||
session_id = _session_id(
|
||||
self._config.owner_user_id,
|
||||
peer_id,
|
||||
thread_id,
|
||||
)
|
||||
with self._session_lock(session_id):
|
||||
self._capture_locked(
|
||||
session_id,
|
||||
peer_id,
|
||||
_captureable_messages(
|
||||
messages,
|
||||
self._should_keep_hidden_message,
|
||||
),
|
||||
)
|
||||
finally:
|
||||
self._end_operation()
|
||||
|
||||
def _search_hits(
|
||||
def _capture_locked(
|
||||
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,
|
||||
session_id: str,
|
||||
peer_id: str,
|
||||
messages: list[Any],
|
||||
) -> None:
|
||||
state = self._load_cursor(session_id)
|
||||
signatures = [_message_signature(message) for message in messages]
|
||||
|
||||
if state.get("commit_pending"):
|
||||
try:
|
||||
with self._actor_peer_scope(peer_id):
|
||||
self._recorder.flush(session_id)
|
||||
except Exception as exc:
|
||||
self._handle_write_error(
|
||||
exc,
|
||||
"OpenViking pending commit retry failed; preserving capture cursor",
|
||||
session_id,
|
||||
)
|
||||
return
|
||||
state = {**state, "commit_pending": False}
|
||||
self._save_cursor(session_id, state)
|
||||
|
||||
start = _matching_prefix_count(state, signatures)
|
||||
append_only = start is not None
|
||||
if append_only:
|
||||
pending = messages[start:]
|
||||
pending_signatures = signatures[start:]
|
||||
else:
|
||||
submitted = set(_string_list(state.get("submitted_signatures")))
|
||||
pending_pairs = [(message, signature) for message, signature in zip(messages, signatures, strict=True) if signature not in submitted]
|
||||
pending = [message for message, _ in pending_pairs]
|
||||
pending_signatures = [signature for _, signature in pending_pairs]
|
||||
|
||||
if not pending:
|
||||
self._save_cursor(
|
||||
session_id,
|
||||
_advanced_cursor(
|
||||
state,
|
||||
signatures,
|
||||
[],
|
||||
max_seen=self._config.max_seen_message_ids,
|
||||
commit_pending=False,
|
||||
),
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
with self._actor_peer_scope(peer_id):
|
||||
self._recorder.record(
|
||||
session_id,
|
||||
pending,
|
||||
peer_id=peer_id,
|
||||
)
|
||||
except self._partial_write_error as exc:
|
||||
consumed = max(
|
||||
0,
|
||||
min(
|
||||
len(pending_signatures),
|
||||
int(getattr(exc, "input_messages_consumed", 0)),
|
||||
),
|
||||
)
|
||||
confirmed = pending_signatures[:consumed]
|
||||
commit_pending = bool(getattr(exc, "commit_pending", False))
|
||||
if confirmed or commit_pending:
|
||||
confirmed_prefix = signatures[: int(start or 0) + consumed] if append_only else None
|
||||
self._save_cursor(
|
||||
session_id,
|
||||
_advanced_cursor(
|
||||
state,
|
||||
confirmed_prefix,
|
||||
confirmed,
|
||||
max_seen=self._config.max_seen_message_ids,
|
||||
commit_pending=commit_pending,
|
||||
),
|
||||
)
|
||||
self._handle_write_error(
|
||||
exc,
|
||||
"OpenViking partially recorded a conversation; confirmed progress was preserved",
|
||||
session_id,
|
||||
)
|
||||
return
|
||||
except Exception as exc:
|
||||
self._handle_write_error(
|
||||
exc,
|
||||
"OpenViking conversation recording failed; capture cursor was not advanced",
|
||||
session_id,
|
||||
)
|
||||
return
|
||||
|
||||
self._save_cursor(
|
||||
session_id,
|
||||
_advanced_cursor(
|
||||
state,
|
||||
signatures,
|
||||
pending_signatures,
|
||||
max_seen=self._config.max_seen_message_ids,
|
||||
commit_pending=False,
|
||||
),
|
||||
)
|
||||
|
||||
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 _resolve_scope(
|
||||
self,
|
||||
user_id: str | None,
|
||||
agent_name: str | None,
|
||||
) -> str:
|
||||
resolved_user = str(user_id or "default")
|
||||
if resolved_user != self._config.owner_user_id:
|
||||
raise MemoryManagerError(f"OpenViking USER API key is bound to DeerFlow owner_user_id {self._config.owner_user_id!r}, but this request belongs to {resolved_user!r}. Refusing to share one credential across users.")
|
||||
return _canonical_peer_id(agent_name, self._config.default_peer_id)
|
||||
|
||||
def _actor_peer_scope(
|
||||
self,
|
||||
peer_id: str,
|
||||
) -> AbstractContextManager[None]:
|
||||
return self._use_actor_peer(peer_id)
|
||||
|
||||
def _session_lock(self, session_id: str) -> threading.RLock:
|
||||
with self._session_locks_guard:
|
||||
return self._session_locks.setdefault(session_id, threading.RLock())
|
||||
return self._session_locks.setdefault(
|
||||
session_id,
|
||||
threading.RLock(),
|
||||
)
|
||||
|
||||
def _begin_operation(self) -> bool:
|
||||
with self._lifecycle:
|
||||
@ -369,172 +474,108 @@ class OpenVikingMemoryManager(MemoryManager):
|
||||
return True
|
||||
|
||||
def _end_operation(self) -> None:
|
||||
should_close = False
|
||||
with self._lifecycle:
|
||||
self._active_operations -= 1
|
||||
if self._active_operations == 0:
|
||||
self._lifecycle.notify_all()
|
||||
should_close = self._close_requested
|
||||
if should_close:
|
||||
try:
|
||||
self._close_resources()
|
||||
except Exception:
|
||||
logger.exception("Failed to close OpenViking memory resources")
|
||||
|
||||
def _close_resources(self) -> None:
|
||||
with self._resource_lock:
|
||||
if self._resources_closed:
|
||||
return
|
||||
self._recorder.close()
|
||||
self._resources_closed = True
|
||||
|
||||
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]:
|
||||
def _load_cursor(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 {}
|
||||
except (OSError, ValueError) as exc:
|
||||
raise MemoryManagerError(f"OpenViking capture cursor is unreadable; refusing unsafe replay (session={session_id})") from exc
|
||||
if not isinstance(value, dict):
|
||||
raise MemoryManagerError(f"OpenViking capture cursor is invalid; refusing unsafe replay (session={session_id})")
|
||||
return value
|
||||
|
||||
def _save_state(self, session_id: str, state: dict[str, Any]) -> None:
|
||||
def _save_cursor(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")
|
||||
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)
|
||||
logger.debug(
|
||||
"Failed to remove OpenViking cursor temp file: %s",
|
||||
temp_path,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
def _handle_write_error(
|
||||
self,
|
||||
exc: Exception,
|
||||
message: str,
|
||||
session_id: str,
|
||||
) -> None:
|
||||
detail = f"{message} (session={session_id})"
|
||||
if self._config.write_failure_policy == "raise":
|
||||
raise MemoryManagerError(detail) from exc
|
||||
logger.error(detail, 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 _matching_submitted_prefix_count(
|
||||
state: dict[str, Any],
|
||||
submitted_ids: list[str],
|
||||
messages: list[OpenVikingMessage],
|
||||
) -> int | None:
|
||||
count = state.get("submitted_prefix_count")
|
||||
digest = state.get("submitted_prefix_digest")
|
||||
if isinstance(count, int) and 0 <= count <= len(messages) and isinstance(digest, str):
|
||||
if _message_sequence_digest(messages[:count]) == digest:
|
||||
return count
|
||||
return None
|
||||
|
||||
# Schema v2 only retained a recent suffix of submitted IDs. When that
|
||||
# suffix still appears intact, it safely anchors the append-only prefix and
|
||||
# avoids a one-time duplicate submission during migration to schema v3.
|
||||
if submitted_ids and len(submitted_ids) <= len(messages):
|
||||
message_ids = [message.message_id for message in messages]
|
||||
width = len(submitted_ids)
|
||||
for start in range(len(message_ids) - width, -1, -1):
|
||||
if message_ids[start : start + width] == submitted_ids:
|
||||
return start + width
|
||||
return None
|
||||
|
||||
|
||||
def _message_sequence_digest(messages: list[OpenVikingMessage]) -> str:
|
||||
digest = hashlib.sha256()
|
||||
for message in messages:
|
||||
encoded = message.message_id.encode()
|
||||
digest.update(len(encoded).to_bytes(8, "big"))
|
||||
digest.update(encoded)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
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]:
|
||||
def _load_official_integration() -> dict[str, Any]:
|
||||
try:
|
||||
from langchain_openviking import (
|
||||
OpenVikingCommitPolicy,
|
||||
OpenVikingPartialWriteError,
|
||||
OpenVikingRetriever,
|
||||
OpenVikingSessionRecorder,
|
||||
has_request_actor_peer_support,
|
||||
)
|
||||
from langchain_openviking.actor_peer import use_actor_peer
|
||||
except ImportError as exc:
|
||||
raise ImportError("The OpenViking memory backend requires langchain-openviking==0.1.0. Install DeerFlow backend dependencies and retry.") from exc
|
||||
if not has_request_actor_peer_support():
|
||||
raise ImportError("The installed OpenViking SDK lacks request-scoped actor-peer support. Install openviking-sdk>=0.1.6,<0.2 and retry.")
|
||||
return {
|
||||
"id": hit.uri,
|
||||
"content": _hit_content(hit),
|
||||
"category": hit.category or "memory",
|
||||
"confidence": hit.score,
|
||||
"source": hit.uri,
|
||||
"score": hit.score,
|
||||
"OpenVikingCommitPolicy": OpenVikingCommitPolicy,
|
||||
"OpenVikingPartialWriteError": OpenVikingPartialWriteError,
|
||||
"OpenVikingRetriever": OpenVikingRetriever,
|
||||
"OpenVikingSessionRecorder": OpenVikingSessionRecorder,
|
||||
"use_actor_peer": use_actor_peer,
|
||||
}
|
||||
|
||||
|
||||
def _format_context(hits: list[OpenVikingSearchHit], *, max_chars: int) -> str:
|
||||
def _format_documents(documents: list[Any], *, max_chars: int) -> str:
|
||||
lines: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for hit in hits:
|
||||
content = " ".join(_hit_content(hit).split())
|
||||
for document in documents:
|
||||
content = " ".join(str(getattr(document, "page_content", "") or "").split())
|
||||
key = content.casefold()
|
||||
if not content or key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
line = f"- [{hit.category or 'memory'}] {content}"
|
||||
metadata = getattr(document, "metadata", {}) or {}
|
||||
category = metadata.get("openviking_category") or "memory"
|
||||
line = f"- [{category}] {content}"
|
||||
candidate = "\n".join([*lines, line])
|
||||
if len(candidate) > max_chars:
|
||||
remaining = max_chars - len("\n".join(lines)) - (1 if lines else 0)
|
||||
@ -543,3 +584,24 @@ def _format_context(hits: list[OpenVikingSearchHit], *, max_chars: int) -> str:
|
||||
break
|
||||
lines.append(line)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _document_to_fact(document: Any) -> dict[str, Any]:
|
||||
metadata = getattr(document, "metadata", {}) or {}
|
||||
uri = metadata.get("openviking_uri") or metadata.get("source") or ""
|
||||
score = metadata.get("openviking_score")
|
||||
return {
|
||||
"id": uri,
|
||||
"content": str(getattr(document, "page_content", "") or ""),
|
||||
"category": metadata.get("openviking_category") or "memory",
|
||||
"confidence": score,
|
||||
"source": uri,
|
||||
"score": score,
|
||||
}
|
||||
|
||||
|
||||
__all__ = [
|
||||
"OpenVikingMemoryManager",
|
||||
"_canonical_peer_id",
|
||||
"_session_id",
|
||||
]
|
||||
|
||||
@ -0,0 +1,177 @@
|
||||
"""Stable OpenViking session identity and transcript-cursor helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
from .config import GENERATED_PEER_PREFIX, is_safe_peer_id
|
||||
|
||||
_SESSION_NAMESPACE = "deerflow-openviking-adapter-v1"
|
||||
_DEFAULT_AGENT_SCOPE = "__default__"
|
||||
_CURSOR_SCHEMA_VERSION = 1
|
||||
|
||||
|
||||
def _canonical_peer_id(
|
||||
agent_name: str | None,
|
||||
default_peer_id: str,
|
||||
) -> str:
|
||||
"""Map DeerFlow's case-insensitive agent names to disjoint peer IDs."""
|
||||
|
||||
if agent_name is None:
|
||||
return default_peer_id
|
||||
|
||||
value = str(agent_name).strip().lower()
|
||||
if not value or value == _DEFAULT_AGENT_SCOPE:
|
||||
raise ValueError(f"Invalid OpenViking peer scope: {agent_name!r}")
|
||||
if is_safe_peer_id(value) and value != default_peer_id and not value.startswith(GENERATED_PEER_PREFIX):
|
||||
return value
|
||||
|
||||
# The generated namespace is reserved, so compatible names, the default
|
||||
# peer, and hashed fallbacks cannot alias one another. The 128-bit digest
|
||||
# also avoids collisions caused by sanitizing or truncating agent names.
|
||||
digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:32]
|
||||
return f"{GENERATED_PEER_PREFIX}{digest}"
|
||||
|
||||
|
||||
def _session_id(
|
||||
owner_user_id: str,
|
||||
peer_id: str,
|
||||
thread_id: str,
|
||||
) -> str:
|
||||
"""Derive one stable OpenViking session for one DeerFlow thread."""
|
||||
|
||||
digest = hashlib.sha256(f"{_SESSION_NAMESPACE}\0{owner_user_id}\0{peer_id}\0{thread_id}".encode()).hexdigest()
|
||||
return f"df_{digest[:48]}"
|
||||
|
||||
|
||||
def _memory_target_uris(peer_id: str) -> list[str]:
|
||||
"""Return the self and current-peer memory roots for a request."""
|
||||
|
||||
return [
|
||||
"viking://user/memories",
|
||||
f"viking://user/peers/{peer_id}/memories",
|
||||
]
|
||||
|
||||
|
||||
def _captureable_messages(
|
||||
messages: list[Any],
|
||||
should_keep_hidden_message: Any,
|
||||
) -> list[Any]:
|
||||
"""Drop DeerFlow-only injected context before handing messages to OpenViking."""
|
||||
|
||||
selected: list[Any] = []
|
||||
for message in messages:
|
||||
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
|
||||
selected.append(message)
|
||||
return selected
|
||||
|
||||
|
||||
def _message_signature(message: Any) -> str:
|
||||
"""Hash stable message semantics without retaining transcript content."""
|
||||
|
||||
additional_kwargs = _message_value(message, "additional_kwargs", {})
|
||||
if not isinstance(additional_kwargs, Mapping):
|
||||
additional_kwargs = {}
|
||||
tool_calls = _message_value(message, "tool_calls", None)
|
||||
if not tool_calls:
|
||||
tool_calls = additional_kwargs.get("tool_calls") or []
|
||||
|
||||
value = {
|
||||
"id": _message_value(message, "id", None),
|
||||
"role": _message_value(message, "type", None) or _message_value(message, "role", None),
|
||||
"content": _message_value(message, "content", ""),
|
||||
"tool_calls": tool_calls,
|
||||
"tool_call_id": _message_value(message, "tool_call_id", None) or _message_value(message, "tool_id", None),
|
||||
"tool_name": _message_value(message, "name", None) or _message_value(message, "tool_name", None),
|
||||
"tool_status": _message_value(message, "status", None) or _message_value(message, "tool_status", None),
|
||||
}
|
||||
encoded = json.dumps(
|
||||
value,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
default=str,
|
||||
separators=(",", ":"),
|
||||
).encode()
|
||||
return hashlib.sha256(encoded).hexdigest()
|
||||
|
||||
|
||||
def _matching_prefix_count(
|
||||
state: dict[str, Any],
|
||||
signatures: list[str],
|
||||
) -> int | None:
|
||||
"""Return the already submitted prefix, including after compaction."""
|
||||
|
||||
count = state.get("submitted_prefix_count")
|
||||
digest = state.get("submitted_prefix_digest")
|
||||
if isinstance(count, int) and 0 <= count <= len(signatures) and isinstance(digest, str):
|
||||
if _sequence_digest(signatures[:count]) == digest:
|
||||
return count
|
||||
return None
|
||||
|
||||
submitted = _string_list(state.get("submitted_signatures"))
|
||||
if submitted and len(submitted) <= len(signatures):
|
||||
width = len(submitted)
|
||||
for start in range(len(signatures) - width, -1, -1):
|
||||
if signatures[start : start + width] == submitted:
|
||||
return start + width
|
||||
return 0 if not state else None
|
||||
|
||||
|
||||
def _advanced_cursor(
|
||||
previous: dict[str, Any],
|
||||
prefix_signatures: list[str] | None,
|
||||
newly_submitted: list[str],
|
||||
*,
|
||||
max_seen: int,
|
||||
commit_pending: bool,
|
||||
) -> dict[str, Any]:
|
||||
"""Advance confirmed capture progress without persisting message content."""
|
||||
|
||||
recent = [
|
||||
*_string_list(previous.get("submitted_signatures")),
|
||||
*newly_submitted,
|
||||
][-max_seen:]
|
||||
state: dict[str, Any] = {
|
||||
"schema_version": _CURSOR_SCHEMA_VERSION,
|
||||
"submitted_signatures": recent,
|
||||
"commit_pending": commit_pending,
|
||||
}
|
||||
if prefix_signatures is not None:
|
||||
state["submitted_prefix_count"] = len(prefix_signatures)
|
||||
state["submitted_prefix_digest"] = _sequence_digest(prefix_signatures)
|
||||
else:
|
||||
state["submitted_prefix_count"] = previous.get("submitted_prefix_count")
|
||||
state["submitted_prefix_digest"] = previous.get("submitted_prefix_digest")
|
||||
return state
|
||||
|
||||
|
||||
def _sequence_digest(signatures: list[str]) -> str:
|
||||
digest = hashlib.sha256()
|
||||
for signature in signatures:
|
||||
encoded = signature.encode()
|
||||
digest.update(len(encoded).to_bytes(8, "big"))
|
||||
digest.update(encoded)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def _string_list(value: Any) -> list[str]:
|
||||
if not isinstance(value, list):
|
||||
return []
|
||||
return [item for item in value if isinstance(item, str)]
|
||||
|
||||
|
||||
def _message_value(message: Any, key: str, default: Any) -> Any:
|
||||
if isinstance(message, Mapping):
|
||||
return message.get(key, default)
|
||||
return getattr(message, key, default)
|
||||
@ -22,6 +22,7 @@ dependencies = [
|
||||
"langchain-anthropic>=1.4.1",
|
||||
"langchain-deepseek>=1.0.1",
|
||||
"langchain-mcp-adapters>=0.2.2",
|
||||
"langchain-openviking==0.1.0",
|
||||
"langchain-openai>=1.2.1",
|
||||
"langfuse>=3.4.1",
|
||||
"langgraph>=1.2.9,<1.3",
|
||||
|
||||
@ -1,85 +1,141 @@
|
||||
"""Regression anchors: OpenViking async memory methods must not block the loop."""
|
||||
"""Regression anchors: OpenViking async methods must not block the loop."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import nullcontext
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from langchain_core.documents import Document
|
||||
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
|
||||
from deerflow.agents.memory.backends.openviking.openviking_manager import (
|
||||
OpenVikingMemoryManager,
|
||||
)
|
||||
|
||||
|
||||
class _BlockingProbeClient:
|
||||
"""Perform real file IO so Blockbuster can detect missing async offload."""
|
||||
class _CommitPolicy:
|
||||
def __init__(self, *, mode: str):
|
||||
self.mode = mode
|
||||
|
||||
def __init__(self, probe_path: Path):
|
||||
self._probe_path = probe_path
|
||||
|
||||
def _probe(self) -> None:
|
||||
self._probe_path.write_text("probe", encoding="utf-8")
|
||||
class _Client:
|
||||
supports_request_actor_peer = True
|
||||
|
||||
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 health(self) -> bool:
|
||||
return True
|
||||
|
||||
def close(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def _manager(tmp_path: Path) -> OpenVikingMemoryManager:
|
||||
class _BlockingRecorder:
|
||||
def __init__(self, *, commit_policy: Any, **kwargs: Any):
|
||||
del kwargs
|
||||
self.commit_policy = commit_policy
|
||||
self.client = _Client()
|
||||
self.probe_path: Path | None = None
|
||||
|
||||
def record(
|
||||
self,
|
||||
session_id: str,
|
||||
messages: list[Any],
|
||||
peer_id: str | None = None,
|
||||
) -> None:
|
||||
del session_id, messages, peer_id
|
||||
assert self.probe_path is not None
|
||||
self.probe_path.write_text("record", encoding="utf-8")
|
||||
|
||||
def flush(self, session_id: str) -> None:
|
||||
del session_id
|
||||
|
||||
def close(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
class _BlockingRetriever:
|
||||
def __init__(self, *, client: Any, **kwargs: Any):
|
||||
del client
|
||||
self.__dict__.update(kwargs)
|
||||
self.filter = None
|
||||
self.target_uri = ""
|
||||
self.session_id = None
|
||||
self.probe_path: Path | None = None
|
||||
|
||||
def __copy__(self) -> _BlockingRetriever:
|
||||
copied = type(self)(client=None)
|
||||
copied.__dict__.update(self.__dict__)
|
||||
return copied
|
||||
|
||||
def invoke(self, query: str) -> list[Document]:
|
||||
del query
|
||||
assert self.probe_path is not None
|
||||
self.probe_path.write_text("retrieve", encoding="utf-8")
|
||||
return [
|
||||
Document(
|
||||
page_content="Prefers concise answers.",
|
||||
metadata={
|
||||
"openviking_uri": ("viking://user/memories/preferences/test.md"),
|
||||
"openviking_category": "preferences",
|
||||
"openviking_score": 0.9,
|
||||
},
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _manager(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> OpenVikingMemoryManager:
|
||||
import deerflow.agents.memory.backends.openviking.openviking_manager as module
|
||||
|
||||
monkeypatch.setenv("OPENVIKING_API_KEY", "user-key")
|
||||
monkeypatch.setattr(
|
||||
module,
|
||||
"_load_official_integration",
|
||||
lambda: {
|
||||
"OpenVikingCommitPolicy": _CommitPolicy,
|
||||
"OpenVikingPartialWriteError": RuntimeError,
|
||||
"OpenVikingRetriever": _BlockingRetriever,
|
||||
"OpenVikingSessionRecorder": _BlockingRecorder,
|
||||
"use_actor_peer": lambda peer_id: nullcontext(),
|
||||
},
|
||||
)
|
||||
manager = OpenVikingMemoryManager.from_config(
|
||||
{
|
||||
"base_url": "http://openviking:1933",
|
||||
"storage_path": str(tmp_path),
|
||||
"auth_mode": "trusted",
|
||||
"account": "deerflow",
|
||||
"owner_user_id": "alice",
|
||||
"startup_policy": "warn",
|
||||
}
|
||||
)
|
||||
manager._client = _BlockingProbeClient(tmp_path / "probe.txt") # type: ignore[assignment]
|
||||
manager._recorder.probe_path = tmp_path / "record.txt"
|
||||
manager._retriever.probe_path = tmp_path / "retrieve.txt"
|
||||
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")]
|
||||
async def test_async_openviking_operations_do_not_block_event_loop(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
manager = _manager(tmp_path, monkeypatch)
|
||||
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") == [
|
||||
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.",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
28
backend/uv.lock
generated
28
backend/uv.lock
generated
@ -936,6 +936,7 @@ dependencies = [
|
||||
{ name = "langchain-google-genai" },
|
||||
{ name = "langchain-mcp-adapters" },
|
||||
{ name = "langchain-openai" },
|
||||
{ name = "langchain-openviking" },
|
||||
{ name = "langfuse" },
|
||||
{ name = "langgraph" },
|
||||
{ name = "langgraph-api" },
|
||||
@ -1015,6 +1016,7 @@ requires-dist = [
|
||||
{ name = "langchain-mcp-adapters", specifier = ">=0.2.2" },
|
||||
{ name = "langchain-ollama", marker = "extra == 'ollama'", specifier = ">=0.3.0" },
|
||||
{ name = "langchain-openai", specifier = ">=1.2.1" },
|
||||
{ name = "langchain-openviking", specifier = "==0.1.0" },
|
||||
{ name = "langfuse", specifier = ">=3.4.1" },
|
||||
{ name = "langgraph", specifier = ">=1.2.9,<1.3" },
|
||||
{ name = "langgraph-api", specifier = ">=0.8.1" },
|
||||
@ -2097,6 +2099,20 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/dc/55/2865b18ee3a3dd11160b8c4b2cf37e75bf2a4a8d1d38868ffffc7b7cc180/langchain_openai-1.2.1-py3-none-any.whl", hash = "sha256:a80732185030d4f453dda6c25feef46f645f665423fdffe38ae3edf1ac3c6c4d", size = 98626, upload-time = "2026-04-24T19:46:41.971Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "langchain-openviking"
|
||||
version = "0.1.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "langchain-core" },
|
||||
{ name = "openviking-sdk" },
|
||||
{ name = "pydantic" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/7f/c0/418dba42bddb5093d20d5e36c3e8d9ceff0b27ce92a6bc2f16fb33535fac/langchain_openviking-0.1.0.tar.gz", hash = "sha256:dc4ee47a122527881eae1fbbe6f9ca857cc0269cd60f08c8d3fd3fff94996e9e", size = 48694, upload-time = "2026-08-03T10:45:28.614Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/04/22/87f547ab4a84fa4094462e1e1cfdbadf28a539b5cd0efdb147b754d3b88a/langchain_openviking-0.1.0-py3-none-any.whl", hash = "sha256:ea32f08fa4bed81743e0d11e5f2f52e940c1464de74408d45bd574c081a4dffb", size = 55506, upload-time = "2026-08-03T10:45:27.421Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "langchain-protocol"
|
||||
version = "0.0.18"
|
||||
@ -3074,6 +3090,18 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/eb/a6/83dc2ab6fa397ee66fba04fe2e74bdf7be3b3870005359ceb7689103c058/opentelemetry_semantic_conventions-0.62b1-py3-none-any.whl", hash = "sha256:cf506938103d331fbb78eded0d9788095f7fd59016f2bda813c3324e5a74a93c", size = 231620, upload-time = "2026-04-24T13:15:35.454Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "openviking-sdk"
|
||||
version = "0.1.6"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "httpx" },
|
||||
]
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/6b/65/021ba0a777750d536b70e7ca5c14888d9e906627b80a9332b2c644082480/openviking_sdk-0.1.6.tar.gz", hash = "sha256:7ecfdebdbe3538556584e4348f19f1c990831c739404ee4e06e251f3d8a6641d", size = 40041, upload-time = "2026-08-03T07:15:38.963Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/6b/86/e0a9ff16d5b5eb38871450110b0ba64bff7a46020c617c482d4ce0ea3e7a/openviking_sdk-0.1.6-py3-none-any.whl", hash = "sha256:61e55f51f2733b80950488bb57299c5fd1bface388242ebff02c245ab6e04bd5", size = 25058, upload-time = "2026-08-03T07:15:37.776Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "orjson"
|
||||
version = "3.11.8"
|
||||
|
||||
@ -1699,17 +1699,18 @@ memory:
|
||||
# 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):
|
||||
# OpenViking example (replace this DeerMem backend_config block when
|
||||
# manager_class is openviking). This first official-adapter integration uses
|
||||
# one credential-bound OpenViking USER key for one DeerFlow user and supports
|
||||
# middleware mode only. Use owner_user_id: default when DeerFlow auth is off.
|
||||
#
|
||||
# backend_config:
|
||||
# base_url: http://openviking:1933
|
||||
# auth_mode: trusted
|
||||
# account: deerflow
|
||||
# owner_user_id: default
|
||||
# api_key_env: OPENVIKING_API_KEY
|
||||
# max_connections: 100
|
||||
# max_keepalive_connections: 20
|
||||
# max_seen_message_ids: 512 # recent-ID fallback for compacted histories
|
||||
# timeout_seconds: 30
|
||||
# default_peer_id: deerflow
|
||||
# max_seen_message_ids: 512 # bounded hash-only capture cursor
|
||||
# startup_policy: fail_fast
|
||||
# failure_policy:
|
||||
# read: fail_open
|
||||
@ -1718,6 +1719,10 @@ memory:
|
||||
# top_k: 8
|
||||
# score_threshold: 0.25
|
||||
# max_injection_chars: 12000
|
||||
# content_mode: overview
|
||||
# injection_query: >-
|
||||
# user profile preferences important entities events ongoing goals
|
||||
# constraints and prior decisions
|
||||
#
|
||||
# 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
|
||||
|
||||
@ -1,50 +1,73 @@
|
||||
# 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.
|
||||
backend. DeerMem remains the default. The OpenViking backend uses the maintained
|
||||
[`langchain-openviking`](https://pypi.org/project/langchain-openviking/)
|
||||
package instead of implementing OpenViking's HTTP protocol inside DeerFlow.
|
||||
|
||||
## Supported behavior
|
||||
## Current scope
|
||||
|
||||
- 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 bounded message watermarks under DeerFlow's runtime home. An ordered
|
||||
prefix digest handles append-only histories of any length, while a recent-ID
|
||||
window handles history compaction without resubmitting known messages.
|
||||
The first official-adapter integration deliberately preserves DeerFlow's
|
||||
existing automatic-memory behavior:
|
||||
|
||||
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.
|
||||
- memory is recalled through DeerFlow's existing fixed memory query;
|
||||
- completed turns are captured by the existing memory middleware;
|
||||
- messages about to be compacted are captured by the existing summarization hook;
|
||||
- every accepted capture is committed to the thread's stable OpenViking Session;
|
||||
- the official adapter handles message conversion, tool calls and results,
|
||||
100-message batching, partial-write progress, commit retry, and SDK transport;
|
||||
- one recorder-owned SDK client is shared with retrieval and closed through
|
||||
DeerFlow's existing memory shutdown contract.
|
||||
|
||||
## OpenViking requirements
|
||||
This backend supports `memory.mode: middleware`. It does not implement DeerMem
|
||||
fact CRUD, import/export, or the Settings memory-document view. OpenViking MCP
|
||||
tools are a separate integration surface and are not enabled by this backend.
|
||||
|
||||
OpenViking must be configured with:
|
||||
## Authentication boundary
|
||||
|
||||
- a VLM provider;
|
||||
- an embedding provider;
|
||||
- persistent workspace storage;
|
||||
- `server.auth_mode: trusted`;
|
||||
- a non-empty `server.root_api_key` when exposed beyond localhost.
|
||||
This version is for one DeerFlow user backed by one ordinary OpenViking **USER
|
||||
API key**. OpenViking derives the account and user from that credential.
|
||||
DeerFlow does not configure trusted account/user headers and must not receive a
|
||||
root key for normal memory traffic.
|
||||
|
||||
DeerFlow passes trusted `X-OpenViking-Account` and
|
||||
`X-OpenViking-User` headers. Do not expose a trusted-mode OpenViking endpoint
|
||||
directly to untrusted clients.
|
||||
The supported server configuration is OpenViking `api_key` mode, where the USER
|
||||
key determines the account and user. DeerFlow supplies its URL and API key
|
||||
explicitly, overrides any ambient actor peer during memory operations, and does
|
||||
not inherit arbitrary HTTP headers from `ovcli.conf`.
|
||||
|
||||
Before enabling this backend, remove legacy `OPENVIKING_ACCOUNT` and
|
||||
`OPENVIKING_USER` values from DeerFlow's repository-root `.env` and service
|
||||
environment, and remove `account` and `user` defaults from
|
||||
`~/.openviking/ovcli.conf`. Those settings belong to trusted-mode
|
||||
configurations and are outside this adapter's supported setup.
|
||||
|
||||
`owner_user_id` binds the configured key to one DeerFlow identity. Use
|
||||
`default` when DeerFlow authentication is disabled. In an authenticated
|
||||
single-user deployment, use that user's DeerFlow ID. A request for another
|
||||
DeerFlow user is rejected before OpenViking is contacted, preventing one USER
|
||||
key from silently sharing memory across users.
|
||||
|
||||
Multi-user credential provisioning and storage are intentionally outside this
|
||||
first adapter PR.
|
||||
|
||||
Existing trusted-mode configurations are not migrated automatically. Configure
|
||||
the OpenViking server in `api_key` mode, replace `auth_mode`, `account`, and the
|
||||
root key with `owner_user_id` and a USER key, and remove the legacy ambient
|
||||
identity settings listed above. Because the credential-bound user and Session
|
||||
mapping differ from the old trusted-user mapping, previously captured
|
||||
trusted-mode data remains in its old OpenViking namespace rather than being
|
||||
silently reassigned.
|
||||
|
||||
## Configure DeerFlow
|
||||
|
||||
Put the trusted OpenViking key in the repository root `.env`:
|
||||
Create or select an OpenViking user, then copy its USER API key into DeerFlow's
|
||||
repository-root `.env`:
|
||||
|
||||
```dotenv
|
||||
OPENVIKING_API_KEY=replace-with-the-same-root-api-key
|
||||
OPENVIKING_API_KEY=replace-with-an-openviking-user-api-key
|
||||
```
|
||||
|
||||
Replace the `memory` section in `config.yaml` with:
|
||||
Select the backend in `config.yaml`:
|
||||
|
||||
```yaml
|
||||
memory:
|
||||
@ -54,13 +77,9 @@ memory:
|
||||
manager_class: openviking
|
||||
mode: middleware
|
||||
backend_config:
|
||||
base_url: http://openviking:1933
|
||||
auth_mode: trusted
|
||||
account: deerflow
|
||||
base_url: http://127.0.0.1:1933
|
||||
owner_user_id: default
|
||||
api_key_env: OPENVIKING_API_KEY
|
||||
max_connections: 100
|
||||
max_keepalive_connections: 20
|
||||
max_seen_message_ids: 512
|
||||
startup_policy: fail_fast
|
||||
failure_policy:
|
||||
read: fail_open
|
||||
@ -69,118 +88,22 @@ memory:
|
||||
top_k: 8
|
||||
score_threshold: 0.25
|
||||
max_injection_chars: 12000
|
||||
content_mode: overview
|
||||
injection_query: >-
|
||||
user profile preferences important entities events ongoing goals
|
||||
constraints and prior decisions
|
||||
```
|
||||
|
||||
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`.
|
||||
For a host-installed OpenViking used by Docker DeerFlow, set `base_url` to
|
||||
`http://host.docker.internal:1933` and `allow_insecure_http: true`. The optional
|
||||
Compose overlay uses the internal `http://openviking:1933` address.
|
||||
|
||||
`max_connections` and `max_keepalive_connections` bound the shared HTTP
|
||||
connection pool. `max_seen_message_ids` bounds only the recent-ID fallback used
|
||||
when a conversation is compacted or rewritten; append-only histories are
|
||||
tracked by a constant-size prefix digest and do not depend on that window.
|
||||
The dependency on `langchain-openviking==0.1.0` is declared by DeerFlow's
|
||||
harness package and is installed by the normal `uv sync` flow.
|
||||
|
||||
## Docker first-time startup
|
||||
## Start the services
|
||||
|
||||
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 <http://localhost:2026> and OpenViking Studio at
|
||||
<http://localhost:1933/studio>.
|
||||
|
||||
## 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:
|
||||
For a local OpenViking process, start and verify the server first:
|
||||
|
||||
```bash
|
||||
openviking-server doctor
|
||||
@ -188,36 +111,67 @@ openviking-server
|
||||
curl http://127.0.0.1:1933/health
|
||||
```
|
||||
|
||||
Set `base_url: http://127.0.0.1:1933`, then start DeerFlow normally:
|
||||
Then start DeerFlow normally:
|
||||
|
||||
```bash
|
||||
make doctor
|
||||
make dev
|
||||
```
|
||||
|
||||
The DeerFlow entrypoint is <http://localhost:2026>.
|
||||
DeerFlow is available at <http://localhost:2026>. OpenViking Studio is
|
||||
available at <http://localhost:1933/studio>.
|
||||
|
||||
## Failure behavior
|
||||
To use the optional Docker service instead:
|
||||
|
||||
- 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.
|
||||
- Retried health, session lookup, and search requests use exponential backoff
|
||||
with jitter so concurrent Gateway workers do not retry in lockstep.
|
||||
- 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.
|
||||
```bash
|
||||
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
|
||||
|
||||
docker compose \
|
||||
-f docker/docker-compose.yaml \
|
||||
-f docker/docker-compose.openviking.yaml \
|
||||
up -d --build
|
||||
```
|
||||
|
||||
Configure OpenViking in API-key mode and obtain a USER key through its identity
|
||||
management flow. Only that USER key belongs in DeerFlow's
|
||||
`OPENVIKING_API_KEY` variable.
|
||||
|
||||
## Identity and session mapping
|
||||
|
||||
One DeerFlow thread maps deterministically to one OpenViking Session. A commit
|
||||
creates an archive inside that Session; it does not create a new Session, so a
|
||||
thread keeps the same identity when the user returns later.
|
||||
|
||||
The default DeerFlow agent uses `default_peer_id` (`deerflow` by default).
|
||||
Named agents use lowercase OpenViking peer IDs. Names that are not valid peer
|
||||
IDs, conflict with the default, or enter the reserved `df-agent-` namespace are
|
||||
mapped to collision-resistant IDs. USER-key identity remains the security
|
||||
boundary; peers separate memory scopes within that user.
|
||||
|
||||
## Retry and failure behavior
|
||||
|
||||
- `read: fail_open` logs retrieval failures and returns no injected OpenViking
|
||||
memory. `read: raise` propagates the retrieval failure to its DeerFlow caller.
|
||||
- `write: log_and_drop` logs capture failures without failing an already
|
||||
generated answer. `write: raise` propagates them.
|
||||
- DeerFlow stores only hashes and counters in a bounded local capture cursor
|
||||
under `{storage_path}/openviking/sessions/`. It never stores message text
|
||||
there.
|
||||
- The cursor prevents full LangGraph transcript snapshots from being submitted
|
||||
again. It also records confirmed progress from partial batches and retries a
|
||||
failed commit before appending more messages.
|
||||
- An unreadable cursor fails closed because replaying an unknown prefix could
|
||||
duplicate private conversation history.
|
||||
- Graceful shutdown stops new memory work, waits up to
|
||||
`shutdown_flush_timeout_seconds` for accepted operations, and closes the
|
||||
recorder-owned SDK client. It does not introduce a new DeerFlow lifecycle or
|
||||
background worker.
|
||||
|
||||
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.
|
||||
is still required. This initial integration does not claim at-least-once
|
||||
delivery.
|
||||
|
||||
@ -237,10 +237,11 @@ 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
|
||||
The optional `openviking` backend connects to an independent OpenViking server
|
||||
through `langchain-openviking` and currently supports one DeerFlow user in
|
||||
`mode: middleware`. Its private configuration uses `base_url`,
|
||||
`owner_user_id`, `api_key_env`, `failure_policy`, and `retrieval` with an
|
||||
ordinary OpenViking USER API key instead of the DeerMem fields shown above. See
|
||||
`docs/OPENVIKING.md` in the repository for the complete configuration and
|
||||
Docker startup sequence.
|
||||
|
||||
|
||||
@ -89,11 +89,13 @@ memory:
|
||||
max_injection_tokens: 2000
|
||||
```
|
||||
|
||||
## OpenViking HTTP backend
|
||||
## OpenViking 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.
|
||||
OpenViking server and recall its memory through the official
|
||||
`langchain-openviking` adapter. This first version supports one DeerFlow user
|
||||
with one OpenViking USER API key in middleware mode; DeerMem remains the
|
||||
default.
|
||||
|
||||
```yaml
|
||||
memory:
|
||||
@ -103,8 +105,7 @@ memory:
|
||||
mode: middleware
|
||||
backend_config:
|
||||
base_url: http://openviking:1933
|
||||
auth_mode: trusted
|
||||
account: deerflow
|
||||
owner_user_id: default
|
||||
api_key_env: OPENVIKING_API_KEY
|
||||
failure_policy:
|
||||
read: fail_open
|
||||
@ -115,9 +116,11 @@ memory:
|
||||
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`.
|
||||
Use `owner_user_id: default` when DeerFlow authentication is disabled. Put the
|
||||
USER key in the server environment, not directly in `config.yaml`. Trusted-mode
|
||||
account headers, root-key memory access, and multi-user key provisioning are not
|
||||
part of this version. See `docs/OPENVIKING.md` for the full boundary and startup
|
||||
guide.
|
||||
|
||||
## Global vs per-agent memory
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user