mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-20 11:42:23 +02:00
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
"""CRUD for the user's memory facts; the Settings > Memory page is the only intended client."""
|
|
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any, AsyncIterator, Dict, List
|
|
|
|
from fastapi import HTTPException
|
|
from pydantic import BaseModel
|
|
from typeguard import typechecked
|
|
|
|
from backend.config.Apps import SubApp
|
|
from backend.apps.memory.store import MemoryFact, MemoryOp, add_fact, apply_ops, delete_fact, list_facts, update_fact
|
|
|
|
|
|
@asynccontextmanager
|
|
async def memory_lifespan() -> AsyncIterator[None]:
|
|
yield
|
|
|
|
|
|
memory = SubApp("memory", memory_lifespan)
|
|
|
|
|
|
class FactBody(BaseModel):
|
|
text: str
|
|
|
|
|
|
@memory.router.get("")
|
|
@typechecked
|
|
async def get_facts() -> Dict[str, List[MemoryFact]]:
|
|
return {"facts": list_facts()}
|
|
|
|
|
|
@memory.router.post("")
|
|
@typechecked
|
|
async def create_fact(body: FactBody) -> MemoryFact:
|
|
fact = add_fact(body.text, source="user")
|
|
if fact is None:
|
|
raise HTTPException(status_code=400, detail="Empty fact, or the memory list is full (60 max); delete something first.")
|
|
return fact
|
|
|
|
|
|
class OpsBody(BaseModel):
|
|
ops: List[MemoryOp]
|
|
|
|
|
|
@memory.router.post("/ops")
|
|
@typechecked
|
|
async def apply_memory_ops(body: OpsBody) -> Dict[str, Any]:
|
|
"""Atomic batch for the MemoryWrite tool: all ops land or none do, cap checked on the final state."""
|
|
if not body.ops:
|
|
raise HTTPException(status_code=400, detail="ops must be a non-empty list.")
|
|
return apply_ops(body.ops).model_dump()
|
|
|
|
|
|
@memory.router.patch("/{fact_id}")
|
|
@typechecked
|
|
async def edit_fact(fact_id: str, body: FactBody) -> MemoryFact:
|
|
fact = update_fact(fact_id, body.text)
|
|
if fact is None:
|
|
raise HTTPException(status_code=404, detail="No such fact (or the new text is empty).")
|
|
return fact
|
|
|
|
|
|
@memory.router.post("/distill/{session_id}")
|
|
@typechecked
|
|
async def distill(session_id: str) -> Dict[str, List[str]]:
|
|
from backend.apps.agents.agents import agent_manager
|
|
from backend.apps.memory.distill import distill_session_memory
|
|
session = agent_manager.sessions.get(session_id)
|
|
if not session:
|
|
try:
|
|
session = await agent_manager.resume_session(session_id)
|
|
except ValueError:
|
|
return {"added": []}
|
|
return {"added": await distill_session_memory(session)}
|
|
|
|
|
|
@memory.router.delete("/{fact_id}")
|
|
@typechecked
|
|
async def remove_fact(fact_id: str) -> Dict[str, bool]:
|
|
if not delete_fact(fact_id):
|
|
raise HTTPException(status_code=404, detail="No such fact.")
|
|
return {"ok": True}
|