Spaces:
Sleeping
Sleeping
File size: 5,157 Bytes
d5dc912 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | # SPDX-License-Identifier: MIT
"""bytecli memory — Semantic memory management (list, search, add)."""
from __future__ import annotations
import argparse
from typing import Any
from cli.client import ByteClient
from cli.config import CLIConfig
from cli.formatters import (
format_timestamp,
output,
print_header,
print_kv,
print_success,
print_table,
)
def register(subparsers: argparse._SubParsersAction) -> None:
parser = subparsers.add_parser(
"memory",
help="Semantic memory management (list, search, add)",
description=(
"Manage the agent's long-term semantic memory. The agent stores observations, "
"rules, and facts here to recall them in future conversations.\n\n"
"Endpoints: GET /api/memory, POST /api/memory, GET /api/memory/search"
),
epilog=(
"Examples:\n"
" bytecli memory list --limit 10\n"
" bytecli memory search 'favorite game'\n"
" bytecli memory add 'Streamer hates backseating' --tags rule,preference\n"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
sub = parser.add_subparsers(dest="memory_cmd", metavar="<action>")
parser.set_defaults(handler=_handle_list)
# list
p_list = sub.add_parser("list", help="List recent semantic memory entries")
p_list.add_argument("--limit", type=int, default=8, help="Max entries to return (default: 8)")
p_list.set_defaults(handler=_handle_list)
# search
p_search = sub.add_parser(
"search",
help="Search memory by text/similarity",
description="Searches semantic memory using embedding similarity or full-text search.",
)
p_search.add_argument("query", help="Search query text")
p_search.add_argument("--limit", type=int, default=8, help="Max results (default: 8)")
p_search.set_defaults(handler=_handle_search)
# add
p_add = sub.add_parser(
"add",
help="Add a new manually-authored memory entry",
description="Inserts a new semantic memory entry. The agent will read this context automatically based on relevance.",
)
p_add.add_argument("content", help="The memory content to store")
p_add.add_argument(
"--type", dest="memory_type", default="manual", help="Memory type (default: manual)"
)
p_add.add_argument("--tags", default="", help="Comma-separated tags (e.g. 'rule,preference')")
p_add.set_defaults(handler=_handle_add)
def _handle_list(args: argparse.Namespace, client: ByteClient, config: CLIConfig) -> None:
params: dict[str, str] = {"limit": str(args.limit)}
if config.channel != "default":
params["channel"] = config.channel
data = client.get("/api/semantic-memory", params)
output(data, json_mode=config.json_output, human_fn=_human_list)
def _human_list(data: dict[str, Any]) -> None:
entries = data.get("entries", [])
print_header(f"Semantic Memory ({len(entries)} entries)")
if not entries:
return
rows = []
for e in entries:
content = str(e.get("content", ""))[:50]
mem_type = str(e.get("memory_type", e.get("type", "—")))
tags = ", ".join(e.get("tags", []) or [])
rows.append([content, mem_type, tags[:30], format_timestamp(e.get("created_at"))])
print_table(["Content", "Type", "Tags", "Created"], rows)
def _handle_search(args: argparse.Namespace, client: ByteClient, config: CLIConfig) -> None:
params: dict[str, str] = {
"query": args.query,
"limit": str(args.limit),
}
if config.channel != "default":
params["channel"] = config.channel
data = client.get("/api/semantic-memory", params)
output(data, json_mode=config.json_output, human_fn=_human_search)
def _human_search(data: dict[str, Any]) -> None:
matches = data.get("matches", [])
diag = data.get("search_diagnostics", {})
print_header(f"Search Results ({len(matches)} matches)")
if diag:
print_kv(
{
"Engine": diag.get("engine", "—"),
"Candidates": diag.get("candidate_count", 0),
"Min Similarity": diag.get("min_similarity", "—"),
}
)
print()
if not matches:
return
rows = []
for m in matches:
content = str(m.get("content", ""))[:50]
similarity = m.get("similarity", m.get("score", "—"))
if isinstance(similarity, float):
similarity = f"{similarity:.4f}"
rows.append([content, str(similarity)])
print_table(["Content", "Similarity"], rows)
def _handle_add(args: argparse.Namespace, client: ByteClient, config: CLIConfig) -> None:
payload: dict[str, Any] = {
"content": args.content,
"memory_type": args.memory_type,
"channel_id": config.channel,
}
if args.tags:
payload["tags"] = [t.strip() for t in args.tags.split(",") if t.strip()]
data = client.put("/api/semantic-memory", payload)
output(
data, json_mode=config.json_output, human_fn=lambda d: print_success("Memory entry added.")
)
|