backstory/src/rag_cli.py
2025-03-18 13:09:52 -07:00

162 lines
5.3 KiB
Python

# cli.py - Command-line interface
import sys
class RagCLI:
"""Command-line interface for the RAG application"""
def __init__(self, core):
self.core = core
def show_help(self):
"""Display available commands"""
print("""help>
docs Show RAG docs
full Show last full top match
headlines Show the RAG headlines
prompt Show the last prompt
response Show the last response
scores Show last RAG scores
why|think Show last response's <think>
context|match Show RAG match info to last prompt
exit|quit Exit the application
""")
def print_document_summary(self, documents):
"""Print brief summary of all documents"""
if not documents:
print("No documents available.")
return
for i, doc in enumerate(documents):
print(f"Document {i+1}:")
print(f" Title: {doc['title']}")
print(f" Text: {doc['text'][:100]}...")
print()
def print_headlines(self, headlines):
"""Print headlines"""
if not headlines:
print("No headlines available.")
return
for source, title in headlines:
print(f"{source}: {title}")
def print_match_info(self, match_info):
"""Print detailed information about a matching document"""
if not match_info:
print("No matching document found.")
return
print("="*50)
print("MATCHING DOCUMENT")
print("="*50)
print(f"Chunk ID: {match_info['id']}")
print(f"Similarity Score: {match_info['distance']:.4f}")
print("\nCHUNK METADATA:")
for key, value in match_info['metadata'].items():
print(f" {key}: {value}")
print("\nMATCHING CHUNK CONTENT:")
print(match_info['chunk'][:500].strip() + ("..." if len(match_info['chunk']) > 500 else ""))
def print_full_document(self, doc, distance=None):
"""Print full document details"""
if not doc:
print("No document found.")
return
print(f"""Context:
Source: {doc["source"]}
Title: {doc["title"]}
Link: {doc["link"]}
{"Distance: " + str(distance) if distance is not None else ""}
Full text:
{doc["text"]}""")
def run(self):
"""Run the command-line interface"""
# Load documents
doc_count = self.core.load_documents()
print(f"Loaded {doc_count} documents.")
while True:
try:
query = input("> ").strip()
except KeyboardInterrupt:
print("\nExiting.")
break
if query in ["exit", "quit"]:
print("\nExiting.")
break
elif query == "docs":
self.print_document_summary(self.core.get_documents())
elif query == "prompt":
last_prompt = self.core.get_last_prompt()
last_system = self.core.get_last_system()
if last_prompt:
print(f"""last prompt>
{"="*10}system{"="*10}
{last_system}
{"="*10}prompt{"="*10}
{last_prompt}""")
else:
print("No prompts yet")
elif query == "response":
last_response = self.core.get_last_response()
if last_response:
print(f"""last response>
{"="*10}response{"="*10}
{last_response}""")
else:
print("No responses yet")
elif query in ["", "help"]:
self.show_help()
elif query == "headlines":
self.print_headlines(self.core.get_headlines())
elif query in ["match", "context"]:
match_info = self.core.get_match_info()
self.print_match_info(match_info)
elif query in ["why", "think"]:
last_thinking = self.core.get_last_thinking()
if last_thinking:
print(f"""
why>
{last_thinking}
""")
else:
print("No processed prompts")
elif query == "scores":
match_count = self.core.get_match_count()
if match_count > 0:
for i in range(match_count):
match_info = self.core.get_match_info(index=i)
self.print_match_info(match_info)
else:
print("No matches to display")
elif query == "full":
doc = self.core.get_top_match()
distance = None
if self.core.last_results and self.core.last_results.get("distances"):
distance = self.core.last_results["distances"][0][0]
self.print_full_document(doc, distance)
else:
# Process the user query
response, debug_info = self.core.process_query(query)
print(f"system len: {debug_info['system_len']}")
print(f"prompt len: {debug_info['prompt_len']}")
print(f"Response>\n{response}")