Prompt Modular

PromptEngine  ← primary public entry point
    │
    ├── PromptBuilder         ← resolves strategy, coordinates subsystems
    │       ├── ContextManager    ← document processing & token budgeting
    │       ├── GuardrailEngine   ← safety & quality instruction injection
    │       └── Strategy (7)      ← prompt template construction
    │               SimpleStrategy
    │               ChainOfThoughtStrategy
    │               MultiHopStrategy
    │               SelfConsistentStrategy
    │               StepBackStrategy
    │               ReActStrategy
    │               LeastToMostStrategy
    │
    └── PromptOptimizer       ← token reduction post-build

from fennec_community.prompt import PromptEngine, Document

# 1. Create the engine (production defaults)
engine = PromptEngine()

# 2. Build a prompt
prompt = engine.build(
    query     = "What caused the 2008 financial crisis?",
    documents = [
        {"content": "The crisis was triggered by...", "source": "wiki", "score": 0.92},
        {"content": "Subprime mortgage lending...",   "source": "fed_report", "score": 0.87},
    ],
    strategy      = "multi_hop",
    output_format = "json",
)

# 3a. Use with OpenAI
response = openai_client.chat.completions.create(
    model    = "gpt-4o",
    messages = prompt.to_messages(),
)

# 3b. Use with Anthropic
payload  = prompt.to_anthropic()
response = anthropic_client.messages.create(
    **payload,
    model      = "claude-opus-4-20250514",
    max_tokens = 1024,
)

@dataclass
class Document:
    content:   str
    source:    str             = ""
    score:     float           = 1.0
    metadata:  Dict[str, Any]  = field(default_factory=dict)
    chunk_id:  Optional[str]   = None
    language:  str             = "en"

doc = Document(
    content  = "The Federal Reserve raised interest rates...",
    source   = "fed_report_2023.pdf",
    score    = 0.94,
    metadata = {"page": 12, "author": "Federal Reserve"},
    chunk_id = "fed_2023_p12_chunk_3",
)

@dataclass
class Message:
    role:    Literal["system", "user", "assistant"]
    content: str

history = [
    Message(role="user",      content="What is inflation?"),
    Message(role="assistant", content="Inflation is the rate at which..."),
    Message(role="user",      content="And what causes it?"),
]

@dataclass
class PromptRequest:
    # Required
    query: str

    # Context
    documents:  List[Document]   = []
    memory:     List[Message]    = []

    # Intent / routing
    prompt_type:   PromptType      = PromptType.QA
    strategy:      PromptStrategy  = PromptStrategy.SIMPLE
    output_format: OutputFormat    = OutputFormat.TEXT
    complexity:    QueryComplexity = QueryComplexity.SIMPLE
    user_profile:  UserProfile     = UserProfile.GENERAL

    # Constraints
    max_context_tokens: int          = 3000
    max_answer_tokens:  int          = 512
    output_schema:      Optional[Dict] = None
    language:           str          = "en"

    # Feature flags
    enable_guardrails:  bool = True
    enable_cot:         bool = False
    enable_citations:   bool = True
    enable_uncertainty: bool = True

    # Metadata
    session_id: str          = ""
    user_id:    str          = ""
    trace_id:   str          = ""
    extra:      Dict[str, Any] = {}

@dataclass
class BuiltPrompt:
    system_prompt: str
    user_prompt:   str
    messages:      List[Message]

    prompt_type:   PromptType
    strategy:      PromptStrategy
    output_format: OutputFormat

    estimated_tokens:    int
    context_tokens_used: int
    documents_included:  int
    documents_truncated: int

    guardrails_applied:  List[str]
    tokens_saved:        int
    optimization_notes:  List[str]

    session_id: str
    trace_id:   str

PromptEngine(
    context_manager:  Optional[ContextManager]  = None,
    guardrail_engine: Optional[GuardrailEngine] = None,
    extra_guardrails: Optional[List[Guardrail]]  = None,
    enable_cache:       bool = True,
    cache_ttl_sec:      int  = 300,
    max_cache_size:     int  = 256,
    enable_auto_detect: bool = True,
    memory_store:  Optional[Any] = None,
    cache_store:   Optional[Any] = None,
    router:        Optional[Any] = None,
)

# Default — production-ready
engine = PromptEngine()

# Custom — add a domain-specific guardrail, disable cache
from fennec_community.prompt import PromptEngine, Guardrail

medical_guardrail = Guardrail(
    name        = "medical_disclaimer",
    instruction = "Always recommend consulting a licensed physician. "
                  "Do not provide specific medical diagnoses.",
    priority    = 120,
)

engine = PromptEngine(
    extra_guardrails = [medical_guardrail],
    enable_cache     = False,
)

engine.build(
    query:              str,
    documents:          Optional[List[Union[Document, Dict, str]]] = None,
    memory:             Optional[List[Union[Message, Dict]]]       = None,
    prompt_type:        Union[PromptType,    str] = PromptType.QA,
    strategy:           Union[PromptStrategy, str] = PromptStrategy.SIMPLE,
    output_format:      Union[OutputFormat,  str] = OutputFormat.TEXT,
    complexity:         Union[QueryComplexity, str] = QueryComplexity.SIMPLE,
    user_profile:       Union[UserProfile,   str] = UserProfile.GENERAL,
    max_context_tokens: int            = 3000,
    max_answer_tokens:  int            = 512,
    output_schema:      Optional[Dict] = None,
    language:           str            = "en",
    enable_guardrails:  bool           = True,
    enable_citations:   bool           = True,
    enable_uncertainty: bool           = True,
    session_id:         str            = "",
    user_id:            str            = "",
    trace_id:           str            = "",
    extra:              Optional[Dict] = None,
) -> BuiltPrompt

prompt = engine.build(query="What is RAG?")

prompt = engine.build(
    query         = "Compare the economic impacts of COVID-19 in the US vs EU.",
    documents     = retrieved_docs,
    memory        = chat_history,
    prompt_type   = "comparison",
    strategy      = "multi_hop",
    output_format = "markdown",
    complexity    = "complex",
    user_profile  = "executive",
    max_context_tokens = 4000,
    max_answer_tokens  = 1024,
    language      = "en",
    enable_citations   = True,
    session_id    = "sess_abc123",
    trace_id      = "trace_xyz789",
)

schema = {
    "type": "object",
    "properties": {
        "summary":    {"type": "string"},
        "key_points": {"type": "array", "items": {"type": "string"}},
        "sources":    {"type": "array", "items": {"type": "string"}},
    }
}

prompt = engine.build(
    query         = "Summarize the key findings.",
    documents     = docs,
    output_format = "structured",
    output_schema = schema,
)

engine.build_from_request(request: PromptRequest) -> BuiltPrompt

from fennec_community.prompt import PromptRequest, PromptType, PromptStrategy

request = PromptRequest(
    query        = "What are the side effects of ibuprofen?",
    documents    = my_docs,
    prompt_type  = PromptType.QA,
    strategy     = PromptStrategy.CHAIN_OF_THOUGHT,
    user_profile = UserProfile.TECHNICAL,
)

prompt = engine.build_from_request(request)

engine.record_feedback(
    trace_id:      str,
    quality_score: float,
    notes:         str = "",
) -> None

# After the LLM response is reviewed
engine.record_feedback(
    trace_id      = prompt.trace_id,
    quality_score = 0.85,
    notes         = "Good answer but missed one key point.",
)

engine.adaptive_strategy_for(prompt_type: PromptType) -> Optional[PromptStrategy]

best = engine.adaptive_strategy_for(PromptType.QA)
if best:
    prompt = engine.build(query=user_query, strategy=best)
else:
    prompt = engine.build(query=user_query)  # use defaults

engine.metrics -> Dict[str, Any]

import json
print(json.dumps(engine.metrics, indent=2))
# {
#   "total_builds": 142,
#   "total_tokens": 284000,
#   "total_tokens_saved": 12400,
#   "cache_hits": 38,
#   "avg_build_ms": 4.72,
#   "cache_hit_rate_pct": 26.8,
#   "builds_by_type": {"qa": 90, "reasoning": 42, "summarization": 10},
#   "builds_by_strategy": {"simple": 60, "cot": 52, "multi_hop": 30}
# }

engine.get_trace_log(last_n: int = 20) -> List[Dict[str, Any]]

traces = engine.get_trace_log(last_n=5)
for t in traces:
    print(f"{t['trace_id']} | {t['prompt_type']} | {t['elapsed_ms']}ms | tokens={t['estimated_tokens']}")

engine.reset_metrics() -> None

# Reset every hour in a scheduled job
engine.reset_metrics()

engine.on(event: str, callback: Callable) -> None

def log_to_datadog(prompt: BuiltPrompt, request: PromptRequest):
    datadog.metric("prompt.tokens", prompt.estimated_tokens, tags=[
        f"type:{prompt.prompt_type.value}",
        f"strategy:{prompt.strategy.value}",
    ])

engine.on("prompt.built", log_to_datadog)

ContextManager(
    dedup_threshold:     float = 0.85,
    min_doc_tokens:      int   = 5,
    use_lost_in_middle:  bool  = True,
    summarize_overflow:  bool  = False,
    max_memory_messages: int   = 10,
)

context_manager.build(request: PromptRequest) -> ContextResult

from fennec_community.prompt import ContextManager, Document, PromptRequest

cm = ContextManager(dedup_threshold=0.80, use_lost_in_middle=True)

request = PromptRequest(query="What is inflation?", documents=my_docs)
result  = cm.build(request)

print(f"Context utilization: {result.utilization_pct}%")
print(f"Documents included:  {len(result.included_docs)}")
print(f"Duplicates removed:  {result.duplicates_removed}")
print(result.context_block)

context_manager.format_memory(
    memory:    List[Message],
    max_turns: Optional[int] = None,
) -> str

User: What is inflation?
Assistant: Inflation is the rate at which...
User: And what causes it?

memory_block = cm.format_memory(memory=chat_history, max_turns=5)

GuardrailEngine(extra_guardrails: Optional[List[Guardrail]] = None)

guardrail_engine.build(request: PromptRequest) -> tuple[str, List[str]]

from fennec_community.prompt import Guardrail, GuardrailLibrary

legal_guardrail = Guardrail(
    name        = "legal_disclaimer",
    instruction = "This is not legal advice. Always recommend consulting a qualified attorney.",
    priority    = 115,  # higher than safe_output, applied first
)

engine = PromptEngine(extra_guardrails=[legal_guardrail])

PromptOptimizer(
    max_total_tokens:  int  = 6000,
    enable_filler:     bool = True,
    enable_dedup:      bool = True,
    enable_whitespace: bool = True,
)

optimizer.optimize(
    system:  str,
    user:    str,
    request: Optional[object] = None,
) -> Tuple[str, str, int, List[str]]

from fennec_community.prompt import PromptOptimizer

optimizer = PromptOptimizer(max_total_tokens=4000)

system_opt, user_opt, saved, notes = optimizer.optimize(
    system = my_system_prompt,
    user   = my_user_prompt,
)

print(f"Saved {saved} tokens via: {notes}")

built_prompt.to_messages() -> List[Dict[str, str]]

response = openai_client.chat.completions.create(
    model    = "gpt-4o",
    messages = prompt.to_messages(),
)

built_prompt.to_anthropic() -> Dict[str, Any]

payload = prompt.to_anthropic()
response = anthropic_client.messages.create(
    **payload,
    model      = "claude-opus-4-20250514",
    max_tokens = 1024,
)

built_prompt.full_text -> str

print(prompt.full_text)
# [SYSTEM]
# You are an expert AI assistant...
#
# [USER]
# ## Context
# --- Source [1] wiki (relevance: 0.92) ---
# ...

context_result.utilization_pct -> float

result = cm.build(request)
print(f"Token budget utilization: {result.utilization_pct}%")
# Token budget utilization: 84.3%

get_strategy(strategy: PromptStrategy) -> BaseStrategy

from fennec_community.prompt import get_strategy, PromptStrategy

impl = get_strategy(PromptStrategy.CHAIN_OF_THOUGHT)

STRATEGY_REGISTRY: Dict[PromptStrategy, BaseStrategy]

from fennec_community.prompt import STRATEGY_REGISTRY, PromptStrategy
from fennec_community.prompt.strategies import BaseStrategy

class MyCustomStrategy(BaseStrategy):
    STRATEGY = PromptStrategy.SIMPLE  # override an existing slot

    def _build_system(self, req, guardrail_block): ...
    def _build_user(self, req, context_block, memory_block): ...

STRATEGY_REGISTRY[PromptStrategy.SIMPLE] = MyCustomStrategy()

metrics_obj.to_dict() -> Dict[str, Any]

import json

# Access via engine property (recommended)
print(json.dumps(engine.metrics, indent=2))

from fennec_community.prompt import PromptEngine
import openai

engine = PromptEngine()
client = openai.OpenAI(api_key="...")

prompt = engine.build(
    query     = "What is the capital of France?",
    documents = [{"content": "France is a country in Europe. Its capital is Paris.", "source": "geo_db"}],
)

response = client.chat.completions.create(
    model    = "gpt-4o",
    messages = prompt.to_messages(),
)
print(response.choices[0].message.content)

from fennec_community.prompt import PromptEngine
import anthropic

engine = PromptEngine()
client = anthropic.Anthropic(api_key="...")

prompt = engine.build(
    query         = "Summarize the quarterly earnings report.",
    documents     = retrieved_docs,
    strategy      = "chain_of_thought",
    output_format = "markdown",
    user_profile  = "executive",
)

response = client.messages.create(
    **prompt.to_anthropic(),
    model      = "claude-opus-4-20250514",
    max_tokens = 2048,
)
print(response.content[0].text)

from fennec_community.prompt import PromptEngine, Message

engine  = PromptEngine()
history = []

def chat(user_message: str, docs=None) -> str:
    prompt = engine.build(
        query     = user_message,
        documents = docs or [],
        memory    = history,
        prompt_type = "conversational",
    )

    # Call your LLM here...
    answer = llm_call(prompt.to_messages())

    # Update history
    history.append(Message(role="user",      content=user_message))
    history.append(Message(role="assistant", content=answer))

    return answer

tools = [
    {
        "name":        "search_database",
        "description": "Search the company knowledge base.",
        "parameters":  {"query": "string", "top_k": "int"},
    },
    {
        "name":        "get_document",
        "description": "Retrieve a specific document by ID.",
        "parameters":  {"doc_id": "string"},
    },
]

prompt = engine.build(
    query       = "Find all invoices from Q3 2024 and calculate the total.",
    prompt_type = "agent",
    strategy    = "react",
    extra       = {"tools": tools},
)

from fennec_community.prompt import PromptEngine, Guardrail, BuiltPrompt, PromptRequest

# Custom guardrail
disclaimer = Guardrail(
    name        = "financial_disclaimer",
    instruction = "This is not financial advice. Past performance is not indicative of future results.",
    priority    = 115,
)

engine = PromptEngine(extra_guardrails=[disclaimer])

# Hook into every build for custom logging
def on_prompt_built(prompt: BuiltPrompt, request: PromptRequest):
    print(f"[{request.trace_id}] Built {prompt.prompt_type.value} | "
          f"{prompt.estimated_tokens} tokens | {prompt.tokens_saved} saved | "
          f"guardrails={prompt.guardrails_applied}")

engine.on("prompt.built", on_prompt_built)

# After collecting feedback over time:
engine.record_feedback(trace_id="abc", quality_score=0.9)
engine.record_feedback(trace_id="def", quality_score=0.6)
# ... at least 5 feedback entries ...

best_strategy = engine.adaptive_strategy_for(PromptType.QA)

prompt = engine.build(
    query    = "What is the return policy?",
    strategy = best_strategy or PromptStrategy.SIMPLE,
)