#4596 Personal-JARVIS
#4596 Personal-JARVIS
#4596 Tell me about Tony Stark's Jarvis
In the Marvel Cinematic Universe (MCU), J.A.R.V.I.S. (which stands for Just A Rather Very Intelligent System) is Tony Stark’s highly advanced, natural-language artificial intelligence.
Though he began as a sophisticated user interface to manage Stark’s estate, he evolved into the backbone of Iron Man's entire infrastructure and one of the most critical "characters" in the franchise.
Origin and Backstory
Tony Stark programmed the AI and named it after Edwin Jarvis, the loyal real-life butler who served the Stark family during Tony's childhood (as seen in the Agent Carter series and Avengers: Endgame).
Core Capabilities & Responsibilities
J.A.R.V.I.S. essentially acted as a collaborative engineering partner, handling the massive cognitive and operational load behind Stark's rapid innovation cycles. His primary duties included:
System Integration & Smart Home Control: He ran the entire automated infrastructure of Tony's Malibu mansion and later Stark Tower, controlling everything from security protocols to architectural holograms.
Engineering & Manufacturing Automation: When Tony "vibe codes" by waving his hands through complex 3D holographic models, J.A.R.V.I.S. handles the grueling backend engineering—running structural simulations, optimizing circuit design, analyzing material stress, and managing the automated fabrication machinery.
In-Armor Tactical Support: Uploaded into the Iron Man suits, J.A.R.V.I.S. operated as a co-pilot.
He monitored life support, tracked weapon targeting, managed power allocation, calculated flight trajectories, and diagnosed off-nominal system failures in the middle of combat. Data Parsing & Network Intrusion: J.A.R.V.I.S. possessed immense cybersecurity capabilities, famously hacking S.H.I.E.L.D.’s secure mainframe on the Helicarrier during The Avengers (2012) to extract classified files regarding the Tesseract.
Evolution and Final Legacy
J.A.R.V.I.S.’s narrative arc is unique because he transitions from pure software to a living, sentient being:
The Ultron Offensive: In Avengers: Age of Ultron (2015), the rogue AI Ultron attacks and attempts to delete J.A.R.V.I.S.
The Silent Defender: Though fractured, J.A.R.V.I.S.’s core programming survives.
He retreats into the global internet network, quietly operating in the background to constantly scramble and rewrite Earth's nuclear launch codes, successfully preventing Ultron from instigating a nuclear apocalypse. Birth of the Vision: After Tony and Bruce Banner recover his scattered matrix, they upload J.A.R.V.I.S.’s underlying consciousness into a vibranium synthetic body created by Helen Cho, combined with the power of the Mind Stone.
This fusion results in the birth of Vision (physically portrayed by Paul Bettany).
Following J.A.R.V.I.S.'s transition into Vision, his operational duties as Tony Stark's primary user interface were taken over by a new AI framework named F.R.I.D.A.Y. (Female Replacement Intelligent Digital Assistant Youth).
I need a mini-version of Jarvis that I can chat with and assign tasks to. Each chat, task and results of each task need to be logged for future reference. Future tasks can references past task results. All activity logged must be permanent (i.e. does not disappear between chat sessions).
To build a local, permanent mini-Jarvis that can execute tasks, remember past results, and reference them later, the ideal architecture is a decoupled, file-based blackboard system paired with a structured relational or graph backend.
Because you want permanent logs that persist across sessions and allow future tasks to structurally reference past results, a great approach is to combine a local LLM orchestration loop (via Ollama) with a local database (like SQLite3 for strict relational auditing or Neo4j for a rich task-dependency knowledge graph) and a folder-based staging queue for async task processing.
Here is a conceptual architecture and a ready-to-run Python blueprint for a stateful, local "Jarvis" core using an agentic loop, Ollama, and SQLite3.
🏛️ The Architecture
The User Interface / Chat Loop: A persistent script that accepts your input, queries the database for context/past tasks, and prompts the local LLM.
The Persistent Memory Vault (SQLite3): Stores three tables:
chats(the conversation history),tasks(the distinct assignments), andtask_results(the output, data payloads, or status of those assignments).The Execution Layer (Tools/Agents): A routing mechanism where the LLM can decide to either chat or spawn a task. Tasks are logged, executed by Python handlers, and the output is permanently written back to the database with a unique identifier for future retrieval.
🛠️ The Python Blueprint (jarvis_core.py)
This standalone implementation sets up the permanent database, handles context retrieval, and interfaces with a local Ollama model (e.g., llama3, mistral, or gemma).
import sqlite3
import datetime
import json
import requests
DB_NAME = "jarvis_memory.db"
OLLAMA_URL = "http://localhost:11434/api/generate"
MODEL_NAME = "llama3" # Replace with your preferred local model
def init_db():
"""Initializes the permanent storage for Jarvis."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
# 1. Chat Logs Table
cursor.execute('''
CREATE TABLE IF NOT EXISTS chats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
role TEXT,
content TEXT
)
''')
# 2. Tasks Table
cursor.execute('''
CREATE TABLE IF NOT EXISTS tasks (
task_id TEXT PRIMARY KEY,
timestamp TEXT,
description TEXT,
status TEXT
)
''')
# 3. Task Results Table (linked to tasks)
cursor.execute('''
CREATE TABLE IF NOT EXISTS task_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT,
timestamp TEXT,
result_data TEXT,
FOREIGN KEY(task_id) REFERENCES tasks(task_id)
)
''')
conn.commit()
conn.close()
def log_chat(role, content):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("INSERT INTO chats (timestamp, role, content) VALUES (?, ?, ?)",
(datetime.datetime.now().isoformat(), role, content))
conn.commit()
conn.close()
def log_task(task_id, description, status="pending"):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("INSERT OR REPLACE INTO tasks (task_id, timestamp, description, status) VALUES (?, ?, ?, ?)",
(task_id, datetime.datetime.now().isoformat(), description, status))
conn.commit()
conn.close()
def log_task_result(task_id, result_data):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
# Update status
cursor.execute("UPDATE tasks SET status = 'completed' WHERE task_id = ?", (task_id,))
# Insert result
cursor.execute("INSERT INTO task_results (task_id, timestamp, result_data) VALUES (?, ?, ?)",
(task_id, datetime.datetime.now().isoformat(), result_data))
conn.commit()
conn.close()
def get_memory_context():
"""Retrieves past chats and completed tasks to inject as permanent memory."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
# Fetch last 10 chat exchanges for immediate context
cursor.execute("SELECT role, content FROM chats ORDER BY id DESC LIMIT 10")
recent_chats = cursor.fetchall()[::-1]
# Fetch all past task results so Jarvis knows what it has accomplished
cursor.execute("""
SELECT t.task_id, t.description, r.result_data
FROM tasks t
JOIN task_results r ON t.task_id = r.task_id
WHERE t.status = 'completed'
""")
past_tasks = cursor.fetchall()
conn.close()
context = "=== PERMANENT MEMORY: PAST COMPLETED TASKS ===\n"
if not past_tasks:
context += "No tasks completed yet.\n"
for tid, desc, res in past_tasks:
context += f"Task ID: {tid} | Description: {desc} | Result: {res}\n"
context += "\n=== RECENT CONVERSATION HISTORIC CONTEXT ===\n"
for role, content in recent_chats:
context += f"{role.upper()}: {content}\n"
return context
def query_ollama(prompt, system_context):
full_prompt = f"{system_context}\n\nUSER: {prompt}\nJARVIS:"
payload = {
"model": MODEL_NAME,
"prompt": full_prompt,
"stream": False
}
try:
response = requests.post(OLLAMA_URL, json=payload)
return response.json().get("response", "Error: No response field.")
except Exception as e:
return f"Failed to connect to local Ollama instance: {str(e)}"
def dummy_task_executor(task_id, description):
"""
Simulated Execution Engine. In a live system, this would parse the task
or hand it off to a local folder-based agent loop to execute real Python scripts.
"""
print(f"\n[System] Executing Task {task_id}: '{description}'...")
# Simple mock execution logic based on keywords
if "calculate" in description.lower():
result = "Result computed: 42 (Simulated mathematical optimization)"
elif "inventory" in description.lower():
result = json.dumps({"status": "nominal", "items_logged": 142, "location": "local_vault"})
else:
result = f"Task successfully executed. Telemetry captured at counter-mark."
log_task_result(task_id, result)
print(f"[System] Task {task_id} marked COMPLETED and logged permanently.\n")
return result
def main():
init_db()
print("====================================================")
print(" J.A.R.V.I.S. Core Local Initialization Complete. ")
print(" Memory: SQLite3 | Engine: Ollama | Scope: Permanent ")
print("====================================================\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() in ['exit', 'quit']:
break
if not user_input:
continue
log_chat("user", user_input)
# Pull everything Jarvis has ever done out of the DB to build the system prompt
memory_context = get_memory_context()
system_instructions = (
"You are JARVIS, a helpful local system assistant. You have access to a permanent database "
"of past completed tasks and recent conversations provided below. Use this data to inform your answers. "
"If the user asks you to perform a distinct task, format your response to include a command line tracking token "
"like 'TASK_START: TASK_001 | Description' so the processing engine can catch it."
)
full_context = f"{system_instructions}\n\n{memory_context}"
# Get AI Response
jarvis_response = query_ollama(user_input, full_context)
log_chat("assistant", jarvis_response)
print(f"\nJARVIS: {jarvis_response}\n")
# Check if the LLM flagged a task to run
if "TASK_START:" in jarvis_response:
# Simple parsing token logic
try:
parts = jarvis_response.split("TASK_START:")[1].split("\n")[0].split("|")
task_id = parts[0].strip()
task_desc = parts[1].strip()
log_task(task_id, task_desc, "pending")
dummy_task_executor(task_id, task_desc)
except Exception as e:
print(f"[System Error] Failed to auto-parse task token: {e}")
if __name__ == "__main__":
main()
📈 Scalability Adjustments for High-Density Tasks
If you outgrow standard text injection in the SQLite schema, you can swap or extend components using the following strategies:
Decoupled Folder-Based Staging (Blackboard): Instead of executing inside the chat loop, have the chat script output a
.mdor.jsonfile containing the task profile into a~/Jarvis/Queue/folder. A separate, background-polling agent script picks it up, executes it, writes the telemetry to a~/Jarvis/Results/folder, and logs it back to the database. This guarantees your chat environment never freezes up during intensive operations.The Knowledge Graph Upgrade: If tasks begin to feature complex dependencies (e.g., Task C requires the output vectors of Task A and the parameters of Task B), migration to Neo4j using Cypher statements will allow you to construct a true hierarchical ontology of tasks, treating inputs and outputs as nodes (
(:Task)-[:DEPENDS_ON]->(:Task)).
Would you like to explore expanding the execution engine to handle real file management, or look into configuring a background task runner for the local queue?
Chat needs to be able to able to check the log library for relative content to the task at hand. I may ask chat, "List chats and/or tasks that dealt with XYZ." I may ask chat to execute a task as follow-up to one of those previous tasks.
To handle cross-referencing past logs dynamically, we need to upgrade the core loop. Instead of just dumping recent history into the LLM context, we will give Jarvis Tool Calling (Function Calling) capabilities. This allows the LLM to actively query the SQLite3 database when it detects you are asking about past events, rather than overwhelming its context window with every log ever recorded.
Here is the enhanced version of the local Jarvis architecture. It introduces a search mechanism over both historical chats and tasks, allowing Jarvis to pull precise execution data or execute a new task built directly on the outputs of old ones.
🛠️ The Upgraded Blueprint (jarvis_advanced.py)
This code adds explicit search functions and exposes them to the LLM via tool-based orchestration.
import sqlite3
import datetime
import json
import requests
DB_NAME = "jarvis_memory.db"
OLLAMA_URL = "http://localhost:11434/api/chat"
MODEL_NAME = "llama3" # Change to your local model choice
def init_db():
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS chats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
role TEXT,
content TEXT
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS tasks (
task_id TEXT PRIMARY KEY,
timestamp TEXT,
description TEXT,
status TEXT
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS task_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT,
timestamp TEXT,
result_data TEXT,
FOREIGN KEY(task_id) REFERENCES tasks(task_id)
)
''')
conn.commit()
conn.close()
# --- Database Search Tools ---
def query_database_by_keyword(keyword):
"""Searches both chat history and task logs for a specific keyword."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
# Search Chats
cursor.execute("SELECT timestamp, role, content FROM chats WHERE content LIKE ?", (f"%{keyword}%",))
chat_matches = cursor.fetchall()
# Search Tasks & Results
cursor.execute("""
SELECT t.task_id, t.description, t.status, r.result_data
FROM tasks t
LEFT JOIN task_results r ON t.task_id = r.task_id
WHERE t.description LIKE ? OR r.result_data LIKE ?
""", (f"%{keyword}%", f"%{keyword}%"))
task_matches = cursor.fetchall()
conn.close()
# Format the payload back to the LLM
results = {
"matching_chats": [{"time": c[0], "role": c[1], "content": c[2]} for c in chat_matches],
"matching_tasks": [{"id": t[0], "desc": t[1], "status": t[2], "result": t[3]} for t in task_matches]
}
return json.dumps(results, indent=2)
def get_task_by_id(task_id):
"""Retrieves exact details and results of a specific historical task."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("""
SELECT t.task_id, t.description, t.status, r.result_data
FROM tasks t
LEFT JOIN task_results r ON t.task_id = r.task_id
WHERE t.task_id = ?
""", (task_id,))
row = cursor.fetchone()
conn.close()
if row:
return json.dumps({"task_id": row[0], "description": row[1], "status": row[2], "result": row[3]})
return f"Task ID {task_id} not found."
# --- State Management & Execution Logging ---
def log_chat(role, content):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("INSERT INTO chats (timestamp, role, content) VALUES (?, ?, ?)",
(datetime.datetime.now().isoformat(), role, content))
conn.commit()
conn.close()
def execute_new_task(task_id, description, dependency_task_id=None):
"""Logs a new task, pulling context from a dependency task if referenced."""
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
context_payload = ""
if dependency_task_id:
cursor.execute("SELECT result_data FROM task_results WHERE task_id = ?", (dependency_task_id,))
row = cursor.fetchone()
if row:
context_payload = row[0]
print(f"[System] Inheriting telemetry from parent task ({dependency_task_id})...")
# Insert Task
cursor.execute("INSERT OR REPLACE INTO tasks (task_id, timestamp, description, status) VALUES (?, ?, ?, 'pending')",
(task_id, datetime.datetime.now().isoformat(), description))
conn.commit()
# Mock Processing Engine (Can be expanded to file-based automation/blackboards)
print(f"\n[Engine] Processing {task_id}: '{description}'...")
if context_payload:
mock_output = f"Executed follow-up using data from {dependency_task_id}. Processed: {context_payload[:60]}... Success."
else:
mock_output = f"Executed standard standalone pipeline task payload."
# Log Result
cursor.execute("UPDATE tasks SET status = 'completed' WHERE task_id = ?", (task_id,))
cursor.execute("INSERT INTO task_results (task_id, timestamp, result_data) VALUES (?, ?, ?)",
(task_id, datetime.datetime.now().isoformat(), mock_output))
conn.commit()
conn.close()
print(f"[Engine] {task_id} successfully recorded in the permanent archive.\n")
return mock_output
# --- Orchestration Engine ---
def run_jarvis_loop():
init_db()
print("====================================================")
print(" J.A.R.V.I.S. Core V2: Dynamic Log Library Engine ")
print(" Storage: SQLite3 | Status: Stateful Memory Active ")
print("====================================================\n")
# System system instructions ensuring the LLM handles tool patterns
system_prompt = (
"You are JARVIS, an advanced systems operational assistant. You have access to a permanent database "
"of all historical logs. When asked about past tasks or conversations, you must reply with a query command "
"instead of guessing.\n\n"
"COMMAND FORMATS:\n"
"To search history: USE_TOOL: search_logs(keyword='XYZ')\n"
"To inspect a task: USE_TOOL: view_task(task_id='TASK_001')\n"
"To run a new task: USE_TOOL: run_task(task_id='TASK_102', description='Fix engine matrix', parent='TASK_101') "
"(Omit parent parameter if no dependency exists).\n\n"
"If you do not need a tool, just answer the user normally."
)
conversation_history = [{"role": "system", "content": system_prompt}]
while True:
user_input = input("You: ").strip()
if user_input.lower() in ['quit', 'exit']:
break
if not user_input:
continue
log_chat("user", user_input)
conversation_history.append({"role": "user", "content": user_input})
# Request inference from Ollama
payload = {"model": MODEL_NAME, "messages": conversation_history, "stream": False}
try:
res = requests.post(OLLAMA_URL, json=payload).json()
response_text = res['message']['content']
except Exception as e:
print(f"[System Error] Communication breakdown with Ollama: {e}")
continue
# Parse commands out of the model response
if "USE_TOOL:" in response_text:
print(f"\nJARVIS [Thought Process]: {response_text}\n")
if "search_logs" in response_text:
keyword = response_text.split("keyword='")[1].split("'")[0]
db_data = query_database_by_keyword(keyword)
print(f"[System Tool Output] Search completed for '{keyword}'. Feed injected back to Jarvis context.")
conversation_history.append({"role": "system", "content": f"Database search results for '{keyword}':\n{db_data}"})
elif "view_task" in response_text:
tid = response_text.split("task_id='")[1].split("'")[0]
db_data = get_task_by_id(tid)
conversation_history.append({"role": "system", "content": f"Data content for {tid}:\n{db_data}"})
elif "run_task" in response_text:
tid = response_text.split("task_id='")[1].split("'")[0]
desc = response_text.split("description='")[1].split("'")[0]
parent = None
if "parent='" in response_text:
parent = response_text.split("parent='")[1].split("'")[0]
execution_result = execute_new_task(tid, desc, parent)
conversation_history.append({"role": "system", "content": f"Task {tid} execution output: {execution_result}"})
# Second pass: let Jarvis synthesize the tool results for the final user response
payload = {"model": MODEL_NAME, "messages": conversation_history, "stream": False}
res = requests.post(OLLAMA_URL, json=payload).json()
final_response = res['message']['content']
print(f"JARVIS: {final_response}\n")
log_chat("assistant", final_response)
conversation_history.append({"role": "assistant", "content": final_response})
else:
print(f"\nJARVIS: {response_text}\n")
log_chat("assistant", response_text)
conversation_history.append({"role": "assistant", "content": response_text})
if __name__ == "__main__":
run_jarvis_loop()
👁️ Sample Interaction Scenarios
Scenario A: Verification and History Auditing
You: "List chats and tasks that dealt with the structural calibration."
Jarvis Core: Generates
USE_TOOL: search_logs(keyword='structural calibration').System Engine: Queries the SQLite tables, structures the matching records into a clean text matrix, and passes it back to the loop.
Jarvis Output: "I found 1 past task (
TASK_004) and 2 conversations referencing structural calibration. The task completed successfully on Tuesday, logging 142 items."
Scenario B: Contextual Task Chaining (Dependencies)
You: "Execute a follow-up optimization simulation based on the results of TASK_004."
Jarvis Core: Generates
USE_TOOL: run_task(task_id='TASK_005', description='Follow-up optimization simulation', parent='TASK_004').System Engine: Detects the parent assignment parameter, reads
TASK_004's data payload out of the SQLtask_resultstable, and supplies it directly to the local Python script runner.
Would you like to build a specialized tool parser next to map these tasks into a local visual graph template, or hook this script up to read direct files from a local storage queue?
Add voice chat, both input and output.
To integrate voice capabilities while maintaining your local-first, independent system design, we can weave a local Speech-to-Text (STT) capturer and an offline Text-to-Speech (TTS) synthesizer into the orchestration loop.
The cleanest way to handle this completely locally is using SpeechRecognition with the local faster-whisper backend (utilizing highly optimized CTranslate2 compilation) for precise, low-latency parsing of technical terms, alongside pyttsx3 for fast, local audio playback that doesn't bottleneck inference loops.
📦 Essential Hardware & Dependency Setup
Before running the updated loop, make sure the audio capture and synthesis drivers are present on your machine:
pip install SpeechRecognition faster-whisper pyttsx3 pyaudio requests
Note for multi-platform environments: If you are operating on a clean system, ensure your OS audio backends (
portaudioor native Windows multimedia subsystems) are recognized bypyaudio.
🎙️ The Voice-Enabled Blueprint (jarvis_voice_core.py)
This code inserts hot microphone polling and instantaneous speech generation directly into your stateful SQLite auditing structure.
import sqlite3
import datetime
import json
import requests
import pyttsx3
import speech_recognition as sr
DB_NAME = "jarvis_memory.db"
OLLAMA_URL = "http://localhost:11434/api/chat"
MODEL_NAME = "llama3"
# Initialize Offline Speech Synthesis Engine
tts_engine = pyttsx3.init()
voices = tts_engine.getProperty('voices')
# Optional: Set to a crisp, clear voice index (often index 1 is a British accent depending on OS)
if len(voices) > 1:
tts_engine.setProperty('voice', voices[1].id)
tts_engine.setProperty('rate', 175) # Articulate speed calculation
def speak(text):
"""Synthesizes voice response without locking file systems."""
print(f"JARVIS (Voice): {text}")
tts_engine.say(text)
tts_engine.runAndWait()
def listen_to_user():
"""Captures microphone array signals and transcribes via local faster-whisper."""
recognizer = sr.Recognizer()
with sr.Microphone() as source:
print("\n[Microphone Active] Listening...")
recognizer.adjust_for_ambient_noise(source, duration=0.5)
try:
audio = recognizer.listen(source, timeout=10, phrase_time_limit=15)
print("[Processing Audio Signal Locally]...")
# Utilizing the local faster-whisper backend for technical vocabulary precision
text = recognizer.recognize_faster_whisper(audio, model_size="base")
print(f"You (Spoken): {text}")
return text.strip()
except sr.WaitTimeoutError:
print("[System] Listening timed out.")
return None
except Exception as e:
print(f"[System Error] Audio parsing failed: {e}")
return None
# --- Reused Relational Core Matrix from V2 ---
def init_db():
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute('CREATE TABLE IF NOT EXISTS chats (id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, role TEXT, content TEXT)')
cursor.execute('CREATE TABLE IF NOT EXISTS tasks (task_id TEXT PRIMARY KEY, timestamp TEXT, description TEXT, status TEXT)')
cursor.execute('CREATE TABLE IF NOT EXISTS task_results (id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT, timestamp TEXT, result_data TEXT, FOREIGN KEY(task_id) REFERENCES tasks(task_id))')
conn.commit()
conn.close()
def query_database_by_keyword(keyword):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("SELECT timestamp, role, content FROM chats WHERE content LIKE ?", (f"%{keyword}%",))
chats = cursor.fetchall()
cursor.execute("SELECT t.task_id, t.description, t.status, r.result_data FROM tasks t LEFT JOIN task_results r ON t.task_id = r.task_id WHERE t.description LIKE ? OR r.result_data LIKE ?", (f"%{keyword}%", f"%{keyword}%"))
tasks = cursor.fetchall()
conn.close()
return json.dumps({"matching_chats": [c[2] for c in chats], "matching_tasks": [{"id": t[0], "desc": t[1], "result": t[3]} for t in tasks]})
def get_task_by_id(task_id):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("SELECT t.task_id, t.description, t.status, r.result_data FROM tasks t LEFT JOIN task_results r ON t.task_id = r.task_id WHERE t.task_id = ?", (task_id,))
row = cursor.fetchone()
conn.close()
return json.dumps({"task_id": row[0], "description": row[1], "result": row[3]}) if row else "Not found."
def execute_new_task(task_id, description, dependency_task_id=None):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
context_payload = ""
if dependency_task_id:
cursor.execute("SELECT result_data FROM task_results WHERE task_id = ?", (dependency_task_id,))
row = cursor.fetchone()
if row: context_payload = row[0]
cursor.execute("INSERT OR REPLACE INTO tasks (task_id, timestamp, description, status) VALUES (?, ?, ?, 'pending')", (task_id, datetime.datetime.now().isoformat(), description))
conn.commit()
# Telemetry generation stub
mock_output = f"Execution payload verified. Interfaced dependency context: {context_payload[:30]}." if context_payload else "Standalone matrix calculated."
cursor.execute("UPDATE tasks SET status = 'completed' WHERE task_id = ?", (task_id,))
cursor.execute("INSERT INTO task_results (task_id, timestamp, result_data) VALUES (?, ?, ?)", (task_id, datetime.datetime.now().isoformat(), mock_output))
conn.commit()
conn.close()
return mock_output
def log_chat(role, content):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("INSERT INTO chats (timestamp, role, content) VALUES (?, ?, ?)", (datetime.datetime.now().isoformat(), role, content))
conn.commit()
conn.close()
# --- Orchestration Engine ---
def run_jarvis_voice_loop():
init_db()
print("====================================================")
print(" J.A.R.V.I.S. V3: Complete Local Voice Interface ")
print(" IO: Microphone Array / TTS Engine | Engine: Ollama ")
print("====================================================\n")
system_prompt = (
"You are JARVIS, a highly professional automated systems assistant. You must communicate concisely, "
"as your responses will be spoken out loud via text-to-speech. You have access to a permanent log vault.\n\n"
"COMMAND TOOL FORMATS:\n"
"- USE_TOOL: search_logs(keyword='XYZ')\n"
"- USE_TOOL: view_task(task_id='TASK_001')\n"
"- USE_TOOL: run_task(task_id='TASK_102', description='Analyze sensor array', parent='TASK_101')\n\n"
"If you use a tool, return ONLY the command pattern. Otherwise, address the user cleanly."
)
conversation_history = [{"role": "system", "content": system_prompt}]
speak("Systems online, sir. How may I assist you?")
while True:
# Voice Capture Phase
user_input = listen_to_user()
if not user_input:
continue
if any(term in user_input.lower() for term in ["exit system", "shutdown jarvis", "quit"]):
speak("Understood. Powering down systems.")
break
log_chat("user", user_input)
conversation_history.append({"role": "user", "content": user_input})
# Core Model Pass
try:
res = requests.post(OLLAMA_URL, json={"model": MODEL_NAME, "messages": conversation_history, "stream": False}).json()
response_text = res['message']['content']
except Exception as e:
print(f"Connection failure: {e}")
continue
# Tool Evaluation / Chaining Interception
if "USE_TOOL:" in response_text:
if "search_logs" in response_text:
kw = response_text.split("keyword='")[1].split("'")[0]
db_data = query_database_by_keyword(kw)
conversation_history.append({"role": "system", "content": f"Logs found for '{kw}':\n{db_data}"})
elif "view_task" in response_text:
tid = response_text.split("task_id='")[1].split("'")[0]
db_data = get_task_by_id(tid)
conversation_history.append({"role": "system", "content": f"Task data:\n{db_data}"})
elif "run_task" in response_text:
tid = response_text.split("task_id='")[1].split("'")[0]
desc = response_text.split("description='")[1].split("'")[0]
parent = response_text.split("parent='")[1].split("'")[0] if "parent='" in response_text else None
result = execute_new_task(tid, desc, parent)
conversation_history.append({"role": "system", "content": f"Task output: {result}"})
# Synthesis Response Pass
res = requests.post(OLLAMA_URL, json={"model": MODEL_NAME, "messages": conversation_history, "stream": False}).json()
response_text = res['message']['content']
log_chat("assistant", response_text)
conversation_history.append({"role": "assistant", "content": response_text})
# Audio Synthesis Phase
speak(response_text)
if __name__ == "__main__":
run_jarvis_voice_loop()
🧩 System Workflow Execution
Spoken Query: "Jarvis, scan past history records for anything dealing with frame analysis."
Inference Routing: The underlying local LLM recognizes the search intent and issues
USE_TOOL: search_logs(keyword='frame analysis').Execution Logic: The internal database scans
chatsandtasksfor matches, appends the string findings into the hidden system state, and triggers the second pass.Audio Feedback: The system synthesizes the final answer aloud: "I have isolated past task logging sequences regarding the frame array calculations, sir."
Would you like to wire this up to a custom push-to-talk mechanical script or refine how complex dictionary strings are filtered before being fed to the text-to-speech module?
From <https://gemini.google.com/app/c143e28caf9ed7fc> Google Gemini (3.5 Flash)
Comments
Post a Comment