refactor: migrate DirectiveDrafter integrations to 0.2.0dev1

pyproject.toml

@@ -47,12 +47,12 @@ Issues = "https://github.com/rlippmann/context-compiler-example-integrations/iss
 [project.optional-dependencies]
 all = [
   "chromadb",
-  "context-compiler-directive-drafter==0.2.0dev0",
+  "context-compiler-directive-drafter==0.2.0dev1",
   "fastapi",
   "litellm",
 ]
 drafter = [
-  "context-compiler-directive-drafter==0.2.0dev0",
+  "context-compiler-directive-drafter==0.2.0dev1",
 ]
 fastapi = [
   "fastapi",
@@ -67,7 +67,7 @@ retrieval = [
 [dependency-groups]
 dev = [
   "chromadb",
-  "context-compiler-directive-drafter==0.2.0dev0",
+  "context-compiler-directive-drafter==0.2.0dev1",
   "fastapi",
   "httpx2>=2.5.0",
   "httpx>=0.28.1",

python/examples/prompt_construction/litellm/with_directive_drafter.py

@@ -20,8 +20,6 @@
 import re
 from collections.abc import Callable, Mapping, Sequence
 from importlib import import_module
-from importlib.resources import as_file, files
-from importlib.resources.abc import Traversable
 from typing import TypedDict, cast
 
 from context_compiler import (
@@ -35,9 +33,9 @@
 from context_compiler.engine import Engine
 from context_compiler_directive_drafter import (
     DRAFT_OUTCOME_DIRECTIVE,
+    get_converter_prompt,
     parse_preprocessor_output,
     preprocess_heuristic,
-    render_prompt,
 )
 
 try:
@@ -55,8 +53,6 @@
 )
 
 logger = logging.getLogger(__name__)
-
-_PROMPTS_DIR = files("context_compiler_directive_drafter").joinpath("prompts")
 SHOW_CONTEXT_COMPILER_TRACE = False
 
 
@@ -236,18 +232,8 @@ def _call_litellm(messages: list[dict[str, str]]) -> str:
     return content
 
 
-def _prompt_file_path() -> Traversable:
-    profile = os.getenv("PREPROCESSOR_PROMPT_PROFILE", "default").strip().lower()
-    if profile == "llama":
-        return _PROMPTS_DIR.joinpath("llama.txt")
-    return _PROMPTS_DIR.joinpath("default.txt")
-
-
 def _llm_fallback_preprocess(message: str, state: _EngineSnapshot) -> str | None:
-    with as_file(_prompt_file_path()) as prompt_path:
-        prompt = render_prompt(prompt_path, state["premise"], state["policies"])
-    if prompt is None:
-        return None
+    del state
 
     try:
         completion = _get_litellm_completion()
@@ -267,7 +253,7 @@ def _llm_fallback_preprocess(message: str, state: _EngineSnapshot) -> str | None
     kwargs: _LiteLLMCallKwargs = {
         "model": preprocessor_model,
         "messages": [
-            {"role": "system", "content": prompt},
+            {"role": "system", "content": get_converter_prompt()},
             {"role": "user", "content": message},
         ],
         "temperature": 0,

python/reference_integrations/litellm_proxy/context_compiler_precall_hook_with_directive_drafter.py

@@ -15,8 +15,6 @@
 import os
 from collections.abc import Callable, Mapping, Sequence
 from importlib import import_module
-from importlib.resources import as_file, files
-from importlib.resources.abc import Traversable
 from typing import Any, cast
 
 try:
@@ -33,11 +31,11 @@ class CustomLogger:  # type: ignore[no-redef]
     DecisionKind,
     create_engine,
 )
+from context_compiler.grammar import CanonicalDirective
 from context_compiler_directive_drafter import (
-    DRAFT_OUTCOME_DIRECTIVE,
-    parse_preprocessor_output,
-    preprocess_heuristic,
-    render_prompt,
+    DirectiveDrafter,
+    DraftResult,
+    get_converter_prompt,
 )
 from context_compiler_example_integrations.reference_integrations.litellm_proxy._checkpoint_support import (
     MODE_PERSISTENT,
@@ -49,7 +47,6 @@ class CustomLogger:  # type: ignore[no-redef]
     resolve_session_context,
 )
 from context_compiler_example_integrations.reference_integrations.litellm_proxy._litellm_support import (
-    EngineSnapshot,
     extract_request_messages,
     render_compiled_state_contract,
     snapshot_engine_state,
@@ -64,7 +61,6 @@ class CustomLogger:  # type: ignore[no-redef]
     "achat_completion",
 }
 
-_PROMPTS_DIR = files("context_compiler_directive_drafter").joinpath("prompts")
 CHECKPOINT_STORE: CheckpointStore = InMemoryCheckpointStore()
 
 
@@ -91,24 +87,12 @@ def _extract_response_content(response: object) -> str | None:
     return None
 
 
-def _prompt_file_path() -> Traversable:
-    profile = os.getenv("PREPROCESSOR_PROMPT_PROFILE", "default").strip().lower()
-    if profile == "llama":
-        return _PROMPTS_DIR.joinpath("llama.txt")
-    return _PROMPTS_DIR.joinpath("default.txt")
-
-
 def _get_litellm_completion() -> Callable[..., object]:
     litellm_module = import_module("litellm")
     return cast(Callable[..., object], litellm_module.completion)
 
 
-def _llm_fallback_preprocess(message: str, state: EngineSnapshot) -> str | None:
-    with as_file(_prompt_file_path()) as prompt_path:
-        prompt = render_prompt(prompt_path, state["premise"], state["policies"])
-    if prompt is None:
-        return None
-
+def _llm_fallback_candidate(message: str) -> str | None:
     preprocessor_model = os.getenv("PREPROCESSOR_MODEL", "").strip()
     if not preprocessor_model:
         preprocessor_model = os.getenv("MODEL", "").strip()
@@ -127,7 +111,7 @@ def _llm_fallback_preprocess(message: str, state: EngineSnapshot) -> str | None:
     kwargs: dict[str, object] = {
         "model": preprocessor_model,
         "messages": [
-            {"role": "system", "content": prompt},
+            {"role": "system", "content": get_converter_prompt()},
             {"role": "user", "content": message},
         ],
         "api_key": api_key,
@@ -139,39 +123,16 @@ def _llm_fallback_preprocess(message: str, state: EngineSnapshot) -> str | None:
 
     try:
         response = completion(**kwargs)
-        raw_output = _extract_response_content(response)
+        return _extract_response_content(response)
     except Exception:
         return None
 
-    parsed = parse_preprocessor_output(raw_output)
-    if parsed is None:
-        return None
-    return parsed.text
 
-
-def _preprocess_last_user_message(
-    message: str, state: EngineSnapshot | None
-) -> str | None:
-    try:
-        heuristic_result = preprocess_heuristic(message)
-        if (
-            heuristic_result["outcome"] == DRAFT_OUTCOME_DIRECTIVE
-            and heuristic_result["directive"]
-        ):
-            parsed = parse_preprocessor_output(heuristic_result["directive"])
-            if parsed is not None:
-                return parsed.text
-    except Exception:
-        logger.debug("litellm_proxy: heuristic_exception", exc_info=True)
-
-    if state is None:
-        return None
-
-    try:
-        return _llm_fallback_preprocess(message, state)
-    except Exception:
-        logger.debug("litellm_proxy: fallback_exception", exc_info=True)
-        return None
+def _draft_last_user_message(message: str) -> DraftResult:
+    drafter = DirectiveDrafter(
+        fallback=_llm_fallback_candidate, fallback_source="litellm_fallback"
+    )
+    return drafter.draft_directive(message)
 
 
 class ContextCompilerPreCallHookWithPreprocessor(CustomLogger):
@@ -219,15 +180,13 @@ async def async_pre_call_hook(
             "litellm_proxy: latest_user_text_present=%s", latest_user_text is not None
         )
         engine_input = latest_user_text
-        drafted_input: str | None = None
+        drafted_result: DraftResult | None = None
 
         if latest_user_text is not None:
-            drafted_input = _preprocess_last_user_message(
-                latest_user_text, snapshot_engine_state(engine)
-            )
-            logger.debug("litellm_proxy: drafted_input=%r", drafted_input)
-            if drafted_input is not None:
-                engine_input = drafted_input
+            drafted_result = _draft_last_user_message(latest_user_text)
+            logger.debug("litellm_proxy: drafted_result=%r", drafted_result)
+            if isinstance(drafted_result.result, CanonicalDirective):
+                engine_input = drafted_result.result.text
 
         if engine_input is not None:
             decision = engine.step(engine_input)

python/reference_integrations/openwebui_pipe/open_webui_pipe_with_directive_drafter.py

@@ -23,8 +23,6 @@
 import logging
 import re
 from collections.abc import AsyncIterator
-from importlib.resources import as_file, files
-from importlib.resources.abc import Traversable
 from typing import Any, Literal, TypedDict, cast
 
 from fastapi import Request  # type: ignore[import-not-found]
@@ -60,16 +58,15 @@ def Field(*, default: Any, description: str = "") -> Any:  # type: ignore[no-red
 from context_compiler.engine import Engine
 from context_compiler_directive_drafter import (
     DRAFT_OUTCOME_DIRECTIVE,
+    get_converter_prompt,
     parse_preprocessor_output,
     preprocess_heuristic,
-    render_prompt,
 )
 
 logger = logging.getLogger(__name__)
 
 _CC_MARKER = "[[cc_state]]"
 _ENGINES_BY_CHAT_KEY: dict[str, Engine] = {}
-_PROMPTS_DIR = files("context_compiler_directive_drafter").joinpath("prompts")
 
 
 class _EngineSnapshot(TypedDict):
@@ -90,15 +87,6 @@ def _is_directive_shaped_input(message: str) -> bool:
     )
 
 
-def _prompt_file_path(profile: str) -> Traversable:
-    # Runtime prompt selection for fallback drafting:
-    # - default: most instruction-following models
-    # - llama: models that need tighter prompt guidance
-    if profile == "llama":
-        return _PROMPTS_DIR.joinpath("llama.txt")
-    return _PROMPTS_DIR.joinpath("default.txt")
-
-
 def _resolve_chat_key(
     user: dict[str, Any],
     chat_id: str | None,
@@ -708,19 +696,16 @@ async def _llm_fallback_preprocess(
         prompt_profile: str,
         model_id: str | None,
     ) -> tuple[str | None, str | None]:
+        del state, prompt_profile
         model_id = _normalize_model_id(model_id)
         if model_id is None:
             return None, None
-        with as_file(_prompt_file_path(prompt_profile)) as prompt_path:
-            prompt = render_prompt(prompt_path, state["premise"], state["policies"])
-        if prompt is None:
-            return None, None
 
         payload: dict[str, Any] = {
             "model": model_id,
             "stream": False,
             "messages": [
-                {"role": "system", "content": prompt},
+                {"role": "system", "content": get_converter_prompt()},
                 {"role": "user", "content": message},
             ],
         }