Initial commit: self-improving ad copy agent eval harness

End-to-end loop: retrieve (ChromaDB) -> generate (Groq) -> evaluate (LLM-as-judge, 4 dimensions) -> store (SQLite) -> feedback (promote winners to golden + memory, flag losers). Includes versioned prompts with a regression gate, FastAPI /run endpoint, Rich CLI, and a pytest regression suite.

.env.example

@@ -0,0 +1 @@
+GROQ_API_KEY=your_groq_api_key_here

.gitignore

@@ -0,0 +1,17 @@
+# Local notes / source briefs
+jd.txt
+prompt.txt
+
+# Secrets
+.env
+
+# Local databases & vector store
+agent.db
+agent.db-wal
+agent.db-shm
+chroma_db/
+
+# Python
+__pycache__/
+*.pyc
+.pytest_cache/

README.md

@@ -0,0 +1,302 @@
+# Self-Improving Ad Copy Agent
+
+A small agentic system that generates direct-to-consumer (DTC) ad copy,
+evaluates its **own** outputs with an LLM-as-judge, stores the results, and
+uses that signal to improve over time. The loop runs end to end:
+
+```
+retrieve → generate → evaluate → log → improve → repeat
+```
+
+Every good output becomes future few-shot fuel and a regression baseline; every
+bad output gets flagged for review. The system gets better the more it runs —
+without any human editing prompts in the hot path.
+
+---
+
+## What it does (and why it's architecturally interesting)
+
+Most "LLM app" demos are a single prompt with no memory and no notion of
+quality. This project is built around three ideas that make it a genuine
+*self-improving* loop:
+
+1. **It judges itself on multiple axes.** Each output is scored 1–5 on four
+   independent dimensions (hook strength, brand alignment, clarity, conversion
+   intent) by a separate judge model. Dimension scores are first-class and
+   stored separately — there is no single "vibe" score driving decisions.
+
+2. **It remembers what worked.** High-scoring outputs are embedded by their
+   brand brief and stored in a local vector store. On every new run, the agent
+   retrieves the most similar past winners and injects them as few-shot
+   examples. The quality bar therefore ratchets upward over time.
+
+3. **It refuses to regress.** Winning outputs are captured into a *golden
+   dataset*. Any change to the prompt is checked against that dataset before it
+   is allowed to drive real runs: if quality drops by more than a tolerance on
+   any known-good brief, the change fails.
+
+The result is a closed feedback loop where generation, evaluation, memory, and
+prompt versioning reinforce each other.
+
+---
+
+## Architecture
+
+```
+              ┌──────────────────────────── agent/core.py ───────────────────────────┐
+brand brief ─▶│ 1. retrieve top-3 high-scoring past outputs   (agent/memory.py, Chroma)│
+              │ 2. inject as few-shot examples                (agent/prompts.py)        │
+              │ 3. generate headline / body / cta             (agent/tools.py → Groq)   │
+              │ 4. judge each variant on 4 dimensions         (evals/judge.py → Groq)   │
+              │ 5. store run + outputs + scores               (db/store.py → SQLite)    │
+              │ 6. promote winners (≥4.0) → golden + memory   (feedback/loop.py)        │
+              │ 7. flag losers (<2.5) for review              (feedback/loop.py)        │
+              └────────────────────────────────────────────────────────────────────────┘
+```
+
+| Layer            | File(s)                          | Responsibility                                  |
+|------------------|----------------------------------|-------------------------------------------------|
+| Persistence      | `db/store.py`                    | SQLite: runs, outputs, golden, flagged          |
+| Prompts          | `agent/prompts.py`               | All prompt text, versioned; nothing hardcoded elsewhere |
+| Memory           | `agent/memory.py`                | ChromaDB + sentence-transformers retrieval      |
+| Agent loop       | `agent/core.py`, `agent/tools.py`| Orchestration + Groq client                     |
+| Evaluation       | `evals/*.py`                     | Rubric, judge, golden dataset, regression runner|
+| Feedback         | `feedback/loop.py`               | Promote winners, flag losers                    |
+| API              | `api/main.py`                    | `POST /run`                                      |
+| CLI / tests      | `scripts/run_eval.py`, `tests/`  | Manual runs and the regression suite            |
+
+---
+
+## Setup
+
+### Requirements
+- Python 3.11+
+- A Groq API key (free at <https://console.groq.com/keys>)
+
+### Install
+
+```bash
+python -m venv .venv
+# Windows:  .venv\Scripts\activate
+# macOS/Linux:  source .venv/bin/activate
+
+pip install -r requirements.txt
+```
+
+### Configure
+
+```bash
+cp .env.example .env          # Windows: copy .env.example .env
+# then edit .env and set GROQ_API_KEY
+```
+
+Only `GROQ_API_KEY` is required. Everything else (SQLite at `./agent.db`,
+ChromaDB at `./chroma_db`) is local — no other external services.
+
+### First run
+
+```bash
+# Run the agent on the built-in FitFuel example brief:
+python scripts/run_eval.py run
+
+# Or run on your own brief:
+python scripts/run_eval.py run --brief my_brief.json
+
+# Inspect golden / flagged counts at any time:
+python scripts/run_eval.py status
+```
+
+The very first run retrieves zero few-shot examples (memory is empty). As you
+run more briefs, winners accumulate and later runs start standing on the
+shoulders of earlier ones.
+
+A brief is a JSON object:
+
+```json
+{
+  "brand": "FitFuel",
+  "product": "High-protein meal replacement shake",
+  "audience": "Busy professionals aged 25-40",
+  "tone": "Energetic and no-nonsense",
+  "goal": "Drive trial purchases"
+}
+```
+
+### Run the API
+
+```bash
+uvicorn api.main:app --reload
+```
+
+Then trigger the full loop:
+
+```bash
+curl -X POST http://127.0.0.1:8000/run \
+  -H "Content-Type: application/json" \
+  -d '{"brand":"FitFuel","product":"High-protein meal replacement shake","audience":"Busy professionals aged 25-40","tone":"Energetic and no-nonsense","goal":"Drive trial purchases"}'
+```
+
+The response contains the three generated variants, all four dimension scores
+per variant, the weighted average, and a feedback summary (what was promoted /
+flagged).
+
+---
+
+## How the self-improving loop works
+
+1. **Retrieve.** The incoming brief is embedded (locally, via
+   `all-MiniLM-L6-v2`) and used to query ChromaDB for the **top-3 most similar
+   past outputs that scored ≥ 3.5/5**. Only proven-good copy is ever retrieved.
+
+2. **Generate.** Those examples are injected into the active generation prompt
+   under a *"Past high-performing examples"* section, and the model produces a
+   headline hook, body copy, and a CTA in one structured JSON response.
+
+3. **Evaluate.** Each of the three variants is immediately scored by the judge
+   model on the four rubric dimensions. Scores are clamped to 1–5 and a weighted
+   average is computed for internal ranking only.
+
+4. **Log.** The run, every variant, and every dimension score are written to
+   SQLite.
+
+5. **Improve.** The feedback loop:
+   - promotes any output with a weighted average **≥ 4.0** into both the golden
+     dataset and ChromaDB memory (so it can be retrieved next time);
+   - flags any output **< 2.5** into the `flagged_outputs` table with the reason
+     *"below quality threshold."*
+
+Because winners re-enter memory, the pool of few-shot exemplars improves run
+over run — that is the "self-improving" part.
+
+---
+
+## Evaluation & regression tests
+
+### The rubric (`evals/rubric.py`)
+Four dimensions, each 1–5:
+
+| Dimension          | Question                                            | Weight |
+|--------------------|-----------------------------------------------------|:------:|
+| `hook_strength`    | Does the headline immediately grab attention?       | 0.30   |
+| `brand_alignment`  | Does the copy reflect the brief tone and audience?  | 0.25   |
+| `clarity`          | Is the message immediately understandable?          | 0.25   |
+| `conversion_intent`| Does it drive toward the stated goal?               | 0.20   |
+
+The weighted average is **internal only** — used for thresholds and ranking.
+All four raw dimensions are always stored separately.
+
+### Golden dataset (`evals/golden.py`)
+Any output with weighted average **≥ 4.0** is captured (brief, output, all
+scores, prompt version, timestamp). This is the regression baseline.
+
+### Regression runner (`evals/runner.py`)
+For each golden entry it regenerates copy for the same brief + variant using the
+prompt version under test, re-judges it, and compares the new weighted score to
+the stored baseline. If any entry **drops by more than 0.5**, the run fails with
+a clear warning.
+
+```bash
+# Run the regression eval against the active prompt and print a Rich table:
+python scripts/run_eval.py regression
+```
+
+### Pytest suite (`tests/test_regression.py`)
+
+```bash
+pytest tests/test_regression.py
+```
+
+It loads the golden dataset, runs the eval runner, and asserts no entry
+regresses more than 0.5 from baseline. The test **skips** (rather than failing
+spuriously) when `GROQ_API_KEY` is unset or the golden dataset is still empty.
+
+---
+
+## Prompt versioning & how to swap versions safely
+
+All prompt text lives in `agent/prompts.py` — nothing is hardcoded anywhere
+else. Each prompt is a named, versioned constant (`GENERATION_PROMPT_V1`,
+`GENERATION_PROMPT_V2`, …). A single constant selects which is live:
+
+```python
+ACTIVE_PROMPT_VERSION = "GENERATION_PROMPT_V1"
+```
+
+Every run logs the version it used (stored on the run and on each output).
+
+**To promote a new prompt safely:**
+
+1. Add a new constant (e.g. `GENERATION_PROMPT_V2`) and register it in
+   `PROMPT_REGISTRY`.
+2. Run the regression check against it **before** making it active:
+   ```bash
+   # Either run the suite, or the CLI regression command after flipping the
+   # constant in a branch:
+   pytest tests/test_regression.py
+   python scripts/run_eval.py regression
+   ```
+3. Only if no golden entry regresses more than 0.5, change
+   `ACTIVE_PROMPT_VERSION` to the new version.
+
+This is what "changing the active version triggers a regression check before it
+is used in a real run" means in practice: the golden dataset is the gate.
+`GENERATION_PROMPT_V2` ships in this repo as a worked example you can promote.
+
+---
+
+## Design Decisions
+
+**Why ChromaDB for memory.** The agent needs *semantic* retrieval — "find past
+briefs like this one" — not exact lookups. ChromaDB gives a persistent local
+vector store with cosine similarity and zero external services, and it pairs
+cleanly with local sentence-transformers embeddings. SQLite alone can't do
+nearest-neighbour search over brief semantics; a hosted vector DB would violate
+the "local only" constraint and add ops overhead for no benefit at this scale.
+
+**Why dimension scoring over a composite score.** A single 1–10 "quality" score
+is unactionable and easy for a judge to anchor on. Scoring four independent
+dimensions tells you *why* copy is weak (great hook, poor clarity) and makes the
+signal far more stable and debuggable. We do compute a weighted average, but
+only for internal thresholds/ranking — the four raw dimensions are always
+stored, so we never lose information by collapsing too early.
+
+**Why SQLite for eval storage.** Eval results are structured, relational, and
+queryable (runs → outputs → scores; golden; flagged). SQLite gives ACID
+guarantees, trivial setup, a single-file database, and real SQL — ideal for run
+history and a golden dataset. It needs no server and ships with Python. A hosted
+DB would add infrastructure with no upside for a local, single-node harness.
+
+**Why the 4.0 threshold for golden inclusion.** On a 1–5 scale, 4.0 means
+"clearly good on the weighted blend" without demanding perfection. Set it lower
+and the golden set fills with mediocre copy, weakening both the regression
+baseline and the few-shot exemplars. Set it higher (e.g. 4.5) and you rarely
+capture anything, so the system never accumulates a baseline or improves. 4.0 is
+the point where entries are good enough to *defend* against regressions and to
+*teach* future runs. (The retrieval floor is a more permissive 3.5 so memory can
+draw on a slightly wider pool of solid examples, while only the strongest ≥4.0
+outputs become protected golden baselines.)
+
+---
+
+## Project layout
+
+```
+self-improving-agent/
+├── README.md
+├── requirements.txt
+├── .env.example
+├── agent/        core loop, tools, memory, versioned prompts
+├── evals/        rubric, LLM judge, golden dataset, regression runner
+├── feedback/     post-run promote/flag loop
+├── db/           SQLite store
+├── api/          FastAPI app (POST /run)
+├── tests/        pytest regression suite
+└── scripts/      run_eval.py CLI (run / regression / status)
+```
+
+## Constraints honoured
+- No LangChain — the agent loop is built directly.
+- No managed eval platform — the judge and runner are custom.
+- No external database — SQLite + ChromaDB, local only.
+- No prompts hardcoded outside `agent/prompts.py`.
+- Synchronous throughout, except where FastAPI's interface applies.

agent/__init__.py

@@ -0,0 +1,15 @@
+"""Agent package init.
+
+Inject the operating system's certificate trust store into Python's SSL stack
+before any network call (Groq API, Hugging Face model download). In corporate
+environments that intercept TLS, the proxy's root CA lives in the OS store but
+not in Python's bundled certifi bundle, which otherwise causes
+CERTIFICATE_VERIFY_FAILED. This keeps verification ON (no insecure downgrade).
+"""
+
+try:  # truststore is optional; ignore if unavailable.
+    import truststore
+
+    truststore.inject_into_ssl()
+except Exception:  # noqa: BLE001 — best-effort; fall back to default certs.
+    pass

agent/core.py

@@ -0,0 +1,95 @@
+"""The main agent loop: retrieve -> generate -> evaluate -> store -> feedback.
+
+This ties memory, generation, the eval judge, persistence, and the feedback
+loop together. It is intentionally synchronous and dependency-injected so it
+can be driven from the API, the CLI, or tests with the same code path.
+"""
+
+from typing import Any, Optional
+
+from agent import prompts, tools
+from agent.memory import Memory
+from db.store import Store
+
+# The three variants the agent always produces, in order.
+VARIANT_TYPES = ("headline", "body", "cta")
+
+
+class Agent:
+    def __init__(
+        self,
+        store: Optional[Store] = None,
+        memory: Optional[Memory] = None,
+        temperature: float = 0.7,
+    ):
+        self.store = store or Store()
+        self.memory = memory or Memory()
+        self.temperature = temperature
+
+    # --------------------------------------------------------------- generate
+    def generate_variants(self, brief: dict[str, Any], few_shot_block: str, prompt_version: str) -> dict[str, str]:
+        """Call the LLM once and parse out the three ad-copy variants."""
+        prompt = prompts.render_generation_prompt(brief, few_shot_block, prompt_version)
+        raw = tools.chat(prompt, temperature=self.temperature)
+        parsed = tools.extract_json(raw)
+        return {
+            "headline": str(parsed.get("headline", "")).strip(),
+            "body": str(parsed.get("body", "")).strip(),
+            "cta": str(parsed.get("cta", "")).strip(),
+        }
+
+    # -------------------------------------------------------------------- run
+    def run(self, brief: dict[str, Any]) -> dict[str, Any]:
+        """Execute the full loop for one brand brief and return outputs + scores."""
+        # Imported here to keep module import order clean (evals/feedback import
+        # nothing from core, but core depends on them at call time).
+        from evals.judge import judge_output
+        from feedback.loop import run_feedback
+
+        prompt_version = prompts.ACTIVE_PROMPT_VERSION
+
+        # 1. Retrieve top-3 most relevant high-scoring past outputs.
+        retrieved = self.memory.retrieve(brief, k=3)
+
+        # 2. Inject them as few-shot examples.
+        few_shot_block = prompts.build_few_shot_block(retrieved)
+
+        # 3. Generate the three variants.
+        variants = self.generate_variants(brief, few_shot_block, prompt_version)
+
+        # 4. Evaluate every variant immediately.
+        # 5. Persist the run and each scored output.
+        run_id = self.store.create_run(brief, prompt_version)
+
+        scored_outputs: list[dict[str, Any]] = []
+        for variant_type in VARIANT_TYPES:
+            content = variants[variant_type]
+            scores = judge_output(brief, variant_type, content)
+            self.store.add_output(run_id, variant_type, content, scores, prompt_version)
+            scored_outputs.append(
+                {
+                    "variant_type": variant_type,
+                    "content": content,
+                    "scores": scores,
+                }
+            )
+
+        # 6 + 7. Feedback loop: promote good outputs to golden + memory,
+        # flag poor ones for review.
+        feedback_summary = run_feedback(
+            store=self.store,
+            memory=self.memory,
+            brief=brief,
+            run_id=run_id,
+            scored_outputs=scored_outputs,
+            prompt_version=prompt_version,
+        )
+
+        return {
+            "run_id": run_id,
+            "brief": brief,
+            "prompt_version": prompt_version,
+            "retrieved_examples": len(retrieved),
+            "outputs": scored_outputs,
+            "feedback": feedback_summary,
+        }

agent/memory.py

@@ -0,0 +1,127 @@
+"""Retrieval and storage logic backed by ChromaDB + sentence-transformers.
+
+Memory holds past generated outputs that scored well, embedded by their brand
+brief. On each new run we retrieve the top-k most similar high-scoring entries
+and feed them back as few-shot examples — this is the substrate of the
+self-improving loop.
+
+Only entries scoring above RETRIEVAL_SCORE_FLOOR (3.5/5) are ever stored or
+returned, so the agent only learns from outputs that actually worked.
+"""
+
+import os
+
+# sentence-transformers pulls in `transformers`, which will try to import a
+# TensorFlow/Keras backend if one is present. We only use the PyTorch path, so
+# disable the TF backend before anything imports transformers. (Avoids the
+# "Keras 3 is not supported" import error in envs that have TF/Keras 3.)
+os.environ.setdefault("USE_TF", "0")
+os.environ.setdefault("USE_TORCH", "1")
+os.environ.setdefault("TRANSFORMERS_NO_ADVISORY_WARNINGS", "1")
+
+import json
+import uuid
+from typing import Any
+
+import chromadb
+from chromadb.utils import embedding_functions
+
+CHROMA_DIR = os.getenv("CHROMA_DIR", "./chroma_db")
+COLLECTION_NAME = "ad_copy_memory"
+EMBED_MODEL = "all-MiniLM-L6-v2"
+
+# Minimum weighted score for an entry to live in / be retrieved from memory.
+RETRIEVAL_SCORE_FLOOR = 3.5
+
+
+def _brief_to_text(brief: dict[str, Any]) -> str:
+    """Flatten a brand brief into a single string used for embedding."""
+    parts = [
+        brief.get("brand", ""),
+        brief.get("product", ""),
+        brief.get("audience", ""),
+        brief.get("tone", ""),
+        brief.get("goal", ""),
+    ]
+    return " | ".join(str(p) for p in parts if p)
+
+
+class Memory:
+    """Vector memory of high-scoring ad copy, keyed by brief similarity."""
+
+    def __init__(self, persist_dir: str = CHROMA_DIR):
+        self._client = chromadb.PersistentClient(path=persist_dir)
+        # sentence-transformers embedding function, computed locally.
+        self._embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
+            model_name=EMBED_MODEL
+        )
+        self._collection = self._client.get_or_create_collection(
+            name=COLLECTION_NAME,
+            embedding_function=self._embed_fn,
+            metadata={"hnsw:space": "cosine"},
+        )
+
+    def add(
+        self,
+        brief: dict[str, Any],
+        variant_type: str,
+        output: str,
+        score: float,
+        prompt_version: str,
+        timestamp: str,
+    ) -> bool:
+        """Store one high-scoring output. Returns False if below the floor."""
+        if score < RETRIEVAL_SCORE_FLOOR:
+            return False
+
+        self._collection.add(
+            ids=[str(uuid.uuid4())],
+            documents=[output],
+            metadatas=[
+                {
+                    "brief": json.dumps(brief),
+                    "variant_type": variant_type,
+                    "score": float(score),
+                    "prompt_version": prompt_version,
+                    "timestamp": timestamp,
+                }
+            ],
+        )
+        return True
+
+    def retrieve(self, brief: dict[str, Any], k: int = 3) -> list[dict[str, Any]]:
+        """Return up to k most similar past entries with score >= the floor.
+
+        Results are ordered by vector similarity to the incoming brief.
+        """
+        count = self._collection.count()
+        if count == 0:
+            return []
+
+        results = self._collection.query(
+            query_texts=[_brief_to_text(brief)],
+            # Over-fetch so the score filter still leaves us close to k.
+            n_results=min(max(k * 3, k), count),
+            where={"score": {"$gte": RETRIEVAL_SCORE_FLOOR}},
+        )
+
+        docs = results.get("documents", [[]])[0]
+        metas = results.get("metadatas", [[]])[0]
+
+        examples: list[dict[str, Any]] = []
+        for doc, meta in zip(docs, metas):
+            examples.append(
+                {
+                    "brief": json.loads(meta.get("brief", "{}")),
+                    "variant_type": meta.get("variant_type", "output"),
+                    "output": doc,
+                    "score": meta.get("score", 0.0),
+                    "prompt_version": meta.get("prompt_version", ""),
+                }
+            )
+            if len(examples) >= k:
+                break
+        return examples
+
+    def count(self) -> int:
+        return self._collection.count()

agent/prompts.py

@@ -0,0 +1,176 @@
+"""All prompts live here and nowhere else.
+
+Each prompt is versioned with an explicit string key. The single source of
+truth for which generation prompt is live is ACTIVE_PROMPT_VERSION. Swapping
+that constant is the supported way to roll a new prompt forward — and it is
+what the regression check guards (see evals/runner.py).
+
+Nothing in the codebase should hardcode prompt text outside this module.
+"""
+
+# ---------------------------------------------------------------------------
+# Active version selector
+# ---------------------------------------------------------------------------
+# Change this to promote a new generation prompt. The regression runner is
+# expected to pass against the golden dataset before this is used for real.
+ACTIVE_PROMPT_VERSION = "GENERATION_PROMPT_V1"
+
+
+# ---------------------------------------------------------------------------
+# Generation prompts
+# ---------------------------------------------------------------------------
+GENERATION_PROMPT_V1 = """You are a senior direct-to-consumer (DTC) ad copywriter.
+
+You will be given a brand brief. Write three distinct pieces of ad copy:
+1. A HEADLINE hook — one short line that immediately grabs attention.
+2. A BODY copy — 2-3 sentences that build desire and reflect the brand tone.
+3. A CTA — a single short call to action that drives the stated goal.
+
+Rules:
+- Match the brief's tone and speak directly to the described audience.
+- Be specific and concrete. Avoid generic filler and clichés.
+- Keep the headline punchy and the CTA action-oriented.
+- Return ONLY valid JSON, no preamble, no markdown fences.
+
+Return exactly this JSON shape:
+{{
+  "headline": "<headline hook>",
+  "body": "<body copy>",
+  "cta": "<call to action>"
+}}
+
+{few_shot_block}
+
+Brand brief:
+{brief}
+"""
+
+# A second version kept as a worked example of how to add and promote prompts.
+# It nudges the model toward tighter, benefit-led copy. To use it, set
+# ACTIVE_PROMPT_VERSION = "GENERATION_PROMPT_V2" and run the regression suite.
+GENERATION_PROMPT_V2 = """You are an award-winning DTC performance copywriter who writes
+copy that converts cold traffic.
+
+You will be given a brand brief. Produce three pieces of ad copy:
+1. HEADLINE — a scroll-stopping hook of at most 8 words.
+2. BODY — 2-3 sentences leading with the strongest benefit, in the brand's voice.
+3. CTA — an imperative call to action tied directly to the stated goal.
+
+Rules:
+- Lead with benefit, not feature. Speak to the exact audience in the brief.
+- No clichés, no hype words ("revolutionary", "game-changer"), no emojis.
+- Return ONLY valid JSON, no preamble, no markdown fences.
+
+Return exactly this JSON shape:
+{{
+  "headline": "<headline hook>",
+  "body": "<body copy>",
+  "cta": "<call to action>"
+}}
+
+{few_shot_block}
+
+Brand brief:
+{brief}
+"""
+
+
+# Registry so other modules can resolve a version string to its template.
+PROMPT_REGISTRY = {
+    "GENERATION_PROMPT_V1": GENERATION_PROMPT_V1,
+    "GENERATION_PROMPT_V2": GENERATION_PROMPT_V2,
+}
+
+
+# ---------------------------------------------------------------------------
+# Few-shot helpers
+# ---------------------------------------------------------------------------
+FEW_SHOT_HEADER = "Past high-performing examples (learn from their style, do not copy verbatim):"
+
+
+def build_few_shot_block(examples: list[dict]) -> str:
+    """Render retrieved high-scoring examples into a prompt section.
+
+    Each example is a dict with keys: brief (dict), variant_type, output, score.
+    Returns an empty string when there are no examples so the prompt stays clean.
+    """
+    if not examples:
+        return ""
+
+    lines = [FEW_SHOT_HEADER, ""]
+    for i, ex in enumerate(examples, 1):
+        brief = ex.get("brief", {})
+        product = brief.get("product", "") if isinstance(brief, dict) else ""
+        lines.append(
+            f"Example {i} ({ex.get('variant_type', 'output')}, "
+            f"score {ex.get('score', 0):.2f}/5) for product '{product}':"
+        )
+        lines.append(f'  "{ex.get("output", "")}"')
+        lines.append("")
+    return "\n".join(lines).strip()
+
+
+def get_generation_prompt(version: str) -> str:
+    """Return the raw template for a version string, defaulting to the active one."""
+    return PROMPT_REGISTRY.get(version, PROMPT_REGISTRY[ACTIVE_PROMPT_VERSION])
+
+
+def render_generation_prompt(brief: dict, few_shot_block: str, version: str | None = None) -> str:
+    """Fill a generation prompt template with the brief and few-shot examples."""
+    import json
+
+    version = version or ACTIVE_PROMPT_VERSION
+    template = get_generation_prompt(version)
+    return template.format(
+        brief=json.dumps(brief, indent=2),
+        few_shot_block=few_shot_block,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Judge prompt (LLM-as-judge). Versioned alongside generation prompts.
+# ---------------------------------------------------------------------------
+JUDGE_PROMPT_VERSION = "JUDGE_PROMPT_V1"
+
+JUDGE_PROMPT_V1 = """You are a strict, fair advertising copy evaluator.
+
+Score the candidate ad copy on FOUR dimensions, each an integer from 1 to 5:
+- hook_strength: does it immediately grab attention?
+- brand_alignment: does it reflect the brief's tone and speak to the audience?
+- clarity: is the message immediately understandable?
+- conversion_intent: does it drive toward the stated goal?
+
+Be discriminating. Reserve 5 for genuinely excellent copy and 1 for copy that
+fails the dimension entirely. Do NOT return a single composite score.
+
+Return ONLY valid JSON, no preamble, no markdown fences, exactly:
+{{
+  "hook_strength": <1-5>,
+  "brand_alignment": <1-5>,
+  "clarity": <1-5>,
+  "conversion_intent": <1-5>,
+  "rationale": "<one short sentence>"
+}}
+
+Brand brief:
+{brief}
+
+Candidate copy ({variant_type}):
+"{output}"
+"""
+
+JUDGE_REGISTRY = {
+    "JUDGE_PROMPT_V1": JUDGE_PROMPT_V1,
+}
+
+
+def render_judge_prompt(brief: dict, variant_type: str, output: str, version: str | None = None) -> str:
+    import json
+
+    version = version or JUDGE_PROMPT_VERSION
+    template = JUDGE_REGISTRY.get(version, JUDGE_PROMPT_V1)
+    return template.format(
+        brief=json.dumps(brief, indent=2),
+        variant_type=variant_type,
+        output=output,
+    )

agent/tools.py

@@ -0,0 +1,75 @@
+"""Agent tools: the LLM client wrapper and the structured-generation helpers.
+
+We deliberately call Groq directly (no LangChain). A single chat-completion
+helper is shared by both the generator and the judge so model/config lives in
+one place.
+"""
+
+import json
+import os
+import re
+from typing import Any, Optional
+
+from groq import Groq
+
+DEFAULT_MODEL = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile")
+
+_client: Optional[Groq] = None
+
+
+def get_client() -> Groq:
+    """Lazily construct a shared Groq client. Requires GROQ_API_KEY."""
+    global _client
+    if _client is None:
+        api_key = os.getenv("GROQ_API_KEY")
+        if not api_key:
+            raise RuntimeError(
+                "GROQ_API_KEY is not set. Copy .env.example to .env and add your key."
+            )
+        _client = Groq(api_key=api_key)
+    return _client
+
+
+def chat(
+    prompt: str,
+    temperature: float = 0.7,
+    model: str | None = None,
+    max_tokens: int = 1024,
+) -> str:
+    """Single-shot chat completion returning the raw assistant text."""
+    client = get_client()
+    resp = client.chat.completions.create(
+        model=model or DEFAULT_MODEL,
+        messages=[{"role": "user", "content": prompt}],
+        temperature=temperature,
+        max_tokens=max_tokens,
+    )
+    return resp.choices[0].message.content or ""
+
+
+def extract_json(text: str) -> dict[str, Any]:
+    """Best-effort parse of a JSON object out of a model response.
+
+    Models occasionally wrap JSON in prose or markdown fences despite
+    instructions; we strip fences and fall back to the first {...} block.
+    """
+    text = text.strip()
+
+    # Strip ```json ... ``` or ``` ... ``` fences if present.
+    fence = re.match(r"^```(?:json)?\s*(.*?)\s*```$", text, re.DOTALL)
+    if fence:
+        text = fence.group(1).strip()
+
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        pass
+
+    # Fall back to the first balanced-looking object.
+    match = re.search(r"\{.*\}", text, re.DOTALL)
+    if match:
+        try:
+            return json.loads(match.group(0))
+        except json.JSONDecodeError as exc:
+            raise ValueError(f"Could not parse JSON from model output: {exc}\nRaw: {text}")
+    raise ValueError(f"No JSON object found in model output.\nRaw: {text}")

api/main.py

@@ -0,0 +1,54 @@
+"""FastAPI app exposing a single endpoint to trigger the full agent loop.
+
+POST /run with a brand brief -> retrieve -> generate -> evaluate -> feedback,
+returning the generated outputs and their scores.
+
+Run with:  uvicorn api.main:app --reload
+"""
+
+from typing import Any
+
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+
+load_dotenv()  # pick up GROQ_API_KEY from .env if present
+
+from agent.core import Agent  # noqa: E402  (after load_dotenv on purpose)
+
+app = FastAPI(title="Self-Improving Ad Copy Agent", version="1.0.0")
+
+# A single shared agent (and therefore shared Store/Memory) for the process.
+_agent: Agent | None = None
+
+
+def get_agent() -> Agent:
+    global _agent
+    if _agent is None:
+        _agent = Agent()
+    return _agent
+
+
+class BrandBrief(BaseModel):
+    brand: str = Field(..., examples=["FitFuel"])
+    product: str = Field(..., examples=["High-protein meal replacement shake"])
+    audience: str = Field(..., examples=["Busy professionals aged 25-40"])
+    tone: str = Field(..., examples=["Energetic and no-nonsense"])
+    goal: str = Field(..., examples=["Drive trial purchases"])
+
+
+@app.get("/")
+def root() -> dict[str, str]:
+    return {"status": "ok", "endpoint": "POST /run with a brand brief"}
+
+
+@app.post("/run")
+def run(brief: BrandBrief) -> dict[str, Any]:
+    """Trigger the full agent loop for a brand brief."""
+    try:
+        agent = get_agent()
+        return agent.run(brief.model_dump())
+    except RuntimeError as exc:  # e.g. missing GROQ_API_KEY
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+    except Exception as exc:  # noqa: BLE001 — surface generation/judge failures
+        raise HTTPException(status_code=502, detail=f"Agent run failed: {exc}") from exc

db/store.py

@@ -0,0 +1,245 @@
+"""SQLite interface for runs, scores, golden dataset, flagged outputs, and versions.
+
+This is the persistence backbone of the system. Everything else depends on it.
+All access goes through the Store class so the schema lives in exactly one place.
+"""
+
+import json
+import os
+import sqlite3
+import threading
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+DEFAULT_SQLITE_PATH = os.getenv("SQLITE_PATH", "./agent.db")
+
+
+def _utcnow() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+class Store:
+    """Thin, explicit wrapper over a local SQLite database.
+
+    A single connection is shared with a lock so the same Store can be used
+    from FastAPI request handlers and CLI scripts without surprises.
+    """
+
+    def __init__(self, path: str = DEFAULT_SQLITE_PATH):
+        self.path = path
+        self._lock = threading.Lock()
+        self._conn = sqlite3.connect(path, check_same_thread=False)
+        self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA journal_mode=WAL;")
+        self._init_schema()
+
+    # ------------------------------------------------------------------ schema
+    def _init_schema(self) -> None:
+        with self._lock, self._conn:
+            self._conn.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS runs (
+                    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+                    brief           TEXT NOT NULL,        -- JSON brand brief
+                    prompt_version  TEXT NOT NULL,
+                    timestamp       TEXT NOT NULL
+                );
+
+                CREATE TABLE IF NOT EXISTS outputs (
+                    id                INTEGER PRIMARY KEY AUTOINCREMENT,
+                    run_id            INTEGER NOT NULL,
+                    variant_type      TEXT NOT NULL,      -- headline | body | cta
+                    content           TEXT NOT NULL,
+                    hook_strength     REAL,
+                    brand_alignment   REAL,
+                    clarity           REAL,
+                    conversion_intent REAL,
+                    weighted_average  REAL,
+                    prompt_version    TEXT NOT NULL,
+                    timestamp         TEXT NOT NULL,
+                    FOREIGN KEY (run_id) REFERENCES runs(id)
+                );
+
+                CREATE TABLE IF NOT EXISTS golden (
+                    id                INTEGER PRIMARY KEY AUTOINCREMENT,
+                    brief             TEXT NOT NULL,
+                    variant_type      TEXT NOT NULL,
+                    output            TEXT NOT NULL,
+                    hook_strength     REAL,
+                    brand_alignment   REAL,
+                    clarity           REAL,
+                    conversion_intent REAL,
+                    weighted_average  REAL NOT NULL,
+                    prompt_version    TEXT NOT NULL,
+                    timestamp         TEXT NOT NULL
+                );
+
+                CREATE TABLE IF NOT EXISTS flagged_outputs (
+                    id                INTEGER PRIMARY KEY AUTOINCREMENT,
+                    run_id            INTEGER,
+                    brief             TEXT NOT NULL,
+                    variant_type      TEXT NOT NULL,
+                    output            TEXT NOT NULL,
+                    weighted_average  REAL,
+                    reason            TEXT NOT NULL,
+                    timestamp         TEXT NOT NULL
+                );
+                """
+            )
+
+    # -------------------------------------------------------------------- runs
+    def create_run(self, brief: dict[str, Any], prompt_version: str) -> int:
+        with self._lock, self._conn:
+            cur = self._conn.execute(
+                "INSERT INTO runs (brief, prompt_version, timestamp) VALUES (?, ?, ?)",
+                (json.dumps(brief), prompt_version, _utcnow()),
+            )
+            return int(cur.lastrowid)
+
+    def get_run(self, run_id: int) -> Optional[dict[str, Any]]:
+        with self._lock:
+            row = self._conn.execute(
+                "SELECT * FROM runs WHERE id = ?", (run_id,)
+            ).fetchone()
+        return dict(row) if row else None
+
+    # ----------------------------------------------------------------- outputs
+    def add_output(
+        self,
+        run_id: int,
+        variant_type: str,
+        content: str,
+        scores: dict[str, float],
+        prompt_version: str,
+    ) -> int:
+        """Persist one generated variant together with its 4 dimension scores."""
+        with self._lock, self._conn:
+            cur = self._conn.execute(
+                """
+                INSERT INTO outputs (
+                    run_id, variant_type, content,
+                    hook_strength, brand_alignment, clarity, conversion_intent,
+                    weighted_average, prompt_version, timestamp
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    run_id,
+                    variant_type,
+                    content,
+                    scores.get("hook_strength"),
+                    scores.get("brand_alignment"),
+                    scores.get("clarity"),
+                    scores.get("conversion_intent"),
+                    scores.get("weighted_average"),
+                    prompt_version,
+                    _utcnow(),
+                ),
+            )
+            return int(cur.lastrowid)
+
+    def get_outputs_for_run(self, run_id: int) -> list[dict[str, Any]]:
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT * FROM outputs WHERE run_id = ? ORDER BY id", (run_id,)
+            ).fetchall()
+        return [dict(r) for r in rows]
+
+    # ------------------------------------------------------------------ golden
+    def add_golden(
+        self,
+        brief: dict[str, Any],
+        variant_type: str,
+        output: str,
+        scores: dict[str, float],
+        prompt_version: str,
+    ) -> int:
+        with self._lock, self._conn:
+            cur = self._conn.execute(
+                """
+                INSERT INTO golden (
+                    brief, variant_type, output,
+                    hook_strength, brand_alignment, clarity, conversion_intent,
+                    weighted_average, prompt_version, timestamp
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    json.dumps(brief),
+                    variant_type,
+                    output,
+                    scores.get("hook_strength"),
+                    scores.get("brand_alignment"),
+                    scores.get("clarity"),
+                    scores.get("conversion_intent"),
+                    scores.get("weighted_average"),
+                    prompt_version,
+                    _utcnow(),
+                ),
+            )
+            return int(cur.lastrowid)
+
+    def get_golden(self) -> list[dict[str, Any]]:
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT * FROM golden ORDER BY id"
+            ).fetchall()
+        result = []
+        for r in rows:
+            d = dict(r)
+            d["brief"] = json.loads(d["brief"])
+            result.append(d)
+        return result
+
+    def golden_exists(self, brief: dict[str, Any], variant_type: str, output: str) -> bool:
+        """Avoid inserting an identical golden entry twice."""
+        with self._lock:
+            row = self._conn.execute(
+                "SELECT 1 FROM golden WHERE brief = ? AND variant_type = ? AND output = ? LIMIT 1",
+                (json.dumps(brief), variant_type, output),
+            ).fetchone()
+        return row is not None
+
+    # ----------------------------------------------------------------- flagged
+    def add_flagged(
+        self,
+        brief: dict[str, Any],
+        variant_type: str,
+        output: str,
+        weighted_average: Optional[float],
+        reason: str,
+        run_id: Optional[int] = None,
+    ) -> int:
+        with self._lock, self._conn:
+            cur = self._conn.execute(
+                """
+                INSERT INTO flagged_outputs (
+                    run_id, brief, variant_type, output, weighted_average, reason, timestamp
+                ) VALUES (?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    run_id,
+                    json.dumps(brief),
+                    variant_type,
+                    output,
+                    weighted_average,
+                    reason,
+                    _utcnow(),
+                ),
+            )
+            return int(cur.lastrowid)
+
+    def get_flagged(self) -> list[dict[str, Any]]:
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT * FROM flagged_outputs ORDER BY id"
+            ).fetchall()
+        result = []
+        for r in rows:
+            d = dict(r)
+            d["brief"] = json.loads(d["brief"])
+            result.append(d)
+        return result
+
+    # ------------------------------------------------------------------- close
+    def close(self) -> None:
+        with self._lock:
+            self._conn.close()

evals/golden.py

@@ -0,0 +1,47 @@
+"""Golden dataset manager.
+
+The golden dataset is the system's regression baseline: outputs that scored
+well enough (weighted average >= GOLDEN_THRESHOLD) are captured with their
+brief, scores, and prompt version. The regression runner later re-scores each
+entry against the active prompt and checks for drift.
+"""
+
+from typing import Any
+
+from db.store import Store
+
+# An output earns a golden slot only at or above this weighted average.
+GOLDEN_THRESHOLD = 4.0
+
+
+class GoldenDataset:
+    def __init__(self, store: Store | None = None):
+        self.store = store or Store()
+
+    def qualifies(self, weighted_average: float) -> bool:
+        return weighted_average >= GOLDEN_THRESHOLD
+
+    def maybe_add(
+        self,
+        brief: dict[str, Any],
+        variant_type: str,
+        output: str,
+        scores: dict[str, float],
+        prompt_version: str,
+    ) -> bool:
+        """Add to the golden dataset if it qualifies and isn't already present.
+
+        Returns True if a new golden entry was created.
+        """
+        if not self.qualifies(scores.get("weighted_average", 0.0)):
+            return False
+        if self.store.golden_exists(brief, variant_type, output):
+            return False
+        self.store.add_golden(brief, variant_type, output, scores, prompt_version)
+        return True
+
+    def all(self) -> list[dict[str, Any]]:
+        return self.store.get_golden()
+
+    def size(self) -> int:
+        return len(self.store.get_golden())

evals/judge.py

@@ -0,0 +1,41 @@
+"""LLM-as-judge scorer.
+
+Uses the same Groq model as the generator (llama-3.3-70b-versatile) but with a
+strict, low-temperature judging prompt. Returns the four dimension scores plus
+a derived weighted average and a one-line rationale. It never collapses the
+output into a single composite the way a naive scorer would.
+"""
+
+from typing import Any
+
+from agent import prompts, tools
+from evals import rubric
+
+# Judging is deterministic-ish: low temperature for stable, repeatable scores.
+JUDGE_TEMPERATURE = 0.0
+
+
+def judge_output(brief: dict[str, Any], variant_type: str, output: str) -> dict[str, Any]:
+    """Score a single ad-copy variant on the four rubric dimensions.
+
+    Returns a dict containing each dimension (1-5), the weighted_average, and a
+    rationale string. Robust to a malformed judge response: dimensions clamp to
+    the valid range and default low.
+    """
+    if not output.strip():
+        # Nothing to score — treat as the floor.
+        scores = rubric.normalize_scores({})
+        scores["rationale"] = "Empty output."
+        return scores
+
+    prompt = prompts.render_judge_prompt(brief, variant_type, output)
+    raw = tools.chat(prompt, temperature=JUDGE_TEMPERATURE, max_tokens=512)
+
+    try:
+        parsed = tools.extract_json(raw)
+    except ValueError:
+        parsed = {}
+
+    scores = rubric.normalize_scores(parsed)
+    scores["rationale"] = str(parsed.get("rationale", "")).strip()
+    return scores

evals/rubric.py

@@ -0,0 +1,59 @@
+"""Scoring rubric definitions.
+
+The rubric is the contract between the judge and the rest of the system:
+the four dimensions, their 1-5 scale, and the weighting used to derive a
+single internal number. Per the design, dimension scores are first-class and
+stored separately; the weighted average exists only for ranking/thresholds.
+"""
+
+# The four scored dimensions, each on a 1-5 integer scale.
+DIMENSIONS = ("hook_strength", "brand_alignment", "clarity", "conversion_intent")
+
+# Weights used ONLY to compute an internal weighted average. They sum to 1.0.
+WEIGHTS = {
+    "hook_strength": 0.30,
+    "brand_alignment": 0.25,
+    "clarity": 0.25,
+    "conversion_intent": 0.20,
+}
+
+# Human-readable descriptions, surfaced in the judge prompt and docs.
+DIMENSION_DESCRIPTIONS = {
+    "hook_strength": "Does the headline immediately grab attention?",
+    "brand_alignment": "Does the copy reflect the brief tone and audience?",
+    "clarity": "Is the message immediately understandable?",
+    "conversion_intent": "Does it drive toward the stated goal?",
+}
+
+SCALE_MIN = 1
+SCALE_MAX = 5
+
+
+def weighted_average(scores: dict[str, float]) -> float:
+    """Compute the internal weighted average from dimension scores.
+
+    Missing dimensions are treated as 0 so a malformed judge response surfaces
+    as a low score rather than silently passing.
+    """
+    total = 0.0
+    for dim, weight in WEIGHTS.items():
+        total += float(scores.get(dim, 0)) * weight
+    return round(total, 4)
+
+
+def clamp(value: float) -> int:
+    """Clamp a raw score into the valid 1-5 integer range."""
+    try:
+        v = int(round(float(value)))
+    except (TypeError, ValueError):
+        v = SCALE_MIN
+    return max(SCALE_MIN, min(SCALE_MAX, v))
+
+
+def normalize_scores(raw: dict) -> dict[str, float]:
+    """Validate/clamp the four dimensions and attach the weighted average."""
+    scores: dict[str, float] = {}
+    for dim in DIMENSIONS:
+        scores[dim] = clamp(raw.get(dim))
+    scores["weighted_average"] = weighted_average(scores)
+    return scores

evals/runner.py

@@ -0,0 +1,129 @@
+"""Eval runner: runs the golden dataset against any prompt config and checks
+for regressions.
+
+For each golden entry we regenerate copy for the same brief + variant using the
+prompt version under test, re-judge it, and compare the new weighted score to
+the entry's stored baseline. If any entry drops by more than REGRESSION_TOLERANCE
+(0.5), the run is marked failed.
+
+This is what guards a prompt-version swap (see agent/prompts.py) and what the
+pytest regression suite drives.
+"""
+
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+from agent import prompts, tools
+from evals.golden import GoldenDataset
+from evals.judge import judge_output
+
+# A golden entry may not drop more than this from its baseline before we fail.
+REGRESSION_TOLERANCE = 0.5
+
+
+@dataclass
+class EntryResult:
+    brief: dict[str, Any]
+    variant_type: str
+    baseline_score: float
+    new_score: float
+    new_output: str
+    regressed: bool
+
+    @property
+    def delta(self) -> float:
+        return round(self.new_score - self.baseline_score, 4)
+
+
+@dataclass
+class EvalReport:
+    prompt_version: str
+    results: list[EntryResult] = field(default_factory=list)
+
+    @property
+    def passed(self) -> bool:
+        return not any(r.regressed for r in self.results)
+
+    @property
+    def regressions(self) -> list[EntryResult]:
+        return [r for r in self.results if r.regressed]
+
+    @property
+    def count(self) -> int:
+        return len(self.results)
+
+
+def _generate_variant(brief: dict[str, Any], variant_type: str, prompt_version: str) -> str:
+    """Regenerate a single variant for a brief using a given prompt version.
+
+    Few-shot examples are intentionally omitted so the regression isolates the
+    prompt itself rather than whatever happens to be in memory.
+    """
+    prompt = prompts.render_generation_prompt(brief, few_shot_block="", version=prompt_version)
+    raw = tools.chat(prompt, temperature=0.7)
+    parsed = tools.extract_json(raw)
+    return str(parsed.get(variant_type, "")).strip()
+
+
+def run_golden_eval(
+    prompt_version: Optional[str] = None,
+    golden: Optional[GoldenDataset] = None,
+) -> EvalReport:
+    """Run every golden entry against `prompt_version` (defaults to active)."""
+    prompt_version = prompt_version or prompts.ACTIVE_PROMPT_VERSION
+    golden = golden or GoldenDataset()
+
+    report = EvalReport(prompt_version=prompt_version)
+
+    for entry in golden.all():
+        brief = entry["brief"]
+        variant_type = entry["variant_type"]
+        baseline = float(entry["weighted_average"])
+
+        new_output = _generate_variant(brief, variant_type, prompt_version)
+        new_scores = judge_output(brief, variant_type, new_output)
+        new_score = float(new_scores["weighted_average"])
+
+        regressed = (baseline - new_score) > REGRESSION_TOLERANCE
+        report.results.append(
+            EntryResult(
+                brief=brief,
+                variant_type=variant_type,
+                baseline_score=baseline,
+                new_score=new_score,
+                new_output=new_output,
+                regressed=regressed,
+            )
+        )
+
+    return report
+
+
+def format_report(report: EvalReport) -> str:
+    """Plain-text summary of a regression run (used as a fallback to Rich)."""
+    lines = [
+        f"Regression eval for prompt version: {report.prompt_version}",
+        f"Entries checked: {report.count}",
+        f"Tolerance: drop > {REGRESSION_TOLERANCE} fails",
+        "",
+    ]
+    if report.count == 0:
+        lines.append("No golden entries yet — nothing to check. (PASS)")
+        return "\n".join(lines)
+
+    for i, r in enumerate(report.results, 1):
+        status = "REGRESSED" if r.regressed else "ok"
+        lines.append(
+            f"  [{i}] {r.variant_type:8s} baseline={r.baseline_score:.2f} "
+            f"new={r.new_score:.2f} delta={r.delta:+.2f}  {status}"
+        )
+
+    lines.append("")
+    if report.passed:
+        lines.append("RESULT: PASS — no entry regressed beyond tolerance.")
+    else:
+        lines.append(
+            f"RESULT: FAIL — {len(report.regressions)} entry(ies) regressed beyond "
+            f"{REGRESSION_TOLERANCE}. Do not promote this prompt version."
+        )
+    return "\n".join(lines)

feedback/loop.py

@@ -0,0 +1,82 @@
+"""Feedback loop: runs after every agent generation.
+
+Responsibilities:
+1. Read the scores from the latest run.
+2. Promote high-scoring outputs (weighted avg >= GOLDEN_THRESHOLD) into the
+   golden dataset AND into ChromaDB memory so future runs can retrieve them.
+3. Flag low-scoring outputs (weighted avg < FLAG_THRESHOLD) into the
+   flagged_outputs table with a clear reason.
+
+This is the closing arc of the self-improving loop: good outputs become future
+few-shot fuel and regression baselines; bad ones get surfaced for review.
+"""
+
+from datetime import datetime, timezone
+from typing import Any
+
+from agent.memory import Memory
+from db.store import Store
+from evals.golden import GOLDEN_THRESHOLD, GoldenDataset
+
+# Outputs scoring strictly below this are flagged for human review.
+FLAG_THRESHOLD = 2.5
+FLAG_REASON = "below quality threshold"
+
+
+def run_feedback(
+    store: Store,
+    memory: Memory,
+    brief: dict[str, Any],
+    run_id: int,
+    scored_outputs: list[dict[str, Any]],
+    prompt_version: str,
+) -> dict[str, Any]:
+    """Process one run's outputs: promote the good, flag the bad.
+
+    `scored_outputs` is a list of {variant_type, content, scores} dicts.
+    Returns a summary describing what changed.
+    """
+    golden = GoldenDataset(store=store)
+    now = datetime.now(timezone.utc).isoformat()
+
+    promoted: list[str] = []
+    flagged: list[str] = []
+
+    for item in scored_outputs:
+        variant_type = item["variant_type"]
+        content = item["content"]
+        scores = item["scores"]
+        weighted = float(scores.get("weighted_average", 0.0))
+
+        # 2. Promote high scorers to golden + memory.
+        if weighted >= GOLDEN_THRESHOLD:
+            added = golden.maybe_add(brief, variant_type, content, scores, prompt_version)
+            memory.add(
+                brief=brief,
+                variant_type=variant_type,
+                output=content,
+                score=weighted,
+                prompt_version=prompt_version,
+                timestamp=now,
+            )
+            if added:
+                promoted.append(variant_type)
+
+        # 3. Flag low scorers for review.
+        elif weighted < FLAG_THRESHOLD:
+            store.add_flagged(
+                brief=brief,
+                variant_type=variant_type,
+                output=content,
+                weighted_average=weighted,
+                reason=FLAG_REASON,
+                run_id=run_id,
+            )
+            flagged.append(variant_type)
+
+    return {
+        "promoted_to_golden": promoted,
+        "flagged_for_review": flagged,
+        "golden_threshold": GOLDEN_THRESHOLD,
+        "flag_threshold": FLAG_THRESHOLD,
+    }

requirements.txt

@@ -0,0 +1,9 @@
+groq==0.13.0
+chromadb==0.5.23
+sentence-transformers==3.3.1
+fastapi==0.115.6
+uvicorn==0.34.0
+rich==13.9.4
+pytest==8.3.4
+python-dotenv==1.0.1
+pydantic==2.10.4

scripts/run_eval.py

@@ -0,0 +1,169 @@
+"""CLI to trigger an agent run or a regression eval and print results nicely.
+
+Usage:
+    python scripts/run_eval.py run                # run the agent on the example brief
+    python scripts/run_eval.py run --brief brief.json
+    python scripts/run_eval.py regression         # run golden regression eval
+    python scripts/run_eval.py status             # show golden / flagged counts
+
+Uses Rich for clean terminal output.
+"""
+
+import argparse
+import json
+import os
+import sys
+
+# Make the project root importable when run as `python scripts/run_eval.py`.
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from dotenv import load_dotenv  # noqa: E402
+
+load_dotenv()
+
+from rich.console import Console  # noqa: E402
+from rich.panel import Panel  # noqa: E402
+from rich.table import Table  # noqa: E402
+
+console = Console()
+
+EXAMPLE_BRIEF = {
+    "brand": "FitFuel",
+    "product": "High-protein meal replacement shake",
+    "audience": "Busy professionals aged 25-40",
+    "tone": "Energetic and no-nonsense",
+    "goal": "Drive trial purchases",
+}
+
+
+def _load_brief(path: str | None) -> dict:
+    if not path:
+        return EXAMPLE_BRIEF
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def cmd_run(args: argparse.Namespace) -> int:
+    from agent.core import Agent
+
+    brief = _load_brief(args.brief)
+    console.print(Panel.fit(json.dumps(brief, indent=2), title="Brand Brief", border_style="cyan"))
+
+    with console.status("[bold green]Running agent loop (retrieve → generate → evaluate → feedback)..."):
+        agent = Agent()
+        result = agent.run(brief)
+
+    console.print(
+        f"\n[bold]Run #{result['run_id']}[/bold]  "
+        f"prompt=[magenta]{result['prompt_version']}[/magenta]  "
+        f"retrieved few-shot examples=[yellow]{result['retrieved_examples']}[/yellow]\n"
+    )
+
+    table = Table(title="Generated Variants & Scores", show_lines=True)
+    table.add_column("Variant", style="cyan", no_wrap=True)
+    table.add_column("Copy", style="white", max_width=50)
+    table.add_column("Hook", justify="right")
+    table.add_column("Brand", justify="right")
+    table.add_column("Clarity", justify="right")
+    table.add_column("Conv", justify="right")
+    table.add_column("Weighted", justify="right", style="bold")
+
+    for o in result["outputs"]:
+        s = o["scores"]
+        table.add_row(
+            o["variant_type"],
+            o["content"],
+            str(s["hook_strength"]),
+            str(s["brand_alignment"]),
+            str(s["clarity"]),
+            str(s["conversion_intent"]),
+            f"{s['weighted_average']:.2f}",
+        )
+    console.print(table)
+
+    fb = result["feedback"]
+    console.print(
+        f"\n[green]Promoted to golden:[/green] {fb['promoted_to_golden'] or 'none'}   "
+        f"[red]Flagged for review:[/red] {fb['flagged_for_review'] or 'none'}"
+    )
+    return 0
+
+
+def cmd_regression(args: argparse.Namespace) -> int:
+    from evals.runner import format_report, run_golden_eval
+
+    with console.status("[bold green]Running golden regression eval..."):
+        report = run_golden_eval()
+
+    table = Table(title=f"Regression Eval — {report.prompt_version}", show_lines=True)
+    table.add_column("#", justify="right")
+    table.add_column("Variant", style="cyan")
+    table.add_column("Baseline", justify="right")
+    table.add_column("New", justify="right")
+    table.add_column("Delta", justify="right")
+    table.add_column("Status", justify="center")
+
+    for i, r in enumerate(report.results, 1):
+        status = "[red]REGRESSED[/red]" if r.regressed else "[green]ok[/green]"
+        table.add_row(
+            str(i), r.variant_type,
+            f"{r.baseline_score:.2f}", f"{r.new_score:.2f}",
+            f"{r.delta:+.2f}", status,
+        )
+
+    if report.count:
+        console.print(table)
+    console.print()
+    if report.passed:
+        console.print(Panel.fit("PASS — no entry regressed beyond tolerance.", border_style="green"))
+    else:
+        console.print(
+            Panel.fit(
+                f"FAIL — {len(report.regressions)} entry(ies) regressed. "
+                "Do not promote this prompt version.",
+                border_style="red",
+            )
+        )
+    # Non-zero exit on failure so it can gate CI / a prompt swap.
+    return 0 if report.passed else 1
+
+
+def cmd_status(args: argparse.Namespace) -> int:
+    from db.store import Store
+    from evals.golden import GoldenDataset
+
+    store = Store()
+    golden = GoldenDataset(store=store)
+    flagged = store.get_flagged()
+
+    console.print(
+        Panel.fit(
+            f"Golden entries: [green]{golden.size()}[/green]\n"
+            f"Flagged outputs: [red]{len(flagged)}[/red]",
+            title="System Status",
+            border_style="cyan",
+        )
+    )
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Self-improving ad copy agent CLI")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    p_run = sub.add_parser("run", help="Run the agent loop on a brief")
+    p_run.add_argument("--brief", help="Path to a brief JSON file (defaults to FitFuel example)")
+    p_run.set_defaults(func=cmd_run)
+
+    p_reg = sub.add_parser("regression", help="Run the golden-dataset regression eval")
+    p_reg.set_defaults(func=cmd_regression)
+
+    p_status = sub.add_parser("status", help="Show golden / flagged counts")
+    p_status.set_defaults(func=cmd_status)
+
+    args = parser.parse_args()
+    return args.func(args)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_regression.py

@@ -0,0 +1,48 @@
+"""Regression test suite.
+
+Loads the golden dataset and runs the eval runner against the active prompt
+version, asserting that no golden entry regresses more than 0.5 from its
+baseline score.
+
+Run with:  pytest tests/test_regression.py
+
+Notes:
+- Requires GROQ_API_KEY (the runner regenerates + re-judges via Groq). If it is
+  not set, the test is skipped rather than failing spuriously.
+- If the golden dataset is empty (fresh install, no runs yet), there is nothing
+  to regress against and the test passes trivially.
+"""
+
+import os
+
+import pytest
+from dotenv import load_dotenv
+
+load_dotenv()
+
+from evals.golden import GoldenDataset  # noqa: E402
+from evals.runner import REGRESSION_TOLERANCE, format_report, run_golden_eval  # noqa: E402
+
+
+@pytest.fixture(scope="module")
+def golden() -> GoldenDataset:
+    return GoldenDataset()
+
+
+def test_no_golden_regressions(golden: GoldenDataset):
+    if not os.getenv("GROQ_API_KEY"):
+        pytest.skip("GROQ_API_KEY not set; regression eval needs live model calls.")
+
+    if golden.size() == 0:
+        pytest.skip("Golden dataset is empty; nothing to regress against yet.")
+
+    report = run_golden_eval(golden=golden)
+
+    # Print a readable summary so failures are diagnosable in CI logs.
+    print("\n" + format_report(report))
+
+    assert report.passed, (
+        f"{len(report.regressions)} golden entry(ies) regressed more than "
+        f"{REGRESSION_TOLERANCE} against prompt '{report.prompt_version}'. "
+        "Do not promote this prompt version."
+    )