From fafc4bd4b6413807f6160cb9e6d80618e571fb5d Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Mon, 9 Mar 2026 01:56:18 -0400 Subject: [PATCH] Add deterministic examples and canonical JSON output --- README.md | 7 + examples/01_persistent_guardrails.py | 44 ++++ examples/02_configuration_and_correction.py | 28 +++ examples/03_ambiguity_with_clarification.py | 38 +++ examples/04_tool_governance_denylist.py | 46 ++++ examples/05_llm_integration_pattern.py | 43 ++++ examples/_util.py | 10 + tests/test_examples.py | 248 ++++++++++++++++++++ 8 files changed, 464 insertions(+) create mode 100644 examples/01_persistent_guardrails.py create mode 100644 examples/02_configuration_and_correction.py create mode 100644 examples/03_ambiguity_with_clarification.py create mode 100644 examples/04_tool_governance_denylist.py create mode 100644 examples/05_llm_integration_pattern.py create mode 100644 examples/_util.py create mode 100644 tests/test_examples.py diff --git a/README.md b/README.md index 3a7a803..c0b4b27 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,9 @@ # context-compiler Deterministic state engine for managing authoritative conversational constraints in LLM applications. + +## Examples +- [Persistent guardrails](examples/01_persistent_guardrails.py) +- [Configuration with correction](examples/02_configuration_and_correction.py) +- [Ambiguity detection with clarification](examples/03_ambiguity_with_clarification.py) +- [Tool governance for agents](examples/04_tool_governance_denylist.py) +- [LLM integration pattern](examples/05_llm_integration_pattern.py) diff --git a/examples/01_persistent_guardrails.py b/examples/01_persistent_guardrails.py new file mode 100644 index 0000000..d0302aa --- /dev/null +++ b/examples/01_persistent_guardrails.py @@ -0,0 +1,44 @@ +"""Example 1: persistent guardrails across turns.""" + +from _util import print_json + +from context_compiler import State, create_engine + + +def build_prompt(state: State, user_input: str) -> str: + prohibit = state["policies"]["prohibit"] + prohibit_text = ", ".join(prohibit) if prohibit else "(none)" + return ( + "System: Follow authoritative conversation state.\n" + "Compiled context:\n" + f"- policies.prohibit: {prohibit_text}\n" + f"User: {user_input}" + ) + + +def main() -> None: + engine = create_engine() + + print("User: don't use docker") + decision1 = engine.step("don't use docker") + print("Decision:") + print_json(decision1) + print("State after turn 1:") + print_json(engine.state) + print() + + print("User: how should I deploy my service?") + decision2 = engine.step("how should I deploy my service?") + print("Decision:") + print_json(decision2) + print("State after turn 2:") + print_json(engine.state) + print() + + print("Host prompt construction with persisted policy:") + prompt = build_prompt(engine.state, "how should I deploy my service?") + print(prompt) + + +if __name__ == "__main__": + main() diff --git a/examples/02_configuration_and_correction.py b/examples/02_configuration_and_correction.py new file mode 100644 index 0000000..e060924 --- /dev/null +++ b/examples/02_configuration_and_correction.py @@ -0,0 +1,28 @@ +"""Example 2: conversational configuration with correction (last-write-wins).""" + +from _util import print_json + +from context_compiler import create_engine + + +def main() -> None: + engine = create_engine() + + print("User: I'm using MacBook M3") + decision1 = engine.step("I'm using MacBook M3") + print("Decision:") + print_json(decision1) + print("State:") + print_json(engine.state) + print() + + print("User: actually MacBook M2") + decision2 = engine.step("actually MacBook M2") + print("Decision:") + print_json(decision2) + print("State (last write wins):") + print_json(engine.state) + + +if __name__ == "__main__": + main() diff --git a/examples/03_ambiguity_with_clarification.py b/examples/03_ambiguity_with_clarification.py new file mode 100644 index 0000000..6e60457 --- /dev/null +++ b/examples/03_ambiguity_with_clarification.py @@ -0,0 +1,38 @@ +"""Example 3: ambiguous directive flow with clarification handling.""" + +from _util import print_json + +from context_compiler import create_engine + + +def fake_llm(user_input: str) -> str: + print(f"LLM would be called with user_input={user_input!r}") + return "[example LLM response]" + + +def main() -> None: + engine = create_engine() + + print("User: no use docker") + decision1 = engine.step("no use docker") + print("Decision:") + print_json(decision1) + print() + + if decision1["kind"] == "clarify": + print("Host behavior: clarification pending, do NOT call LLM.") + print(f"Prompt to user: {decision1['prompt_to_user']}") + else: + fake_llm("no use docker") + print() + + print("User: yes") + decision2 = engine.step("yes") + print("Decision:") + print_json(decision2) + print("State after clarification acceptance:") + print_json(engine.state) + + +if __name__ == "__main__": + main() diff --git a/examples/04_tool_governance_denylist.py b/examples/04_tool_governance_denylist.py new file mode 100644 index 0000000..94b9967 --- /dev/null +++ b/examples/04_tool_governance_denylist.py @@ -0,0 +1,46 @@ +"""Example 4: host-side tool governance using policies.prohibit denylist.""" + +from dataclasses import dataclass + +from _util import print_json + +from context_compiler import create_engine + + +@dataclass +class Tool: + name: str + + +def block_tool(tool: Tool) -> None: + print(f"Blocked tool: {tool.name}") + + +def allow_tool(tool: Tool) -> None: + print(f"Allowed tool: {tool.name}") + + +def main() -> None: + engine = create_engine() + + user_input = "don't use docker" + print(f"User: {user_input}") + decision = engine.step(user_input) + print("Decision:") + print_json(decision) + print("State after turn:") + state = engine.state + print_json(state) + print() + + print("Host-side tool denylist behavior:") + tools = [Tool("docker"), Tool("kubectl")] + for tool in tools: + if tool.name in state["policies"]["prohibit"]: + block_tool(tool) + else: + allow_tool(tool) + + +if __name__ == "__main__": + main() diff --git a/examples/05_llm_integration_pattern.py b/examples/05_llm_integration_pattern.py new file mode 100644 index 0000000..641fffb --- /dev/null +++ b/examples/05_llm_integration_pattern.py @@ -0,0 +1,43 @@ +"""Example 5: host integration pattern using Decision API.""" + +from _util import canonical_json, print_json + +from context_compiler import Engine, State, create_engine + + +def fake_llm(state: State | None, user_input: str) -> str: + print("LLM would be called with:") + print(f"state: {canonical_json(state)}") + print("user_input:", user_input) + return "[example LLM response]" + + +def handle_turn(engine_input: str, engine: Engine) -> None: + decision = engine.step(engine_input) + print(f"User: {engine_input}") + print("Decision:") + print_json(decision) + + if decision["kind"] == "passthrough": + print("Host action: passthrough -> call fake_llm() without state") + fake_llm(None, engine_input) + elif decision["kind"] == "update": + print("Host action: update -> call fake_llm() with compiled state") + fake_llm(decision["state"], engine_input) + elif decision["kind"] == "clarify": + print("Host action: clarify -> show prompt, DO NOT call LLM") + print("prompt_to_user:", decision["prompt_to_user"]) + print() + + +def main() -> None: + engine = create_engine() + + handle_turn("hello there", engine) + handle_turn("don't use docker", engine) + handle_turn("no use kubernetes", engine) + handle_turn("yes", engine) + + +if __name__ == "__main__": + main() diff --git a/examples/_util.py b/examples/_util.py new file mode 100644 index 0000000..cf4751f --- /dev/null +++ b/examples/_util.py @@ -0,0 +1,10 @@ +import json +from typing import Any + + +def canonical_json(obj: Any) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":")) + + +def print_json(obj: Any) -> None: + print(canonical_json(obj)) diff --git a/tests/test_examples.py b/tests/test_examples.py new file mode 100644 index 0000000..ba0542d --- /dev/null +++ b/tests/test_examples.py @@ -0,0 +1,248 @@ +import json +import subprocess +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +EXAMPLES_DIR = REPO_ROOT / "examples" + + +def _run_example(script_name: str) -> str: + script = EXAMPLES_DIR / script_name + completed = subprocess.run( + [sys.executable, str(script)], + check=True, + capture_output=True, + text=True, + cwd=REPO_ROOT, + ) + return completed.stdout + + +def _canonical_json(obj: object) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":")) + + +def test_persistent_guardrails_example_output() -> None: + output = _run_example("01_persistent_guardrails.py") + + assert "User: don't use docker" in output + assert ( + _canonical_json( + { + "kind": "update", + "prompt_to_user": None, + "state": { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker"]}, + "version": 1, + }, + } + ) + in output + ) + assert _canonical_json({"kind": "passthrough", "prompt_to_user": None, "state": None}) in output + assert ( + _canonical_json( + { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker"]}, + "version": 1, + } + ) + in output + ) + assert "Compiled context:" in output + assert "- policies.prohibit: docker" in output + assert "User: how should I deploy my service?" in output + + +def test_configuration_and_correction_example_output() -> None: + output = _run_example("02_configuration_and_correction.py") + + assert "User: I'm using MacBook M3" in output + assert ( + _canonical_json( + { + "kind": "update", + "prompt_to_user": None, + "state": { + "facts": {"focus.device": "MacBook M3"}, + "policies": {"prohibit": []}, + "version": 1, + }, + } + ) + in output + ) + assert ( + _canonical_json( + { + "facts": {"focus.device": "MacBook M3"}, + "policies": {"prohibit": []}, + "version": 1, + } + ) + in output + ) + assert "User: actually MacBook M2" in output + assert ( + _canonical_json( + { + "kind": "update", + "prompt_to_user": None, + "state": { + "facts": {"focus.device": "MacBook M2"}, + "policies": {"prohibit": []}, + "version": 1, + }, + } + ) + in output + ) + assert ( + _canonical_json( + { + "facts": {"focus.device": "MacBook M2"}, + "policies": {"prohibit": []}, + "version": 1, + } + ) + in output + ) + assert "State (last write wins):" in output + + +def test_ambiguity_with_clarification_example_output() -> None: + output = _run_example("03_ambiguity_with_clarification.py") + + assert "User: no use docker" in output + assert ( + _canonical_json( + { + "kind": "clarify", + "prompt_to_user": "Did you mean to prohibit 'docker'?", + "state": None, + } + ) + in output + ) + assert "do NOT call LLM" in output + assert "Prompt to user: Did you mean to prohibit 'docker'?" in output + assert "User: yes" in output + assert ( + _canonical_json( + { + "kind": "update", + "prompt_to_user": None, + "state": { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker"]}, + "version": 1, + }, + } + ) + in output + ) + assert ( + _canonical_json( + { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker"]}, + "version": 1, + } + ) + in output + ) + + +def test_tool_governance_denylist_example_output() -> None: + output = _run_example("04_tool_governance_denylist.py") + + assert "User: don't use docker" in output + assert "Decision:" in output + assert ( + _canonical_json( + { + "kind": "update", + "prompt_to_user": None, + "state": { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker"]}, + "version": 1, + }, + } + ) + in output + ) + assert "State after turn:" in output + assert ( + _canonical_json( + { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker"]}, + "version": 1, + } + ) + in output + ) + assert "Host-side tool denylist behavior:" in output + assert "Blocked tool: docker" in output + assert "Allowed tool: kubectl" in output + + +def test_llm_integration_pattern_example_output() -> None: + output = _run_example("05_llm_integration_pattern.py") + docker_state = { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker"]}, + "version": 1, + } + docker_kubernetes_state = { + "facts": {"focus.device": None}, + "policies": {"prohibit": ["docker", "kubernetes"]}, + "version": 1, + } + + assert "User: hello there" in output + assert _canonical_json({"kind": "passthrough", "prompt_to_user": None, "state": None}) in output + assert "Host action: passthrough -> call fake_llm() without state" in output + assert "state: null" in output + assert "User: don't use docker" in output + assert ( + _canonical_json( + { + "kind": "update", + "prompt_to_user": None, + "state": docker_state, + } + ) + in output + ) + assert "Host action: update -> call fake_llm() with compiled state" in output + assert f"state: {_canonical_json(docker_state)}" in output + assert "User: no use kubernetes" in output + assert ( + _canonical_json( + { + "kind": "clarify", + "prompt_to_user": "Did you mean to prohibit 'kubernetes'?", + "state": None, + } + ) + in output + ) + assert "Host action: clarify -> show prompt, DO NOT call LLM" in output + assert "User: yes" in output + assert "prompt_to_user: Did you mean to prohibit 'kubernetes'?" in output + assert ( + _canonical_json( + { + "kind": "update", + "prompt_to_user": None, + "state": docker_kubernetes_state, + } + ) + in output + ) + assert f"state: {_canonical_json(docker_kubernetes_state)}" in output