Personal Agent MVP

A minimal, cost-effective MVP of a personalized AI agent — not a chatbot. It recognizes individual users, loads their profile/history/long-term memory, makes grounded product recommendations, proactively reaches out on events (birthdays), and replies to inbound messages — all as a single runnable Python service.

Built with LangGraph as the orchestration brain instead of n8n/Make.com, so the core logic lives in plain, testable Python code.

What it can do

Requirement	How it's implemented
Recognize users via existing API	User API adapter layer (app/mock_api.py) — mock data out of the box, real REST API via .env
Access profile, preferences, history	Fetched per user at workflow start (load_user node)
Remember context	Long-term memory facts table in SQLite (app/memory.py), auto-written after every interaction
Personalized recommendations	Rule-based candidate pre-filter + LLM only picks & justifies (anti-hallucination guardrail)
React to events (birthdays)	APScheduler daily cron job (app/scheduler.py)
Connect to API / database	SQLite storage layer + httpx adapters for external APIs
Automated workflows	7-node LangGraph state machine shared by both trigger paths

Quick Start

git clone <this-repo>
cd personal-agent-mvp

python3 -m venv .venv
.venv/bin/pip install -r requirements.txt

# optional: copy the config template and edit it
cp .env.example .env

# run the full demo: birthday scan -> personalized message -> memory write-back -> inbound chat
.venv/bin/python run_birthday_demo.py

No API keys needed — LLM_MOCK_MODE=true (default) runs the entire pipeline offline with a built-in mock LLM, mock users and mock catalog.

Other run modes:

.venv/bin/python -m app.scheduler            # resident process: daily birthday job (9:00 by default)
.venv/bin/uvicorn app.api:app --port 8000    # webhook service: POST /webhook/message

Try the inbound chat path:

curl -X POST http://localhost:8000/webhook/message \
  -H "Content-Type: application/json" \
  -d '{"user_id": "u_1001", "text": "any gift without nuts?"}'

Architecture

                    ┌───────────────────────────────────────────┐
                    │            Trigger layer (2 entries)      │
                    │  1. APScheduler cron (daily 9:00)         │
                    │  2. FastAPI webhook (inbound messages)    │
                    └────────────────────┬──────────────────────┘
                                         │ user_id / text
                                         ▼
                    ┌───────────────────────────────────────────┐
                    │      LangGraph StateGraph (7 nodes)       │
                    │  load_user -> load_memories ->            │
                    │  fetch_candidates -> personalize ->       │
                    │  guardrail -> deliver -> persist_memory   │
                    └────┬─────────────┬──────────────┬─────────┘
                         │             │              │
          ┌──────────────▼───┐  ┌──────▼───────┐  ┌───▼─────────────┐
          │ External adapters│  │   LLM layer  │  │   Data layer    │
          │ - User API (REST)│  │ OpenAI-      │  │ SQLite:         │
          │ - Resend (email) │  │ compatible,  │  │ - memory_facts  │
          │ - console (demo) │  │ JSON output, │  │ - interactions  │
          └──────────────────┘  │ offline mock │  │ - catalog       │
                                └──────────────┘  └─────────────────┘

Workflow nodes

1. load_user — identify user via existing API, fetch profile + order history
2. load_memories — load active long-term memory facts
3. fetch_candidates — business-rule pre-filter (preference tags, history categories, price range, stock, allergy bans) down to ~8 candidates
4. personalize — LLM composes message + picks 1–3 recommendations from candidates only, returns strict JSON
5. guardrail — every recommendation id must exist in the candidate set; hallucinated ids are dropped; empty message falls back to plain greeting
6. deliver — send via channel (console / Resend email) and log the interaction
7. persist_memory — LLM extracts new memory facts (add / invalidate) for the next conversation

Both trigger paths (birthday event and inbound message) run through the same graph, so memory, recommendation and guardrail logic are written once.

Components

Layer	Tech	Notes
Orchestration	LangGraph	state machine, conditional edges
Scheduling	APScheduler	cron trigger, resident process
API / webhook	FastAPI + uvicorn	/webhook/message, /jobs/birthday/run, /health
LLM	OpenAI-compatible HTTP API	swap to DeepSeek/Qwen/self-hosted via OPENAI_BASE_URL; offline mock mode included
Database	SQLite (stdlib sqlite3)	single file, zero ops; swap to Postgres later
External APIs	httpx	user-system adapter, Resend email
Config	python-dotenv	everything controlled by .env

Database schema (3 tables)

• memory_facts — long-term memory: (user_id, fact, fact_type, active, source, timestamps). active=0 soft-deletes so memory is auditable and recoverable. Upgrade path: add embeddings + pgvector, same interface.
• interactions — every outbound/inbound touch: trigger, channel, message, recommendations JSON, status. Powers observability and recent context.
• catalog — recommendation candidate pool: id, name, category, price, tags, stock. In production, sync from your own product DB/API.

Configuration

Copy .env.example to .env and set what you need:

Variable	Default	Purpose
LLM_MOCK_MODE	true	run offline without any API key
OPENAI_API_KEY / OPENAI_BASE_URL / LLM_MODEL	—	any OpenAI-compatible provider
DATABASE_PATH	./data/agent.db	SQLite file location
CHANNEL	console	console or resend_email
RESEND_API_KEY / EMAIL_FROM	—	real email delivery (3,000 free/month)
BIRTHDAY_CRON_HOUR / TIMEZONE	9 / Asia/Shanghai	birthday job schedule
USER_API_BASE / USER_API_KEY	—	your existing user API (empty = mock data)

Going to production

MVP	Production	Change required
SQLite	PostgreSQL (+pgvector)	rewrite app/db.py only
Keyword-matched facts	semantic memory retrieval	memory.load_facts
Mock users	real REST API	set USER_API_BASE / USER_API_KEY
console / email	WhatsApp Business, SMS	add one function in app/channels.py
Direct invocation	LangGraph checkpointer + send queue	pass checkpointer at graph compile
Standalone	n8n as integration glue calling this service	keep both

Project structure

personal-agent-mvp/
├── .env.example              # all config switches
├── requirements.txt
├── run_birthday_demo.py      # one-command end-to-end demo
└── app/
    ├── config.py             # settings from .env
    ├── state.py              # LangGraph shared state
    ├── mock_api.py           # user-system adapter (mock + real REST)
    ├── db.py                 # SQLite layer (memory / interactions / catalog)
    ├── memory.py             # long-term memory facts (add / invalidate / load)
    ├── recommendations.py    # rule-based candidate pre-filter
    ├── llm.py                # structured-JSON LLM client + offline mock
    ├── channels.py           # delivery channels (console / Resend)
    ├── graph.py              # the 7-node LangGraph workflow
    ├── scheduler.py          # daily birthday cron
    └── api.py                # FastAPI webhook service

Verified behaviors

• Birthday scan picks the right users and personalizes per profile/history
• After an interaction, memory facts are written back (e.g. "severely allergic to nuts — must avoid nut products")
• Allergy memory filters banned products out of future recommendations
• A hallucinated recommendation id (sku_999) is intercepted by the guardrail