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
|
import os
import sys
import subprocess
from rich.console import Console
from rich.prompt import Prompt
from rich.panel import Panel
from cerebral.config import ORG_OUTPUT_DIR
from cerebral.llm import CerebrasProvider
from cerebral.agent import CerebralAgent
class CLIApp:
def __init__(self, agent: CerebralAgent, console: Console):
self.agent = agent
self.console = console
self.current_session_file = None
def run(self):
self.console.print(Panel.fit("🤖 [bold blue]Modular Cerebral Agent[/bold blue] initialized.\n- Type [bold]/image /path/to/img.png <prompt>[/bold] to attach images.\n- Type [bold]/exit[/bold] to quit.\n- Type [bold]/memory[/bold] for DB tools.", border_style="blue"))
while True:
try:
user_input = Prompt.ask("\n[bold magenta]You[/bold magenta]")
clean_input = user_input.strip().lower()
if clean_input == '/memory':
help_text = (
"[bold cyan]/memory count[/bold cyan] : Print the number of lines in persistent memory.\n"
"[bold cyan]/memory rebuild[/bold cyan] : Manually reserialize the FAISS database from the log.\n"
"[bold cyan]/memory compress[/bold cyan] : Use the local LLM to scrub duplicates and compress the log."
)
self.console.print(Panel.fit(help_text, title="🧠Memory Commands", border_style="cyan"))
continue
if clean_input == '/memory count':
count = self.agent.memory.get_line_count()
self.console.print(f"[bold cyan]Persistent Memory Lines:[/bold cyan] {count}")
continue
if clean_input == '/memory rebuild':
self.agent.memory.rebuild_index()
continue
if clean_input == '/memory compress':
self.agent.memory.compress_persistent_memory()
continue
if clean_input in ['/exit', '/quit']:
self.console.print("\n[dim italic]Initiating shutdown sequence...[/dim italic]")
self.agent.shutdown()
self.console.print("[bold red]Exiting...[/bold red]")
break
if not user_input.strip():
continue
image_path = None
prompt = user_input
if user_input.startswith("/image "):
parts = user_input.split(" ", 2)
if len(parts) >= 2:
image_path = parts[1]
prompt = parts[2] if len(parts) > 2 else "What is this?"
self.console.print(f"[dim italic]Processing image locally...[/dim italic]")
self.console.print("[bold green]Agent:[/bold green]")
if not self.current_session_file:
full_response = ""
for chunk in self.agent.chat_stream(prompt, image_path=image_path):
print(chunk, end="", flush=True)
full_response += chunk
print("\n")
generated_name = self.agent.generate_session_filename(prompt, full_response)
self.current_session_file = os.path.join(ORG_OUTPUT_DIR, generated_name)
self.console.print(f"[bold green]Session log created at:[/bold green] [cyan]{self.current_session_file}[/cyan]")
with open(self.current_session_file, "w") as f:
f.write(f"* User Prompt: {user_input}\n** Response\n{full_response}\n")
try:
self.console.print("[dim italic]Triggering emacsclient...[/dim italic]")
subprocess.run(
["emacsclient", "-n", self.current_session_file],
check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
)
except Exception as e:
self.console.print(f"[dim red]Failed to trigger emacsclient: {e}[/dim red]")
else:
full_response = ""
with open(self.current_session_file, "a") as f:
f.write(f"\n* User Prompt: {user_input}\n** Response\n")
for chunk in self.agent.chat_stream(prompt, image_path=image_path):
print(chunk, end="", flush=True)
f.write(chunk)
full_response += chunk
f.write("\n")
print("\n")
except KeyboardInterrupt:
self.console.print("\n[bold red]Interrupted. Saving memories...[/bold red]")
self.agent.shutdown()
break
except Exception as e:
self.console.print(f"[bold red]An error occurred: {e}[/bold red]")
if __name__ == "__main__":
console = Console()
try:
provider = CerebrasProvider()
except ValueError as e:
console.print(f"[bold red]Configuration Error: {e}[/bold red]")
sys.exit(1)
with console.status("[bold green]Booting up systems...[/bold green]", spinner="dots") as status:
agent = CerebralAgent(provider=provider, log=console.print)
app = CLIApp(agent, console)
app.run()
|
