# Project export: Vision Cortex
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: UC Berkeley AI Hackathon 2026
- Tagline: Most AI assistants only know what you type into them. But the most important context in your life isn't typed; it happens during conversations.Your life has no search bar. We're building one.
- Devpost: https://devpost.com/software/cortex-9rn8tg
- GitHub: https://github.com/VamikaSinghal/second-brain
- Video: https://www.youtube.com/embed/9WJX5YJ0xMA?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 2 GitHub contributor(s) — vchsi (3 commits), Vamika Singhal (1 commits)
## Devpost submission (written by the team)
### Inspiration
Every AI tool you use starts from zero. You explain your project to ChatGPT, then re-explain it to Claude, then again to whatever agent you spin up next week. Meanwhile your actual context, the decisions buried in Slack threads, the half-finished docs in Notion, the conversation you had walking to lunch, lives scattered across a dozen apps that don't talk to each other. And it's about to get worse: wearables like Ray-Ban Meta glasses and Omi are generating a continuous stream of personal context that nothing captures in a usable way. Two things made now the right moment. MCP gave us a real interoperability substrate: a way for any AI to read from a shared memory layer instead of every vendor building a walled garden. And the wearables ecosystem finally opened up enough, through Meta's Device Access Toolkit and Omi's ambient audio, that we could actually pull from it. So we built the missing piece: a persistent memory layer that sits underneath all of it.
### What it does
Cortex is a persistent, cross-agent memory and context layer. It ingests from your wearables, including Ray-Ban Meta glasses and Omi, and your productivity tools, including Claude, ChatGPT, OpenClaw, Claude Code, Codex, Slack, email, Notion, and Google Docs. It processes everything into a unified knowledge graph and exposes that context through an MCP server, so any AI you use can read your full personal context without you re-explaining yourself. Deepgram powers the real-time voice layer for the wearable experience. We used Deepgram to transcribe audio input from the Meta glasses, turning live conversations and ambient speech into structured text that Cortex could filter, process, and store as memory. We also used Deepgram TTS for the output path, so Cortex could respond back through the Meta glasses as spoken audio instead of only returning text. The Token Company API sits early in the ingestion pipeline as a cleanup and normalization layer. Every input source has noise: Slack has messy threads, docs have unfinished fragments, agent sessions have repeated prompts, and natural conversations from the Meta glasses have the most fluff of all. The Token Company API helped clean those raw inputs before memory extraction, reducing filler, repetition, and irrelevant conversational noise so the graph receives the actual signal instead of a transcript landfill. The storage is deliberately dual-layer: a local markdown folder that doubles as an Obsidian vault and a git repo, alongside Redis vector embeddings for fast semantic retrieval. The local layer keeps the system human-readable, user-owned, and diffable, while Redis makes the memory layer fast enough for real agent use. The MCP server is the part that matters most: it turns a personal knowledge base into something every agent in your stack can actually use.
### How we built it
The architecture runs in four phases: ingest → clean → process → store → expose. The processing pipeline is the heart of it, with seven stages: normalize each source into a common text envelope, clean the input using The Token Company API, segment and filter, extract into the dual store, resolve entities against the existing graph, reconcile against what we already know, then persist and index. The core principle we kept coming back to: aggregation has to happen at write time, with the existing graph in context. If you defer entity resolution to query time, the same person or project fragments into five different nodes and the whole graph rots. For the wearable streams, we used a three-tier model: stream → working set → graph. Raw perceptual data never enters the knowledge graph directly. Audio from the Meta glasses first goes through Deepgram speech-to-text, which gives us the live transcript stream. That transcript is then cleaned through The Token Company API to strip out the natural fluff of spoken conversation, including filler words, repeated phrases, rambling, and low-value fragments. After that, Anthropic Haiku powers the Meta glasses OpenClaw agent, helping decide what is relevant enough to keep, summarize, or respond to. That cleaned transcript becomes one of the highest-value salience signals because spoken phrases like “this is important,” “remind me,” “we decided,” or references to “this” and “that” can tell the system which visual or conversational moments are worth keeping. On the output side, Deepgram TTS turns Cortex responses back into speech, making the wearable interaction loop feel natural instead of forcing the user to look at a screen. Continuous video and audio get cascade-filtered with a simple rule: keep what's surprising, drop what's predictable. Audio deixis from the Deepgram transcript acts as the cheapest high-value signal for deciding whether a visual frame or moment should be promoted from raw stream into working memory. The Token Company API improves that signal by cleaning the transcript before it reaches the extraction layer, so Haiku and the memory pipeline are reasoning over meaning instead of noise. Only after filtering and consolidation does anything become structured memory. Hackathon pragmatics shaped a lot of the build: Anthropic Haiku as the lightweight model for the Meta glasses OpenClaw agent and relevance gate, The Token Company API for cleaning noisy raw input across sources, Deepgram for live speech input and spoken output, Voyage AI for embeddings, Redis for fast vector retrieval, and a vertical slice end-to-end before widening to more sources.
### Challenges we ran into
Entity fragmentation was the big one, and the reason write-time aggregation became non-negotiable. Continuous sensor data was the second: you cannot dump a video feed or raw audio stream into a knowledge graph, so we had to design the filtering and consolidation bridges carefully, balancing high-recall early gates against high-precision late consolidation. The wearable voice loop also had its own constraints. Real-time transcription needs to be fast enough to feel ambient, accurate enough to preserve meaning, and structured enough to become useful memory. Deepgram helped us bridge that gap by giving us low-latency speech-to-text for input and TTS for audio output through the glasses. The next problem was input quality. Human conversation is messy. People ramble, repeat themselves, trail off, use filler words, point at things and say “that,” and change topics mid-sentence. That is fine for humans, but brutal for a memory graph. The Token Company API helped us clean that input before extraction so Cortex could preserve the decision, commitment, fact, or context without storing every bit of conversational junk around it. Hardware availability for the demo was a real constraint, which pushed us toward pre-loading exactly the data our demo story needed rather than ingesting everything live. And underneath it all, the 24-hour clock forced us to be honest about what was a slice and what was scope creep. We also sat with the harder, non-technical problems: the trust paradox of a startup asking to hold continuous personal data, and whether this is a product or a feature.
### Accomplishments we're proud of
A working end-to-end vertical slice: real data in one side, structured memory out the other, queryable by an external agent through the MCP server. We also built a real wearable interaction loop: Deepgram transcribes audio from the Meta glasses into usable context, The Token Company API cleans the noisy transcript, Anthropic Haiku powers the Meta glasses OpenClaw agent’s reasoning layer, Cortex processes and retrieves relevant memory, and Deepgram TTS can speak the response back through the glasses. We're proud of getting the dual store to actually behave as one coherent layer. We're also proud of the architectural discipline we held to: separating reference knowledge, the encyclopedia layer, from active state, including commitments, open questions, and deadlines. That distinction is what makes Cortex useful for day-to-day task assistance and not just a search index over your life.
### What we learned
MCP is the real differentiator. The storage mechanism is replaceable; the exposure layer is the value. Voice is the natural interface for wearable memory. Deepgram gave us both sides of that loop: speech-to-text for capturing context and TTS for responding back through the glasses. Cleaning input is not optional. The Token Company API helped turn noisy agent sessions, app data, and especially messy Meta glasses conversations into cleaner memory candidates. Small models are enough when the pipeline is designed well. Anthropic Haiku gave us a fast, lightweight reasoning layer for the Meta glasses OpenClaw agent and relevance filtering. Aggregate at write time, never at query time. Entity resolution with graph context during ingestion is what prevents fragmentation. Raw perceptual data never enters the graph. Streams and structured memory have to stay architecturally separate. Reference knowledge vs. active state is the distinction that unlocks actual task help. Cheap gates buy you a lot. A Haiku relevance filter and a single combined extract-and-resolve call gave us most of the quality at a fraction of the cost and latency.
### What's next
for Cortex Widen ingestion well beyond the demo slice, then harden the wearable integration as Meta's Wearables DAT platform reaches broader availability later in 2026. We also want to make the voice loop more proactive: Deepgram listens and transcribes when useful, The Token Company API cleans the transcript, Haiku decides whether the moment matters, Cortex updates memory, and the glasses only speak back when the response is actually valuable. Beyond features, the real work is the trust model: credibly answering why someone should let Cortex hold continuous personal context, and finding a monetization path that doesn't undermine the "you own your data" promise the local markdown layer is built on.
## README (from the GitHub repository)
# Cortex - Universal Context Layer
> *One layer. Every AI knows you.*
Built at UC Berkeley AI Hackathon 2026.
---
## The Problem
Every AI you talk to starts from zero. Every app you use forgets what the others know. Your second brain is already there — it's just scattered across 8 silos.
## What Cortex Does
Cortex is a **persistent context layer** that:
1. **Captures** everything — select any text in any app, press `Cmd+Shift+V`, it's saved
2. **Processes** with Claude — extracts entities, decisions, insights, open questions
3. **Stores** in a GitHub repo (version-controlled markdown) + Redis vector search
4. **Exposes as an MCP server** — so Claude, and any AI, can call `get_context(query)` and instantly know your full history
## Demo
You walk up to the judges and say:
*"Three weeks ago I was in a meeting discussing a startup idea. Watch what happens when I ask Claude about it now."*
Claude — through Cortex's MCP — pulls the transcript, the Slack thread, the Notion note, and a ChatGPT conversation where you refined the model. It surfaces them as one coherent answer with timestamps and sources.
*"And it's been doing this passively — I didn't tag anything, I didn't organize anything. It just knew."*
---
## Architecture
```
CAPTURE LAYER
Global hotkey (Cmd+Shift+V) — any app, any text
Claude Desktop (auto-saves via MCP + Project instructions)
Omi wearable — ambient audio transcripts
Meta Ray-Bans — visual context via VisionClaw
↓
PROCESSING LAYER — Claude API
Extract: entities, decisions, insights, open questions
Tag: source, timestamp, topic, people
Link: connect related ideas across sources
↓
STORAGE LAYER
GitHub repo — markdown files, version-controlled, browsable
Redis Stack — vector embeddings for semantic search
SQLite — metadata index
↓
QUERY LAYER — MCP Server
get_context(query) → semantic search across all memory
get_recent(hours) → what happened lately
get_about_person(name) → everything about someone
get_open_questions() → unresolved threads → GitHub Issues
```
---
## Project Structure
```
cortex/
├── ingest.py # Claude extraction pipeline (raw text → structured context)
├── github_store.py # Push notes to GitHub repo via API
├── redis_store.py # Voyage AI embeddings + Redis vector search
├── mcp_server.py # MCP server — exposes Cortex to any AI
├── capture.py # Global hotkey menu bar app (Cmd+Shift+V)
├── webhook.py # FastAPI webhook for ChatGPT/Gemini live capture
├── requirements.txt
├── .env.example
└── SETUP.md # Full setup guide
```
---
## Quickstart
```bash
git clone https://github.com/VamikaSinghal/cortex-ai
cd cortex-ai
python3.12 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
# Fill in: ANTHROPIC_API_KEY, GITHUB_TOKEN, CORTEX_REPO, VOYAGE_API_KEY
# Start Redis
docker run -d -p 6379:6379 --name cortex-redis redis/redis-stack
# Init GitHub repo structure + Redis index
python github_store.py
python redis_store.py
# Start the MCP server (connect to Claude Desktop via claude_desktop_config.json)
python mcp_server.py
# Start the global capture tool (menu bar app)
python capture.py
```
See [SETUP.md](SETUP.md) for the full step-by-step guide including Claude Desktop config and Claude Project setup.
---
## Tech Stack
| Layer | Tool |
|---|---|
| AI / reasoning | Claude (Anthropic API) |
| Embeddings | Voyage AI `voyage-3` |
| Vector search | Redis Stack |
| Note storage | GitHub repo (markdown) |
| MCP server | Python `mcp` SDK |
| Menu bar app | `rumps` + `pynput` |
| Wearable audio | Omi |
| Wearable vision | Meta Ray-Bans + VisionClaw |
| Observability | Arize Phoenix |
---
## Prize Tracks
- 🏆 **Ddoski's Toolbox** — ultimate productivity/knowledge tool
- 🤖 **Anthropic** — Claude as reasoning engine, MCP server is Claude-native
- 🔴 **Redis** — vector search is the core memory retrieval mechanism
- 📊 **Arize** — telemetry on every context retrieval call
---
## Team
Built by Vamika Singhal at UC Berkeley AI Hackathon 2026.
## Detected evidence (automated analysis)
Indexed codebase: 9 recognized source files, 78 KB.
- Anthropic (technology) — detected in the code
- FastAPI (technology) — detected in the code
- Python (language) — detected in the code
## Codebase structure (from repository index)
### Files (14 of 14)
```
.env.example
.gitignore
capture.py
claude_desktop_config.json
CORTEX_PROJECT_SYSTEM_PROMPT.txt
github_store.py
ingest.py
instrumentation.py
mcp_server.py
README.md
redis_store.py
requirements.txt
SETUP.md
ui.py
```
### Dependencies
- requirements.txt: anthropic@>=0.40.0, arize-otel@>=0.13.0, fastapi@>=0.100.0, mcp@>=1.0.0, openinference-instrumentation-anthropic@>=1.0.0, pynput@>=1.7.0, pyperclip@>=1.8.0, python-dotenv@>=1.0.0, redis[hiredis]@>=5.0.0, requests@>=2.31.0, rumps@>=0.4.0, uvicorn@>=0.20.0, voyageai@>=0.2.0
### Recent commits (newest first)
- Fix heading formatting in README.md
- initial commit
- Architecture v2: richer schema, query planner, metadata indexing
- Add Streamlit chat UI
- Revert "First two segments of the pipeline; untested as of now"
- Revert "Update team members in README"
- First two segments of the pipeline; untested as of now
- Update team members in README
- docs: add README
- init: Cortex capture pipeline
## Key source files (fetched from GitHub, selected and truncated for size)
### SETUP.md
```markdown
# Cortex — Setup Guide
## 1. Create the GitHub repo
```bash
# Go to github.com/new → create "cortex-brain" (private or public)
# Then generate a Personal Access Token:
# github.com/settings/tokens → New token → repo scope
```
## 2. Install dependencies
```bash
cd cortex/
pip install -r requirements.txt
```
## 3. Set environment variables
```bash
cp .env.example .env
# Edit .env with your actual keys
source .env # or use direnv / python-dotenv
```
## 4. Start Redis
```bash
docker run -d -p 6379:6379 --name cortex-redis redis/redis-stack
```
## 5. Init repo structure + Redis index
```bash
python github_store.py # creates folders in your GitHub repo
python redis_store.py # creates the Redis vector search index
```
## 6. Connect to Claude Desktop
Edit `~/Library/Application Support/Claude/claude_desktop_config.json`:
```json
{
"mcpServers": {
"cortex": {
"command": "python",
"args": ["/absolute/path/to/cortex/mcp_server.py"],
"env": {
"ANTHROPIC_API_KEY": "...",
"GITHUB_TOKEN": "...",
"CORTEX_REPO": "vamikasinghal/cortex-brain",
"OPENAI_API_KEY": "...",
"REDIS_URL": "redis://localhost:6379"
}
}
}
}
```
Restart Claude Desktop. You should see "cortex" in the MCP tools list.
## 7. Set up Claude Project for auto-capture
1. Go to claude.ai → Projects → New Project → "Cortex"
2. Paste the contents of `CORTEX_PROJECT_SYSTEM_PROMPT.txt` as the project instructions
3. Connect the Cortex MCP server to this project
Now **every Claude conversation in this project automatically saves to your second brain.**
## 8. Test it
```
You: I just decided to use GitHub for Cortex instead of Obsidian.
Claude: [saves automatically]
Claude: ✅ Saved to Cortex — 1 decision, 0 questions, 1 summary → github.com/vamikasinghal/cortex-brain
```
---
## Ingesting other sources
### ChatGPT / Gemini (batch export)
```bash
# Download export from chatgpt.com/settings or Google Takeout
# Then run:
python ingest_batch.py --file conversations.json --source chatgpt
```
*(ingest_batch.py — build this in Phase 1)*
### Slack (live, via existing MCP)
Slack MCP is already connected — add a `ingest_slack_channel(channel_id)` tool to mcp_server.py that calls the Slack MCP and feeds results through the pipeline.
### Omi (live webhook)
Configure Omi to POST transcripts to a webhook endpoint. Run a simple FastAPI server alongside mcp_server.py:
```bash
# In mcp_server.py or a separate webhook.py
@app.post("/omi-webhook")
async def omi_webhook(body: dict):
transcript = body.get("transcript", "")
extracted = extract_context(transcript, source="omi")
save_extracted_context(extracted, raw_text=transcript)
```
### Notion (via existing MCP)
Use the Notion MCP to read pages, pipe text through `extract_context()`, push to GitHub.
```
### requirements.txt
```
anthropic>=0.40.0
mcp>=1.0.0
redis[hiredis]>=5.0.0
voyageai>=0.2.0
requests>=2.31.0
python-dotenv>=1.0.0
rumps>=0.4.0
pynput>=1.7.0
pyperclip>=1.8.0
uvicorn>=0.20.0
fastapi>=0.100.0
arize-otel>=0.13.0
openinference-instrumentation-anthropic>=1.0.0
```
### instrumentation.py
```python
"""
cortex/instrumentation.py
--------------------------
Centralized Arize AX tracing setup.
Import and call setup_tracing() ONCE at the top of any entry point
(mcp_server.py, ui.py, capture.py) BEFORE any Anthropic/Voyage clients
are created. Auto-instrumentation then traces every Claude API call.
"""
import os
from opentelemetry import trace
_tracer = None
def setup_tracing(project_name: str = "cortex") -> bool:
"""
Initialize Arize AX tracing. Returns True if successful.
Safe to call multiple times — only initializes once.
"""
global _tracer
if _tracer is not None:
return True
space_id = os.environ.get("ARIZE_SPACE_ID", "")
api_key = os.environ.get("ARIZE_API_KEY", "")
if not space_id or not api_key:
print("⚠️ Arize tracing disabled — ARIZE_SPACE_ID or ARIZE_API_KEY not set")
return False
try:
from arize.otel import register
from openinference.instrumentation.anthropic import AnthropicInstrumentor
tracer_provider = register(
space_id=space_id,
api_key=api_key,
project_name=project_name,
)
# Auto-instrument all Anthropic (Claude) API calls
AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)
_tracer = trace.get_tracer(project_name)
print(f"✅ Arize tracing enabled → project '{project_name}'")
return True
except Exception as e:
print(f"⚠️ Arize tracing setup failed: {e}")
return False
def get_tracer():
"""Get the OpenTelemetry tracer. Returns a no-op tracer if not initialized."""
return trace.get_tracer("cortex")
```
### capture.py
```python
"""
cortex/capture.py
-----------------
Global hotkey capture tool. Lives in your Mac menu bar.
Press Cmd+Shift+C anywhere to save selected text to Cortex.
How it works:
1. You select text in any app (ChatGPT, Slack, browser, Notes, anywhere)
2. Press Cmd+Shift+C
3. It copies the selection to clipboard (simulates Cmd+C)
4. Grabs the clipboard text
5. Runs it through the Cortex pipeline (extract → GitHub → Redis)
6. Shows a macOS notification with what was captured
Run: python capture.py
"""
import os
import sys
import time
import threading
import subprocess
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()
import rumps
import pyperclip
from pynput import keyboard
from ingest import extract_context, format_extraction_summary
from github_store import save_extracted_context
try:
from redis_store import embed_and_store
REDIS_AVAILABLE = True
except Exception:
REDIS_AVAILABLE = False
# ── Hotkey config ─────────────────────────────────────────────────────────────
HOTKEY = {keyboard.Key.cmd, keyboard.Key.shift, keyboard.KeyCode.from_char('v')}
current_keys = set()
hotkey_triggered = False
# ── Notification ──────────────────────────────────────────────────────────────
def notify(title: str, message: str):
"""Show a macOS notification."""
subprocess.run([
"osascript", "-e",
f'display notification "{message}" with title "{title}"'
], capture_output=True)
# ── Core capture function ─────────────────────────────────────────────────────
def capture_and_save(app_instance=None):
"""
Grab selected text and save to Cortex.
Called when hotkey is triggered.
"""
# Update menu bar icon to show we're working
if app_instance:
app_instance.title = "🧠 ..."
try:
# Just read whatever is currently in the clipboard
# Flow: user selects text → Cmd+C → then Cmd+Shift+V to save to Cortex
selected_text = pyperclip.paste()
if not selected_text or not selected_text.strip():
notify("Cortex", "⚠️ Clipboard is empty — copy something first (Cmd+C)")
if app_instance:
app_instance.title = "🧠"
return
if len(selected_text.strip()) < 20:
notify("Cortex", "⚠️ Too short to capture")
if app_instance:
app_instance.title = "🧠"
return
# Detect source app
source = detect_source_app()
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Capturing from {source}")
print(f"Text: {selected_text[:100]}{'...' if len(selected_text) > 100 else ''}")
# Extract context with Claude
extracted = extract_context(selected_text, source=source)
summary = format_extraction_summary(extracted)
print(summary)
# Save to GitHub
saved_files = save_extracted_context(extracted, raw_text=selected_text)
# Embed in Redis
if REDIS_AVAILABLE:
try:
embed_and_store(extracted, raw_text=selected_text)
except Exception:
pass
# Build notification message
counts = []
if extracted.get("KEY_INSIGHTS"):
counts.append(f"{len(extracted['KEY_INSIGHTS'])} insights")
if extracted.get("DECISIONS"):
counts.append(f"{len(extracted['DECISIONS'])} decisions")
if extracted.get("OPEN_QUESTIONS"):
counts.append(f"{len(extracted['OPEN_QUESTIONS'])} questions")
if extracted.get("ACTION_ITEMS"):
counts.append(f"{len(extracted['ACTION_ITEMS'])} actions")
msg = ", ".join(counts) if counts else "summary saved"
notify("✅ Cortex", f"Captured from {source}: {msg}")
print(f"✅ Saved {len(saved_files)} files to GitHub")
except Exception as e:
notify("Cortex", f"❌ Error: {str(e)[:60]}")
print(f"Error: {e}")
finally:
if app_instance:
app_instance.title = "🧠"
def detect_source_app() -> str:
"""Try to detect which app is frontmost."""
try:
result = subprocess.run(
["osascript", "-e", 'tell application "System Events" to get name of first application process whose frontmost is true'],
capture_output=True, text=True
)
app_name = result.stdout.strip().lower()
# Map common apps to source labels
app_map = {
"google chrome": "chrome",
"safari": "safari",
"firefox": "firefox",
"slack": "slack",
"notes": "apple-notes",
"messages": "imessage",
"mail": "email",
"notion": "notion",
"arc": "arc",
}
for key, label in app_map.items():
if key in app_name:
return label
return app_name or "unknown"
except Exception:
return "unknown"
# ── Menu bar app ──────────────────────────────────────────────────────────────
class CortexApp(rumps.App):
def __init__(self):
super().__init__("🧠", quit_button="Quit Cortex")
self.menu = [
rumps.MenuItem("Cortex — Universal Context Layer", callback=None),
None, # separator
rumps.MenuItem("Capture selection (Cmd+Shift+C)", callback=self.manual_capture),
None,
rumps.MenuItem("Open GitHub repo", callback=self.open_repo),
rumps.MenuItem("Status", callback=self.show_status),
]
# Start global hotkey listener in background thread
self.listener_thread = threading.Thread(target=self.start_hotkey_listener, daemon=True)
self.listener_thread.start()
print("🧠 Cortex capture running. Press Cmd+Shift+C anywhere to capture selected text.")
def start_hotkey_listener(self):
pressed = set()
def on_press(key):
pressed.add(key)
# Check if Cmd+Shift+C is all held
if (
key
[truncated — 1406 more characters]
```
### ui.py
```python
"""
cortex/ui.py
------------
Streamlit chat UI for Cortex — your universal second brain.
Run: streamlit run ui.py
"""
import os
import streamlit as st
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()
from instrumentation import setup_tracing
setup_tracing(project_name="cortex")
from anthropic import Anthropic
from redis_store import search_context, get_recent_context
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
# ── Page config ───────────────────────────────────────────────────────────────
st.set_page_config(
page_title="Cortex",
page_icon="🧠",
layout="wide",
initial_sidebar_state="expanded",
)
# ── Styles ────────────────────────────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
# ── Source emoji map ──────────────────────────────────────────────────────────
SOURCE_EMOJI = {
"claude": "⚡",
"claude-chat": "⚡",
"chatgpt": "🤖",
"gemini": "✨",
"slack": "💬",
"imessage": "💬",
"notion": "📝",
"chrome": "🌐",
"safari": "🧭",
"arc": "🌐",
"omi": "🎙️",
"email": "📧",
"apple-notes": "📓",
"unknown": "📌",
}
TYPE_COLOR = {
"insight": "#7c7cff",
"decision": "#ff7c7c",
"open-question": "#ffb07c",
"action": "#7cffb0",
"summary": "#c0c0c0",
"person": "#ff7ce0",
}
def source_emoji(source: str) -> str:
for key, emoji in SOURCE_EMOJI.items():
if key in source.lower():
return emoji
return "📌"
def type_color(note_type: str) -> str:
return TYPE_COLOR.get(note_type, "#888")
# ── Answer with context ───────────────────────────────────────────────────────
def ask_cortex(question: str, context_chunks: list[dict]) -> str:
"""Ask Claude to answer using retrieved Cortex context."""
if not context_chunks:
return "I couldn't find anything relevant in your Cortex. Try capturing more context first using Cmd+C → Cmd+Shift+V."
# Format context for Claude
context_text = "\n\n".join([
f"[{c.get('type', 'note')} from {c.get('source', '?')} on {c.get('timestamp', '')[:10]}]\n{c.get('content', '')}"
for c in context_chunks
])
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=1000,
system="""You are Cortex — a personal AI that knows everything about the user based on their captured context.
You have access to the user's second brain: notes, decisions, insights, and memories captured from their AI chats, Slack, iMessage, and other apps.
Answer questions directly and personally, as if you are their most knowledgeable assistant.
- Reference specific details from the context (dates, sources, exact decisions)
- Be concise but complete
- If the context is partial, say so and answer with what you have
- Never say "based on the provided context" — just answer naturally""",
messages=[{
"role": "user",
"content": f"Context from my second brain:\n\n{context_text}\n\n---\n\nQuestion: {question}"
}]
)
return response.content[0].text
# ── Sidebar ───────────────────────────────────────────────────────────────────
with st.sidebar:
st.markdown("### 🧠 Cortex")
st.markdown("*Your universal second brain*")
st.divider()
# Recent captures
st.markdown("**Recent captures**")
try:
recent = get_recent_context(since=datetime.now().replace(hour=0, minute=0, second=0), top_k=10)
if recent:
for item in recent[:8]:
ts = item.get("timestamp", "")[:16].replace("T", " ")
src = source_emoji(item.get("source", ""))
content_preview = item.get("content", "")[:80]
st.markdown(f"""{src} {ts}
{content_preview}...
""", unsafe_allow_html=True)
else:
st.caption("No captures today yet. Select text anywhere and press Cmd+C → Cmd+Shift+V.")
except Exception:
st.caption("Redis not connected — start with: docker start cortex-redis")
st.divider()
# Stats
repo = os.environ.get("CORTEX_REPO", "")
if repo:
st.markdown(f"**[📁 GitHub repo](https://github.com/{repo})**")
st.markdown("**Hotkey:** Select text → `Cmd+C` → `Cmd+Shift+V`")
# Top K slider
st.divider()
top_k = st.slider("Context depth", min_value=3, max_value=15, value=6,
help="How many memory chunks to retrieve per question")
# ── Main chat area ────────────────────────────────────────────────────────────
st.markdown('', unsafe_allow_html=True)
st.markdown('Ask anything about your life, work, decisions, and conversations.
', unsafe_allow_html=True)
# Init chat history
if "messages" not in st.session_state:
st.session_state.messages = []
# Render chat history
for msg in st.session_state.messages:
with st.chat_message(msg["role"], avatar="🧠" if msg["role"] == "assistant" else "👤"):
st.markdown(msg["content"])
# Show sources if available
if msg.get("sources"):
[truncated — 2726 more characters]
```
### ingest.py
```python
"""
cortex/ingest.py
----------------
Core extraction pipeline. Takes raw text from any source,
calls Claude to extract structured context, returns a dict.
Schema v2: Each item has a stable ID, kind, confidence, importance,
entity references, and topic tags.
"""
import hashlib
import json
import re
import os
from datetime import datetime
# Initialize Arize tracing BEFORE creating Anthropic client
# so auto-instrumentation can wrap the client at import time
from instrumentation import setup_tracing, get_tracer
setup_tracing(project_name="cortex")
from anthropic import Anthropic
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
_tracer = get_tracer()
EXTRACTION_SYSTEM_PROMPT = """You are a context extraction engine for Cortex, a personal second brain.
Given any text (conversation, document, message, transcript), extract structured, atomic context items.
Return a single JSON object with these fields:
"records": list of atomic claims, decisions, or observations — each with:
- "id": stable 8-char hex id prefixed "mem_" (hash of content)
- "kind": "claim" | "decision" | "event" | "preference" | "observation"
- "content": the atomic statement, 1-2 sentences, self-contained
- "confidence": "confirmed" (stated as fact) | "reported" (someone said) | "inferred" (implied)
- "importance": integer 1-5 (5=life-changing decision, 4=significant, 3=notable, 2=useful detail, 1=minor)
- "entity_ids": list of stable entity IDs involved, e.g. ["person_vamika-singhal", "project_cortex"]
- "topics": list of 1-3 topic tags, e.g. ["architecture", "redis", "memory"]
- "occurred_at": ISO datetime string if a time is mentioned, else null
"tasks": list of action items and open questions — each with:
- "id": "task_" + 8 hex chars
- "kind": "action" | "question" | "decision-pending"
- "content": the task or question
- "status": "open"
- "importance": 1-5
- "entity_ids": list
- "topics": list
"entities": people, projects, orgs, and topics mentioned — each with:
- "id": stable slug: "person_first-last" | "project_name" | "org_name"
- "kind": "person" | "project" | "org"
- "name": canonical full name
- "aliases": other names or spellings seen in this text
- "context": 1 sentence describing who/what this is
"summary": 2-3 sentence summary of the overall content
Rules:
- Atomic claims: one fact per record, no compound statements
- Stable IDs: always use the same slug for the same entity (person_vamika-singhal not person_vamika)
- Only extract signal, not noise — empty lists are fine
- Return ONLY valid JSON, no markdown fences, no commentary"""
def make_id(prefix: str, content: str) -> str:
"""Generate a stable short ID from content hash."""
return prefix + hashlib.sha256(content.encode()).hexdigest()[:8]
def extract_context(raw_text: str, source: str = "unknown") -> dict:
"""
Extract structured context from raw text using Claude.
Returns dict with keys: records, tasks, entities, summary,
plus metadata: _source, _source_id, _timestamp, _raw_length
"""
if not raw_text or not raw_text.strip():
return _empty_extraction(source)
MAX_INPUT_CHARS = 40_000
truncated = raw_text[:MAX_INPUT_CHARS]
if len(raw_text) > MAX_INPUT_CHARS:
truncated += f"\n\n[... truncated {len(raw_text) - MAX_INPUT_CHARS} chars ...]"
now = datetime.now().isoformat()
source_id = make_id("src_", source + now[:16])
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=3000,
system=EXTRACTION_SYSTEM_PROMPT,
messages=[{
"role": "user",
"content": (
f"Source: {source}\n"
f"Source ID: {source_id}\n"
f"Captured: {now}\n\n"
f"---\n\n{truncated}\n\n---\n\n"
f"Extract context as JSON:"
)
}]
)
raw_output = response.content[0].text.strip()
extracted = _parse_json_response(raw_output)
# Ensure required fields
extracted.setdefault("records", [])
extracted.setdefault("tasks", [])
extracted.setdefault("entities", [])
extracted.setdefault("summary", "")
# Handle old-format responses (Claude sometimes uses old field names)
_migrate_old_format(extracted)
# Attach metadata
extracted["_source"] = source
extracted["_source_id"] = source_id
extracted["_timestamp"] = now
extracted["_raw_length"] = len(raw_text)
return extracted
def _migrate_old_format(extracted: dict):
"""Migrate old KEY_INSIGHTS / DECISIONS / etc. fields to new schema."""
for insight in extracted.pop("KEY_INSIGHTS", []):
extracted["records"].append({
"id": make_id("mem_", insight),
"kind": "claim",
"content": insight,
"confidence": "confirmed",
"importance": 3,
"entity_ids": [],
"topics": [],
"occurred_at": None,
})
for decision in extracted.pop("DECISIONS", []):
extracted["records"].append({
"id": make_id("mem_", decision),
"kind": "decision",
"content": decision,
"confidence": "confirmed",
"importance": 4,
"entity_ids": [],
"topics": [],
"occurred_at": None,
})
for question in extracted.pop("OPEN_QUESTIONS", []):
extracted["tasks"].append({
"id": make_id("task_", question),
"kind": "question",
"content": question,
"status": "open",
"importance": 3,
"entity_ids": [],
"topics": [],
})
for action in extracted.pop("ACTION_ITEMS", []):
extracted["tasks"].append({
"id": make_id("task_", action),
"kind": "action",
"content": action,
"status": "open",
"importance": 3,
"entity_ids": [],
"topics": [],
[truncated — 3698 more characters]
```
### github_store.py
```python
"""
cortex/github_store.py
----------------------
Push extracted context to GitHub repo via the REST API.
Repo structure (v2):
sources/src_*.md — immutable source manifests
records/YYYY-MM-DD/mem_*.md — dated atomic claims and decisions
entities/people/person_*.md — canonical person pages (append-only)
entities/projects/proj_*.md — canonical project pages
tasks/open/task_*.md — open actions and questions
tasks/resolved/task_*.md — completed tasks
knowledge/decisions/ — high-importance curated decisions
views/ — generated summaries (person dossiers, project briefs)
indexes/entity-registry.json — canonical entity ID → name mapping
Every file has YAML front matter with stable IDs, kind, status,
confidence, importance, source_ids, entity_ids, topics.
"""
import base64
import hashlib
import json
import os
import re
import requests
from datetime import datetime
from typing import Optional
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
CORTEX_REPO = os.environ.get("CORTEX_REPO", "")
GITHUB_API = "https://api.github.com"
BRANCH = os.environ.get("CORTEX_BRANCH", "main")
# ── GitHub helpers ─────────────────────────────────────────────────────────────
def _headers() -> dict:
return {
"Authorization": f"Bearer {GITHUB_TOKEN}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
def _get_existing_sha(filepath: str) -> Optional[str]:
url = f"{GITHUB_API}/repos/{CORTEX_REPO}/contents/{filepath}"
resp = requests.get(url, headers=_headers(), params={"ref": BRANCH})
if resp.status_code == 200:
return resp.json().get("sha")
return None
def _get_existing_content(filepath: str) -> Optional[str]:
url = f"{GITHUB_API}/repos/{CORTEX_REPO}/contents/{filepath}"
resp = requests.get(url, headers=_headers(), params={"ref": BRANCH})
if resp.status_code == 200:
data = resp.json()
return base64.b64decode(data["content"]).decode("utf-8")
return None
def push_file(filepath: str, content: str, commit_message: str) -> bool:
if not GITHUB_TOKEN or not CORTEX_REPO:
raise EnvironmentError("GITHUB_TOKEN and CORTEX_REPO must be set")
url = f"{GITHUB_API}/repos/{CORTEX_REPO}/contents/{filepath}"
sha = _get_existing_sha(filepath)
payload = {
"message": commit_message,
"content": base64.b64encode(content.encode("utf-8")).decode("ascii"),
"branch": BRANCH,
}
if sha:
payload["sha"] = sha
resp = requests.put(url, headers=_headers(), json=payload)
return resp.status_code in (200, 201)
def _slug(text: str, max_len: int = 50) -> str:
s = re.sub(r"[^\w\s-]", "", text.lower())
s = re.sub(r"[\s_]+", "-", s).strip("-")
return s[:max_len]
# ── Front matter builder ───────────────────────────────────────────────────────
def _frontmatter(fields: dict) -> str:
"""Build YAML front matter from a dict."""
lines = ["---"]
for k, v in fields.items():
if isinstance(v, list):
if v:
lines.append(f"{k}:")
for item in v:
lines.append(f" - {item}")
else:
lines.append(f"{k}: []")
elif v is None:
lines.append(f"{k}: null")
else:
# Quote strings that contain special chars
sv = str(v)
if any(c in sv for c in [":", "#", "[", "]", "{", "}"]):
lines.append(f'{k}: "{sv}"')
else:
lines.append(f"{k}: {sv}")
lines.append("---")
return "\n".join(lines)
# ── Main save function ─────────────────────────────────────────────────────────
def save_extracted_context(extracted: dict, raw_text: str = "") -> list[str]:
"""
Save all extracted context items to the GitHub repo (new v2 structure).
Returns list of file paths successfully saved.
"""
source = extracted.get("_source", "unknown")
source_id = extracted.get("_source_id", "src_unknown")
timestamp = extracted.get("_timestamp", datetime.now().isoformat())
date_str = timestamp[:10]
captured_at = timestamp
saved = []
# ── 1. Source manifest ─────────────────────────────────────────────────────
source_path = f"sources/{source_id}.md"
source_fm = _frontmatter({
"id": source_id,
"kind": "source",
"source": source,
"captured_at": captured_at,
"raw_length": extracted.get("_raw_length", 0),
"scope": "private",
})
source_content = f"""{source_fm}
# Source: {source} · {date_str}
**Captured:** {captured_at}
**From:** `{source}`
## Summary
{extracted.get('summary', '_No summary extracted._')}
## Raw Excerpt
> {raw_text[:400].strip()}{"..." if len(raw_text) > 400 else ""}
"""
if push_file(source_path, source_content, f"source({source}): {source_id} @ {date_str}"):
saved.append(source_path)
# ── 2. Records (claims, decisions, observations) ───────────────────────────
for record in extracted.get("records", []):
rid = record.get("id", "mem_unknown")
kind = record.get("kind", "claim")
content = record.get("content", "")
confidence = record.get("confidence", "confirmed")
importance = record.get("importance", 3)
entity_ids = record.get("entity_ids", [])
topics = record.get("topics", [])
occurred_at = record.get("occurred_at")
if not content.strip():
continue
filepath = f"records/{date_str}/{rid}.md"
fm = _frontmatter({
"id": rid,
"kind": kind,
"status": "active",
"occurred_at": occurred_at or captured_at,
"captured_at": captured_at,
"source_ids": [source_id],
"entity_ids": entity_ids,
"topics": topics,
"confidence": confidence,
"importance": importance,
"scope":
[truncated — 6616 more characters]
```
### redis_store.py
```python
"""
cortex/redis_store.py
---------------------
Embed context items and store in Redis Stack for semantic + metadata search.
Uses Voyage AI (voyage-3, 1024-dim) for embeddings.
Index fields:
content — TEXT (full-text search)
source — TEXT
kind — TAG (claim, decision, event, action, question, person, summary)
status — TAG (active, resolved, archived)
confidence — TAG (confirmed, reported, inferred)
topics — TAG (comma-separated)
entity_ids — TAG (comma-separated stable IDs)
importance — NUMERIC (1-5)
timestamp_unix — NUMERIC
embedding — VECTOR (FLAT, COSINE, 1024-dim)
"""
import json
import os
import struct
from datetime import datetime
from typing import Optional
from opentelemetry import trace
_tracer = trace.get_tracer("cortex.redis")
import redis
from redis.commands.search.field import TextField, VectorField, TagField, NumericField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
import voyageai
VOYAGE_API_KEY = os.environ.get("VOYAGE_API_KEY", "")
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
INDEX_NAME = "cortex_idx"
VECTOR_DIM = 1024
DOC_PREFIX = "cortex:note:"
_redis_client = None
_voyage_client = None
def _redis() -> redis.Redis:
global _redis_client
if _redis_client is None:
_redis_client = redis.from_url(REDIS_URL, decode_responses=False)
return _redis_client
def _embed(text: str) -> list[float]:
global _voyage_client
if _voyage_client is None:
_voyage_client = voyageai.Client(api_key=VOYAGE_API_KEY)
result = _voyage_client.embed([text[:32000]], model="voyage-3")
return result.embeddings[0]
def setup_index(drop_existing: bool = False):
"""Create (or recreate) the Redis vector search index."""
r = _redis()
if drop_existing:
try:
r.ft(INDEX_NAME).dropindex()
print(f"Dropped existing index '{INDEX_NAME}'")
except Exception:
pass
else:
try:
r.ft(INDEX_NAME).info()
print(f"Index '{INDEX_NAME}' already exists.")
return
except Exception:
pass
schema = (
TextField("content"),
TextField("source"),
TagField("kind"),
TagField("status"),
TagField("confidence"),
TagField("topics"),
TagField("entity_ids"),
NumericField("importance"),
NumericField("timestamp_unix"),
VectorField(
"embedding",
"FLAT",
{
"TYPE": "FLOAT32",
"DIM": VECTOR_DIM,
"DISTANCE_METRIC": "COSINE",
}
)
)
r.ft(INDEX_NAME).create_index(
schema,
definition=IndexDefinition(prefix=[DOC_PREFIX], index_type=IndexType.HASH)
)
print(f"✅ Created Redis index '{INDEX_NAME}'")
def _pack_embedding(embedding: list[float]) -> bytes:
return struct.pack(f"{len(embedding)}f", *embedding)
def _store_item(
r: redis.Redis,
key: str,
content: str,
source: str,
kind: str,
timestamp_unix: int,
embedding: list[float],
status: str = "active",
confidence: str = "confirmed",
importance: int = 3,
topics: list[str] = None,
entity_ids: list[str] = None,
):
mapping = {
b"content": content.encode("utf-8"),
b"source": source.encode("utf-8"),
b"kind": kind.encode("utf-8"),
b"status": status.encode("utf-8"),
b"confidence": confidence.encode("utf-8"),
b"importance": str(importance).encode("utf-8"),
b"topics": (",".join(topics or [])).encode("utf-8"),
b"entity_ids": (",".join(entity_ids or [])).encode("utf-8"),
b"timestamp_unix": str(timestamp_unix).encode("utf-8"),
b"embedding": _pack_embedding(embedding),
# Legacy compat
b"type": kind.encode("utf-8"),
b"tags": (",".join([kind, source] + (topics or []))).encode("utf-8"),
}
r.hset(key, mapping=mapping)
def embed_and_store(extracted: dict, raw_text: str = "") -> list[str]:
"""
Embed all extracted context items and store in Redis.
Accepts both v2 (records/tasks/entities) and legacy (KEY_INSIGHTS etc.) formats.
Returns list of Redis keys stored.
"""
r = _redis()
source = extracted.get("_source", "unknown")
timestamp = extracted.get("_timestamp", datetime.now().isoformat())
try:
timestamp_unix = int(datetime.fromisoformat(timestamp).timestamp())
except Exception:
timestamp_unix = int(datetime.now().timestamp())
stored_keys = []
key_counter = [0]
def _next_key(kind: str, item_id: str = "") -> str:
k = f"{DOC_PREFIX}{kind}:{source}:{timestamp_unix}:{item_id or key_counter[0]}"
key_counter[0] += 1
return k
def _embed_store(
content: str,
kind: str,
item_id: str = "",
status: str = "active",
confidence: str = "confirmed",
importance: int = 3,
topics: list = None,
entity_ids: list = None,
):
if not content.strip():
return
try:
embedding = _embed(content)
key = _next_key(kind, item_id)
_store_item(
r, key, content, source, kind, timestamp_unix, embedding,
status=status, confidence=confidence, importance=importance,
topics=topics or [], entity_ids=entity_ids or []
)
stored_keys.append(key)
except Exception as e:
print(f" ⚠️ Failed to embed {kind}: {e}", flush=True)
# ── v2 format ──────────────────────────────────────────────────────────────
for record in extracted.get("records", []):
_embed_store(
content=record.get("content", ""),
kind=record.get("kind", "claim"),
item_id=record.get("id", ""),
[truncated — 7542 more characters]
```
[1 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]