532 lines
19 KiB
Python
532 lines
19 KiB
Python
"""Memory API router for retrieving and managing memory data.
|
|
|
|
All endpoints accept an optional ``agent`` query parameter:
|
|
|
|
- omitted / empty / ``"global"`` -> operate on the per-user **global**
|
|
memory file (``users/{uid}/memory.json``).
|
|
- valid agent name -> operate on the **per-agent** memory file
|
|
(``users/{uid}/agents/{agent}/memory.json``).
|
|
|
|
Without this parameter the API used to be silently global-only, so any
|
|
memory written while chatting inside a custom agent (which is saved
|
|
per-agent) was invisible from the settings page. See ``GET /api/memory/agents``
|
|
for discovering which agent-scoped buckets the current user owns.
|
|
"""
|
|
|
|
import re
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
|
|
from deerflow.agents.memory.updater import (
|
|
clear_memory_data,
|
|
create_memory_fact,
|
|
delete_memory_fact,
|
|
get_memory_data,
|
|
import_memory_data,
|
|
reload_memory_data,
|
|
update_memory_fact,
|
|
)
|
|
from deerflow.config.agents_config import AGENT_NAME_PATTERN
|
|
from deerflow.config.memory_config import get_memory_config
|
|
from deerflow.config.paths import get_paths
|
|
from deerflow.runtime.user_context import get_effective_user_id
|
|
|
|
router = APIRouter(prefix="/api", tags=["memory"])
|
|
|
|
|
|
# Sentinel values that mean "the per-user global memory file" (no agent).
|
|
# We accept several spellings so the frontend can model the dropdown choice
|
|
# either by omitting the param or by sending an explicit "global"/"".
|
|
_GLOBAL_AGENT_VALUES: frozenset[str] = frozenset({"", "global", "__global__"})
|
|
|
|
|
|
def _resolve_agent_name(agent: str | None) -> str | None:
|
|
"""Validate the ``agent`` query parameter and normalize the global sentinel.
|
|
|
|
Returns ``None`` for the global memory bucket, or the lowercased agent
|
|
name otherwise. Raises 400 if the supplied name fails the agent-name
|
|
pattern (mirrors :func:`FileMemoryStorage._validate_agent_name`, but
|
|
surfaces a clean HTTP error instead of letting the storage layer raise
|
|
``ValueError`` via a 500).
|
|
"""
|
|
if agent is None:
|
|
return None
|
|
stripped = agent.strip()
|
|
if stripped.lower() in _GLOBAL_AGENT_VALUES:
|
|
return None
|
|
if not AGENT_NAME_PATTERN.match(stripped):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid agent name {stripped!r}: must match {AGENT_NAME_PATTERN.pattern}",
|
|
)
|
|
return stripped.lower()
|
|
|
|
|
|
_AgentQuery = Query(
|
|
None,
|
|
description=(
|
|
"Optional agent name. Omit (or pass 'global') to address the per-user "
|
|
"global memory bucket; pass a custom-agent name to address that "
|
|
"agent's per-user memory bucket."
|
|
),
|
|
examples=["my-agent"],
|
|
pattern=r"^[A-Za-z0-9-]+$|^global$|^$",
|
|
)
|
|
|
|
|
|
class ContextSection(BaseModel):
|
|
"""Model for context sections (user and history)."""
|
|
|
|
summary: str = Field(default="", description="Summary content")
|
|
updatedAt: str = Field(default="", description="Last update timestamp")
|
|
|
|
|
|
class UserContext(BaseModel):
|
|
"""Model for user context."""
|
|
|
|
workContext: ContextSection = Field(default_factory=ContextSection)
|
|
personalContext: ContextSection = Field(default_factory=ContextSection)
|
|
topOfMind: ContextSection = Field(default_factory=ContextSection)
|
|
|
|
|
|
class HistoryContext(BaseModel):
|
|
"""Model for history context."""
|
|
|
|
recentMonths: ContextSection = Field(default_factory=ContextSection)
|
|
earlierContext: ContextSection = Field(default_factory=ContextSection)
|
|
longTermBackground: ContextSection = Field(default_factory=ContextSection)
|
|
|
|
|
|
class Fact(BaseModel):
|
|
"""Model for a memory fact."""
|
|
|
|
id: str = Field(..., description="Unique identifier for the fact")
|
|
content: str = Field(..., description="Fact content")
|
|
category: str = Field(default="context", description="Fact category")
|
|
confidence: float = Field(default=0.5, description="Confidence score (0-1)")
|
|
createdAt: str = Field(default="", description="Creation timestamp")
|
|
source: str = Field(default="unknown", description="Source thread ID")
|
|
sourceError: str | None = Field(default=None, description="Optional description of the prior mistake or wrong approach")
|
|
|
|
|
|
class MemoryResponse(BaseModel):
|
|
"""Response model for memory data."""
|
|
|
|
version: str = Field(default="1.0", description="Memory schema version")
|
|
lastUpdated: str = Field(default="", description="Last update timestamp")
|
|
user: UserContext = Field(default_factory=UserContext)
|
|
history: HistoryContext = Field(default_factory=HistoryContext)
|
|
facts: list[Fact] = Field(default_factory=list)
|
|
|
|
|
|
def _map_memory_fact_value_error(exc: ValueError) -> HTTPException:
|
|
"""Convert updater validation errors into stable API responses."""
|
|
if exc.args and exc.args[0] == "confidence":
|
|
detail = "Invalid confidence value; must be between 0 and 1."
|
|
else:
|
|
detail = "Memory fact content cannot be empty."
|
|
return HTTPException(status_code=400, detail=detail)
|
|
|
|
|
|
class FactCreateRequest(BaseModel):
|
|
"""Request model for creating a memory fact."""
|
|
|
|
content: str = Field(..., min_length=1, description="Fact content")
|
|
category: str = Field(default="context", description="Fact category")
|
|
confidence: float = Field(default=0.5, ge=0.0, le=1.0, description="Confidence score (0-1)")
|
|
|
|
|
|
class FactPatchRequest(BaseModel):
|
|
"""PATCH request model that preserves existing values for omitted fields."""
|
|
|
|
content: str | None = Field(default=None, min_length=1, description="Fact content")
|
|
category: str | None = Field(default=None, description="Fact category")
|
|
confidence: float | None = Field(default=None, ge=0.0, le=1.0, description="Confidence score (0-1)")
|
|
|
|
|
|
class MemoryConfigResponse(BaseModel):
|
|
"""Response model for memory configuration."""
|
|
|
|
enabled: bool = Field(..., description="Whether memory is enabled")
|
|
storage_path: str = Field(..., description="Path to memory storage file")
|
|
debounce_seconds: int = Field(..., description="Debounce time for memory updates")
|
|
max_facts: int = Field(..., description="Maximum number of facts to store")
|
|
fact_confidence_threshold: float = Field(..., description="Minimum confidence threshold for facts")
|
|
injection_enabled: bool = Field(..., description="Whether memory injection is enabled")
|
|
max_injection_tokens: int = Field(..., description="Maximum tokens for memory injection")
|
|
|
|
|
|
class MemoryStatusResponse(BaseModel):
|
|
"""Response model for memory status."""
|
|
|
|
config: MemoryConfigResponse
|
|
data: MemoryResponse
|
|
|
|
|
|
class MemoryAgentInfo(BaseModel):
|
|
"""One per-agent memory bucket the current user owns."""
|
|
|
|
name: str = Field(..., description="Custom agent name (matches AGENT_NAME_PATTERN).")
|
|
factCount: int = Field(0, description="Number of facts stored for this agent.")
|
|
lastUpdated: str = Field("", description="lastUpdated field of the per-agent memory file, if any.")
|
|
|
|
|
|
class MemoryAgentsResponse(BaseModel):
|
|
"""Inventory of memory buckets visible to the current user."""
|
|
|
|
globalAvailable: bool = Field(..., description="Whether a per-user global memory file exists on disk.")
|
|
globalFactCount: int = Field(0, description="Facts in the per-user global memory file.")
|
|
globalLastUpdated: str = Field("", description="lastUpdated field of the global memory file, if any.")
|
|
agents: list[MemoryAgentInfo] = Field(default_factory=list, description="Sorted by name (case-insensitive).")
|
|
|
|
|
|
@router.get(
|
|
"/memory/agents",
|
|
response_model=MemoryAgentsResponse,
|
|
summary="List Memory Buckets",
|
|
description=(
|
|
"Enumerate which memory buckets exist for the current user: the "
|
|
"per-user global file, plus every per-agent memory file the user "
|
|
"has accumulated by chatting inside custom agents. Drives the "
|
|
"agent-selector dropdown on the Memory settings page."
|
|
),
|
|
)
|
|
async def list_memory_agents() -> MemoryAgentsResponse:
|
|
"""Inspect the on-disk memory layout for the current user.
|
|
|
|
This intentionally walks the filesystem instead of querying the
|
|
custom-agents API: a user can have leftover per-agent memory for an
|
|
agent they later renamed/deleted, and we still want them to be able to
|
|
open it (or clear it) from the UI.
|
|
"""
|
|
import json
|
|
|
|
user_id = get_effective_user_id()
|
|
paths = get_paths()
|
|
user_dir = paths.user_dir(user_id)
|
|
|
|
# Global bucket
|
|
global_file = paths.user_memory_file(user_id)
|
|
global_available = global_file.exists()
|
|
global_facts = 0
|
|
global_updated = ""
|
|
if global_available:
|
|
try:
|
|
with open(global_file, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
global_facts = len(data.get("facts", []) or [])
|
|
global_updated = data.get("lastUpdated", "") or ""
|
|
except (OSError, json.JSONDecodeError):
|
|
# Treat unreadable file as "exists but empty" — better than 500
|
|
# since this endpoint is purely informational.
|
|
pass
|
|
|
|
agents: list[MemoryAgentInfo] = []
|
|
agents_root = user_dir / "agents"
|
|
if agents_root.is_dir():
|
|
for entry in sorted(agents_root.iterdir(), key=lambda p: p.name.lower()):
|
|
if not entry.is_dir():
|
|
continue
|
|
mem_file = entry / "memory.json"
|
|
if not mem_file.exists():
|
|
continue
|
|
# Defensive: agent-dir name may have been hand-edited; keep only
|
|
# those that round-trip through AGENT_NAME_PATTERN so the UI
|
|
# never receives a value it can't echo back to us.
|
|
if not AGENT_NAME_PATTERN.match(entry.name):
|
|
continue
|
|
facts = 0
|
|
updated = ""
|
|
try:
|
|
with open(mem_file, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
facts = len(data.get("facts", []) or [])
|
|
updated = data.get("lastUpdated", "") or ""
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
agents.append(MemoryAgentInfo(name=entry.name, factCount=facts, lastUpdated=updated))
|
|
|
|
return MemoryAgentsResponse(
|
|
globalAvailable=global_available,
|
|
globalFactCount=global_facts,
|
|
globalLastUpdated=global_updated,
|
|
agents=agents,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/memory",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Get Memory Data",
|
|
description=(
|
|
"Retrieve the current memory data including user context, history, and facts. "
|
|
"Pass ``?agent=<name>`` to read a per-agent bucket; omit it to read the per-user global bucket."
|
|
),
|
|
)
|
|
async def get_memory(agent: str | None = _AgentQuery) -> MemoryResponse:
|
|
"""Get the current memory data for the global or per-agent bucket.
|
|
|
|
Returns:
|
|
The current memory data with user context, history, and facts.
|
|
|
|
Example Response:
|
|
```json
|
|
{
|
|
"version": "1.0",
|
|
"lastUpdated": "2024-01-15T10:30:00Z",
|
|
"user": {
|
|
"workContext": {"summary": "Working on DeerFlow project", "updatedAt": "..."},
|
|
"personalContext": {"summary": "Prefers concise responses", "updatedAt": "..."},
|
|
"topOfMind": {"summary": "Building memory API", "updatedAt": "..."}
|
|
},
|
|
"history": {
|
|
"recentMonths": {"summary": "Recent development activities", "updatedAt": "..."},
|
|
"earlierContext": {"summary": "", "updatedAt": ""},
|
|
"longTermBackground": {"summary": "", "updatedAt": ""}
|
|
},
|
|
"facts": [
|
|
{
|
|
"id": "fact_abc123",
|
|
"content": "User prefers TypeScript over JavaScript",
|
|
"category": "preference",
|
|
"confidence": 0.9,
|
|
"createdAt": "2024-01-15T10:30:00Z",
|
|
"source": "thread_xyz"
|
|
}
|
|
]
|
|
}
|
|
```
|
|
"""
|
|
agent_name = _resolve_agent_name(agent)
|
|
memory_data = get_memory_data(agent_name, user_id=get_effective_user_id())
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.post(
|
|
"/memory/reload",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Reload Memory Data",
|
|
description="Reload memory data from the storage file, refreshing the in-memory cache.",
|
|
)
|
|
async def reload_memory(agent: str | None = _AgentQuery) -> MemoryResponse:
|
|
"""Reload memory data from file.
|
|
|
|
This forces a reload of the memory data from the storage file,
|
|
useful when the file has been modified externally.
|
|
|
|
Returns:
|
|
The reloaded memory data.
|
|
"""
|
|
agent_name = _resolve_agent_name(agent)
|
|
memory_data = reload_memory_data(agent_name, user_id=get_effective_user_id())
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.delete(
|
|
"/memory",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Clear All Memory Data",
|
|
description="Delete all saved memory data and reset the memory structure to an empty state.",
|
|
)
|
|
async def clear_memory(agent: str | None = _AgentQuery) -> MemoryResponse:
|
|
"""Clear all persisted memory data for the requested bucket."""
|
|
agent_name = _resolve_agent_name(agent)
|
|
try:
|
|
memory_data = clear_memory_data(agent_name, user_id=get_effective_user_id())
|
|
except OSError as exc:
|
|
raise HTTPException(status_code=500, detail="Failed to clear memory data.") from exc
|
|
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.post(
|
|
"/memory/facts",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Create Memory Fact",
|
|
description="Create a single saved memory fact manually.",
|
|
)
|
|
async def create_memory_fact_endpoint(
|
|
request: FactCreateRequest,
|
|
agent: str | None = _AgentQuery,
|
|
) -> MemoryResponse:
|
|
"""Create a single fact manually."""
|
|
agent_name = _resolve_agent_name(agent)
|
|
try:
|
|
memory_data = create_memory_fact(
|
|
content=request.content,
|
|
category=request.category,
|
|
confidence=request.confidence,
|
|
agent_name=agent_name,
|
|
user_id=get_effective_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
raise _map_memory_fact_value_error(exc) from exc
|
|
except OSError as exc:
|
|
raise HTTPException(status_code=500, detail="Failed to create memory fact.") from exc
|
|
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.delete(
|
|
"/memory/facts/{fact_id}",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Delete Memory Fact",
|
|
description="Delete a single saved memory fact by its fact id.",
|
|
)
|
|
async def delete_memory_fact_endpoint(
|
|
fact_id: str,
|
|
agent: str | None = _AgentQuery,
|
|
) -> MemoryResponse:
|
|
"""Delete a single fact from memory by fact id."""
|
|
agent_name = _resolve_agent_name(agent)
|
|
try:
|
|
memory_data = delete_memory_fact(fact_id, agent_name, user_id=get_effective_user_id())
|
|
except KeyError as exc:
|
|
raise HTTPException(status_code=404, detail=f"Memory fact '{fact_id}' not found.") from exc
|
|
except OSError as exc:
|
|
raise HTTPException(status_code=500, detail="Failed to delete memory fact.") from exc
|
|
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.patch(
|
|
"/memory/facts/{fact_id}",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Patch Memory Fact",
|
|
description="Partially update a single saved memory fact by its fact id while preserving omitted fields.",
|
|
)
|
|
async def update_memory_fact_endpoint(
|
|
fact_id: str,
|
|
request: FactPatchRequest,
|
|
agent: str | None = _AgentQuery,
|
|
) -> MemoryResponse:
|
|
"""Partially update a single fact manually."""
|
|
agent_name = _resolve_agent_name(agent)
|
|
try:
|
|
memory_data = update_memory_fact(
|
|
fact_id=fact_id,
|
|
content=request.content,
|
|
category=request.category,
|
|
confidence=request.confidence,
|
|
agent_name=agent_name,
|
|
user_id=get_effective_user_id(),
|
|
)
|
|
except ValueError as exc:
|
|
raise _map_memory_fact_value_error(exc) from exc
|
|
except KeyError as exc:
|
|
raise HTTPException(status_code=404, detail=f"Memory fact '{fact_id}' not found.") from exc
|
|
except OSError as exc:
|
|
raise HTTPException(status_code=500, detail="Failed to update memory fact.") from exc
|
|
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.get(
|
|
"/memory/export",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Export Memory Data",
|
|
description="Export the current global memory data as JSON for backup or transfer.",
|
|
)
|
|
async def export_memory(agent: str | None = _AgentQuery) -> MemoryResponse:
|
|
"""Export the current memory data."""
|
|
agent_name = _resolve_agent_name(agent)
|
|
memory_data = get_memory_data(agent_name, user_id=get_effective_user_id())
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.post(
|
|
"/memory/import",
|
|
response_model=MemoryResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Import Memory Data",
|
|
description="Import and overwrite the current global memory data from a JSON payload.",
|
|
)
|
|
async def import_memory(
|
|
request: MemoryResponse,
|
|
agent: str | None = _AgentQuery,
|
|
) -> MemoryResponse:
|
|
"""Import and persist memory data."""
|
|
agent_name = _resolve_agent_name(agent)
|
|
try:
|
|
memory_data = import_memory_data(request.model_dump(), agent_name, user_id=get_effective_user_id())
|
|
except OSError as exc:
|
|
raise HTTPException(status_code=500, detail="Failed to import memory data.") from exc
|
|
|
|
return MemoryResponse(**memory_data)
|
|
|
|
|
|
@router.get(
|
|
"/memory/config",
|
|
response_model=MemoryConfigResponse,
|
|
summary="Get Memory Configuration",
|
|
description="Retrieve the current memory system configuration.",
|
|
)
|
|
async def get_memory_config_endpoint() -> MemoryConfigResponse:
|
|
"""Get the memory system configuration.
|
|
|
|
Returns:
|
|
The current memory configuration settings.
|
|
|
|
Example Response:
|
|
```json
|
|
{
|
|
"enabled": true,
|
|
"storage_path": ".deer-flow/memory.json",
|
|
"debounce_seconds": 30,
|
|
"max_facts": 100,
|
|
"fact_confidence_threshold": 0.7,
|
|
"injection_enabled": true,
|
|
"max_injection_tokens": 2000
|
|
}
|
|
```
|
|
"""
|
|
config = get_memory_config()
|
|
return MemoryConfigResponse(
|
|
enabled=config.enabled,
|
|
storage_path=config.storage_path,
|
|
debounce_seconds=config.debounce_seconds,
|
|
max_facts=config.max_facts,
|
|
fact_confidence_threshold=config.fact_confidence_threshold,
|
|
injection_enabled=config.injection_enabled,
|
|
max_injection_tokens=config.max_injection_tokens,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/memory/status",
|
|
response_model=MemoryStatusResponse,
|
|
response_model_exclude_none=True,
|
|
summary="Get Memory Status",
|
|
description="Retrieve both memory configuration and current data in a single request.",
|
|
)
|
|
async def get_memory_status(agent: str | None = _AgentQuery) -> MemoryStatusResponse:
|
|
"""Get the memory system status including configuration and data.
|
|
|
|
Returns:
|
|
Combined memory configuration and current data.
|
|
"""
|
|
agent_name = _resolve_agent_name(agent)
|
|
config = get_memory_config()
|
|
memory_data = get_memory_data(agent_name, user_id=get_effective_user_id())
|
|
|
|
return MemoryStatusResponse(
|
|
config=MemoryConfigResponse(
|
|
enabled=config.enabled,
|
|
storage_path=config.storage_path,
|
|
debounce_seconds=config.debounce_seconds,
|
|
max_facts=config.max_facts,
|
|
fact_confidence_threshold=config.fact_confidence_threshold,
|
|
injection_enabled=config.injection_enabled,
|
|
max_injection_tokens=config.max_injection_tokens,
|
|
),
|
|
data=MemoryResponse(**memory_data),
|
|
)
|