Spaces:
Runtime error
Runtime error
File size: 6,646 Bytes
b74c6ae | 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 | """The agent's tools for earlier chats, scoped to one notebook's chat memory.
Paths are shown as `chats/{chat_id}/{topic}.md` so a citation tells a chat file from a post (`sources/...`). Only
topic files can be searched or read: the corpus root is the notebook's own area, and `clean_rel` rejects anything that
climbs out of it."""
import re
from app.chat_memory import files, search
from app.corpus import MAX_LINE_CHARS, MAX_READ_LINES, MAX_SEARCH_HITS, Corpus, CorpusError
PREFIX = "chats/"
def is_chat_path(path: str) -> bool:
return path.strip().lstrip("/").startswith(PREFIX)
class ChatTools:
def __init__(self, corpus: Corpus, on_step=None):
self.corpus = corpus
self.on_step = on_step or (lambda kind, detail: None)
# Paths
def _rel(self, path: str) -> str:
rel = Corpus.clean_rel(path)
rel = rel.removeprefix(PREFIX)
name = rel.rsplit("/", 1)[-1]
if not rel.endswith(".md") or name in (files.TOPICS, files.INDEX):
raise CorpusError(f"{path} is not an earlier chat topic. Use list_chat_topics to see them.")
if not self.corpus.exists(rel):
raise CorpusError(f"No such earlier chat topic: {path}. Use list_chat_topics to see them.")
return rel
def topic_files(self) -> list[str]:
manifest = self.corpus._read_manifest(remote=False)
skip = (files.TOPICS, files.INDEX)
return sorted(r for r in manifest if "/" in r and r.endswith(".md") and r.rsplit("/", 1)[-1] not in skip)
@staticmethod
def shown(rel: str) -> str:
return PREFIX + rel
# Tools
def list_chat_topics(self, contains: str = "", since: str = "") -> str:
"""List the topics of earlier chats in this notebook, newest first, one per line as
`date | path | topic | keywords | gist`. Filter to lines containing `contains` (case insensitive) and to
topics last active on or after `since` (a date like 2026-09-24)."""
self.on_step("recall", f"topics of earlier chats{f' about {contains}' if contains else ''}")
if not self.corpus.exists(files.INDEX):
return "No earlier chat topics yet."
rows = []
for line in self.corpus.read_lines(files.INDEX):
parts = [p.strip() for p in line.split("|")]
if len(parts) < 3:
continue
if since and parts[0] < since:
continue
if contains and contains.lower() not in line.lower():
continue
rows.append(" | ".join([parts[0], self.shown(parts[1]), *parts[2:]]))
return "\n".join(rows[:60]) or "No matching topics."
def search_chats(self, query: str, since: str = "", exact: bool = False) -> str:
"""Ranked search of earlier chats in this notebook (best matches first). Matches the words used and
the topic's keywords, so a related word can find a chat that used another word. Optionally only topics last
active on or after `since` (a date like 2026-09-24). Set exact=true to match `query` as a regular expression
instead, for a literal phrase or a name. Returns `path:line: [topic, date] text`."""
if exact:
return self.grep_chats(query)
self.on_step("recall", f"earlier chats for {query}")
manifest = self.corpus._read_manifest(remote=False)
key = (str(self.corpus.root), tuple(sorted(manifest.items())))
def load():
units = []
for rel in self.topic_files():
units += search.units_for(rel, "\n".join(self.corpus.read_lines(rel)))
return units
hits = search.cut(search.index_for(key, load).search(query, since))
if not hits:
return "No matches in earlier chats. Try other words, list_chat_topics, or exact=true for a phrase."
return "\n".join(f"{self.shown(u.path)}:{u.line}: [{u.topic}, {u.last}] "
f"{u.text.strip().replace(chr(10), ' ')[:search.SNIPPET_CHARS]}" for _, u in hits)
def grep_chats(self, pattern: str, path: str = "") -> str:
"""Search the text of earlier chat topics with a case insensitive regular expression (like grep), for an
exact word or phrase. Optionally limit to one `path`. Returns `path:line: text` matches. The agent reaches
this through search_chats(exact=true): this model stops answering when it is given more than six tools."""
self.on_step("recall", f"earlier chats for {pattern}")
try:
rx = re.compile(pattern, re.IGNORECASE)
except re.error:
rx = re.compile(re.escape(pattern), re.IGNORECASE)
targets = [self._rel(path)] if path else self.topic_files()
hits: list[str] = []
total = 0
for rel in targets:
for n, line in enumerate(self.corpus.read_lines(rel), start=1):
if rx.search(line):
total += 1
if len(hits) < MAX_SEARCH_HITS:
hits.append(f"{self.shown(rel)}:{n}: {line.strip()[:MAX_LINE_CHARS]}")
if not hits:
return "No matches in earlier chats."
out = "\n".join(hits)
return out + (f"\n... {total - len(hits)} more matches. Narrow the pattern." if total > len(hits) else "")
def read_chat(self, path: str, start_line: int = 1, end_line: int = 0) -> str:
"""Read an earlier chat topic with line numbers (its summary, then the exact words that were kept). Reads at
most 160 lines per call. Cite with these line numbers."""
rel = self._rel(path)
lines = self.corpus.read_lines(rel)
start = max(1, start_line)
end = min(len(lines), end_line if end_line >= start else start + MAX_READ_LINES - 1)
end = min(end, start + MAX_READ_LINES - 1)
self.on_step("recall", f"{rel} (lines {start} to {end})")
body = "\n".join(f"{n}| {lines[n - 1]}" for n in range(start, end + 1))
more = f"\n[{len(lines) - end} more lines]" if end < len(lines) else ""
return f"{self.shown(rel)} ({len(lines)} lines)\n{body}{more}"
def quote(self, path: str, start: int, end: int) -> str | None:
"""Text of a cited line range, or None if the citation does not point at real content."""
try:
rel = self._rel(path)
lines = self.corpus.read_lines(rel)
except CorpusError:
return None
if start < 1 or start > len(lines):
return None
end = max(start, min(end, len(lines), start + 40))
text = "\n".join(lines[start - 1:end]).strip()
return text or None
|