Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 148 additions & 0 deletions demos/06_context_compaction.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
"""Demo 6: host-side prompt replacement from authoritative compiled state."""

from context_compiler import create_engine
from context_compiler.const import FOCUS_PRIMARY, STATE_FACTS
from demos.common import is_verbose, print_info_report

DEMO_NAME = "06_context_compaction — superseded directives eliminated"


def _build_baseline_prompt(transcript_turns: list[str]) -> str:
transcript_lines = "\n".join(f"User: {turn}" for turn in transcript_turns)
return (
"You are a helpful assistant.\n"
"Use the full transcript context below:\n"
f"{transcript_lines}\n"
"Respond using the latest user preference."
)


def _build_compiled_prompt(compiled_focus: str) -> str:
return (
"You are a helpful assistant.\n"
"Host-side authoritative compiled context:\n"
f"- facts.focus.primary: {compiled_focus}\n"
"Use only this compiled state as the active context."
)


def _print_verbose_report(
*,
transcript_turns: list[str],
compiled_context: str,
baseline_prompt: str,
compiled_prompt: str,
baseline_context_length: int,
compiled_context_length: int,
context_reduction: int,
baseline_prompt_length: int,
compiled_prompt_length: int,
prompt_reduction: int,
) -> None:
print(DEMO_NAME)
print()
print("Raw transcript context:")
for turn in transcript_turns:
print(f"User: {turn}")
print()
print("Compiled context:")
print(compiled_context)
print()
print("Baseline prompt:")
print(baseline_prompt)
print()
print("Compiled prompt:")
print(compiled_prompt)
print()
print("Context size comparison:")
print(f"baseline context length: {baseline_context_length}")
print(f"compiled context length: {compiled_context_length}")
print(f"reduction: {context_reduction}%")
print()
print("Prompt size comparison:")
print(f"baseline prompt length: {baseline_prompt_length}")
print(f"compiled prompt length: {compiled_prompt_length}")
print(f"reduction: {prompt_reduction}%")
print("result: compiled authoritative state replaced superseded transcript directives")


def _print_compact_report(
*,
baseline_context_length: int,
compiled_context_length: int,
baseline_prompt_length: int,
compiled_prompt_length: int,
context_reduction: int,
prompt_reduction: int,
) -> None:
print(DEMO_NAME)
print(f"context: {baseline_context_length} → {compiled_context_length} chars")
print(f"prompt: {baseline_prompt_length} → {compiled_prompt_length} chars")
print(f"reduction: context {context_reduction}%; prompt {prompt_reduction}%")
print("result: compiled authoritative state replaced superseded transcript directives")


def main() -> None:
engine = create_engine()
transcript_turns = [
"use vegetarian curry",
"actually vegan curry",
"actually tofu curry",
"actually lentil curry",
"actually chickpea curry",
]
for turn in transcript_turns:
engine.step(turn)

compiled_focus = engine.state[STATE_FACTS][FOCUS_PRIMARY]
assert compiled_focus is not None

baseline_context = "\n".join(f"User: {turn}" for turn in transcript_turns)
compiled_context = f"- facts.focus.primary: {compiled_focus}"

baseline_prompt = _build_baseline_prompt(transcript_turns)
compiled_prompt = _build_compiled_prompt(compiled_focus)

baseline_context_length = len(baseline_context)
compiled_context_length = len(compiled_context)
context_reduction = round((1 - (compiled_context_length / baseline_context_length)) * 100)
baseline_prompt_length = len(baseline_prompt)
compiled_prompt_length = len(compiled_prompt)
prompt_reduction = round((1 - (compiled_prompt_length / baseline_prompt_length)) * 100)

if is_verbose():
_print_verbose_report(
transcript_turns=transcript_turns,
compiled_context=compiled_context,
baseline_prompt=baseline_prompt,
compiled_prompt=compiled_prompt,
baseline_context_length=baseline_context_length,
compiled_context_length=compiled_context_length,
context_reduction=context_reduction,
baseline_prompt_length=baseline_prompt_length,
compiled_prompt_length=compiled_prompt_length,
prompt_reduction=prompt_reduction,
)
else:
_print_compact_report(
baseline_context_length=baseline_context_length,
compiled_context_length=compiled_context_length,
baseline_prompt_length=baseline_prompt_length,
compiled_prompt_length=compiled_prompt_length,
context_reduction=context_reduction,
prompt_reduction=prompt_reduction,
)

print_info_report(
name=DEMO_NAME,
baseline_context_length=baseline_context_length,
compiled_context_length=compiled_context_length,
context_reduction_percent=context_reduction,
baseline_prompt_length=baseline_prompt_length,
compiled_prompt_length=compiled_prompt_length,
prompt_reduction_percent=prompt_reduction,
)


if __name__ == "__main__":
main()
41 changes: 27 additions & 14 deletions demos/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,8 @@
These scripts compare baseline prompting vs compiler-mediated prompting using the
Context Compiler decision/state API.
They are illustrative manual demos, not benchmarks or CI tests.
They demonstrate common LLM failure modes and how authoritative compiled state
can improve reliability.

## Requirements

Expand Down Expand Up @@ -35,7 +37,7 @@ export MODEL=gpt-4.1-mini

## Run

Run one demo:
Run a single demo:

```bash
uv run python -m demos.run_demo 1
Expand All @@ -47,28 +49,39 @@ Run all demos:
uv run python -m demos.run_demo all
```

Verbose mode:
Run all demos with detailed traces:

```bash
uv run python -m demos.run_demo all --verbose
```

Output modes:
## Output modes

- default (concise): each demo prints the following:
- scenario name + short description
- `baseline: PASS|FAIL`
- `compiler: PASS|FAIL`
- `Default (concise)`:
- scenario name + description
- for evaluative demos (`01`–`05`):
- `baseline: PASS|FAIL`
- `compiler: PASS|FAIL`
- expected behavior
- actual outcome (plain English)
- actual outcome
- `result: ...` (short deterministic description)
- a blank line between demos when running `all`
- final summary:
- `Baseline results: X passed, Y failed`
- `Compiler results: X passed, Y failed`
- `--verbose`: prints detailed traces
- for `06_context_compaction`:
- `context: <baseline> → <compiled> chars`
- `prompt: <baseline> → <compiled> chars`
- `reduction: context <pct>%; prompt <pct>%`
- `result: ...`
- when running `all`:
- blank line between demos
- final summary with evaluative totals
- informational summary line for `06_context_compaction` (non-scored)
- `Verbose (--verbose)`:
- user inputs
- compiler decisions and state
- compiler decisions and compiled state
- prompts/messages sent to the LLM
- output excerpts
- host checks and final verdict
- for `06_context_compaction`, also:
- raw transcript context
- compiled context
- baseline and compiled prompts
- context and prompt size comparisons
40 changes: 40 additions & 0 deletions demos/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,18 @@ class DemoReport(TypedDict):
demo_pass: bool


class InfoReport(TypedDict):
name: str
baseline_context_length: int
compiled_context_length: int
context_reduction_percent: int
baseline_prompt_length: int
compiled_prompt_length: int
prompt_reduction_percent: int


LAST_REPORT: DemoReport | None = None
LAST_INFO_REPORT: InfoReport | None = None


def canonical_json(obj: Any) -> str:
Expand Down Expand Up @@ -138,6 +149,35 @@ def consume_last_report() -> DemoReport | None:
return value


def print_info_report(
*,
name: str,
baseline_context_length: int,
compiled_context_length: int,
context_reduction_percent: int,
baseline_prompt_length: int,
compiled_prompt_length: int,
prompt_reduction_percent: int,
) -> None:
global LAST_INFO_REPORT
LAST_INFO_REPORT = {
"name": name,
"baseline_context_length": baseline_context_length,
"compiled_context_length": compiled_context_length,
"context_reduction_percent": context_reduction_percent,
"baseline_prompt_length": baseline_prompt_length,
"compiled_prompt_length": compiled_prompt_length,
"prompt_reduction_percent": prompt_reduction_percent,
}


def consume_last_info_report() -> InfoReport | None:
global LAST_INFO_REPORT
value = LAST_INFO_REPORT
LAST_INFO_REPORT = None
return value


def build_compiled_system_prompt(state: State) -> str:
focus_value = state[STATE_FACTS][FOCUS_PRIMARY]
prohibit = state[STATE_POLICIES][POLICY_PROHIBIT]
Expand Down
53 changes: 45 additions & 8 deletions demos/run_demo.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,13 @@
import sys
from pathlib import Path

from demos.common import VERBOSE_ENV_VAR, DemoReport, consume_last_report
from demos.common import (
VERBOSE_ENV_VAR,
DemoReport,
InfoReport,
consume_last_info_report,
consume_last_report,
)
from demos.llm_client import MissingDemoConfigError

DEMO_FILES: dict[str, str] = {
Expand All @@ -15,17 +21,24 @@
"3": "03_llm_ambiguity_block.py",
"4": "04_llm_tool_governance.py",
"5": "05_llm_prompt_drift.py",
"6": "06_context_compaction.py",
}

SCORED_DEMOS = {"1", "2", "3", "4", "5"}

def _run(path: Path, *, verbose: bool) -> DemoReport | None:

def _verbose_demo_label(path: Path) -> str:
return path.stem.replace("_llm", "")


def _run(path: Path, *, verbose: bool) -> tuple[DemoReport | None, InfoReport | None]:
if verbose:
print(f"===== Running {path.name} =====")
print(f"===== Running {_verbose_demo_label(path)} =====")
old_value = os.getenv(VERBOSE_ENV_VAR)
os.environ[VERBOSE_ENV_VAR] = "1" if verbose else "0"
try:
runpy.run_path(str(path), run_name="__main__")
return consume_last_report()
return consume_last_report(), consume_last_info_report()
finally:
if old_value is None:
os.environ.pop(VERBOSE_ENV_VAR, None)
Expand Down Expand Up @@ -60,7 +73,7 @@ def main() -> None:
nargs="?",
default="all",
choices=["all", *DEMO_FILES.keys()],
help="Demo number (1-5) or all",
help="Demo number (1-6) or all",
)
parser.add_argument(
"--verbose",
Expand All @@ -74,14 +87,22 @@ def main() -> None:
baseline_fail_count = 0
compiler_pass_count = 0
compiler_fail_count = 0
informational_reports: list[InfoReport] = []
for index, key in enumerate(sorted(DEMO_FILES.keys())):
if index > 0:
if index > 0 and not args.verbose:
print()
try:
result = _run(root / DEMO_FILES[key], verbose=args.verbose)
result, info_report = _run(root / DEMO_FILES[key], verbose=args.verbose)
except MissingDemoConfigError as exc:
_print_config_error(exc)
raise SystemExit(2) from exc

if info_report is not None:
informational_reports.append(info_report)

if key not in SCORED_DEMOS:
continue

if result is None:
baseline_fail_count += 1
compiler_fail_count += 1
Expand All @@ -96,9 +117,25 @@ def main() -> None:
compiler_pass_count += 1
else:
compiler_fail_count += 1
print("=" * 60)
print()
print("Summary:")
print()
print("Evaluative demos:")
print(f"Baseline results: {baseline_pass_count} passed, {baseline_fail_count} failed")
print(f"Compiler results: {compiler_pass_count} passed, {compiler_fail_count} failed")
if informational_reports:
print()
print("Informational demo:")
for report in informational_reports:
demo_id = report["name"].split(" — ", maxsplit=1)[0]
print(
f"{demo_id} — context {report['baseline_context_length']} "
f"→ {report['compiled_context_length']} chars "
f"({report['context_reduction_percent']}% reduction); "
f"prompt {report['baseline_prompt_length']} "
f"→ {report['compiled_prompt_length']} chars "
f"({report['prompt_reduction_percent']}% reduction)"
)
return

try:
Expand Down
Loading