Design Patterns for SLM-First Systems

from langchain_openai import ChatOpenAIfrom langchain_ollama import OllamaLLMfrom langchain_core.messages import AIMessage, HumanMessage, SystemMessagefrom langgraph.graph import StateGraph, ENDfrom langgraph.checkpoint.memory import MemorySaverfrom typing import TypedDict, Literal, Annotatedimport operatorimport jsonimport os# Frontier LLM - used only for reasoning steps# Set LLM_MODEL to your provider's current frontier model stringllm = ChatOpenAI(model=os.environ["LLM_MODEL"], temperature=0)# SLM - used for routing, validation, classification# Run locally via Ollama, or swap for any small model API.# OllamaLLM returns a plain string, so every node below parses it with json.loadsslm = OllamaLLM(model="phi4-mini", temperature=0)# Shared state schema - every graph in this article uses this baseclass AgentState(TypedDict):    messages: Annotated[list, operator.add]    route: str    confidence: float    intent: dict    loop_count: int    last_action_result: str    escalate: bool

# Don't do thisdef route_with_llm(state: AgentState) -> AgentState:    response = llm.invoke(        f"Classify this request: {state['messages'][-1].content}\n"        "Return one of: simple_lookup, tool_call, complex_reasoning"    )    state["route"] = response.content.strip()    return state

ROUTER_PROMPT = """You are a request router. Classify the request into exactly one category.Categories:- simple_lookup: factual questions, status checks, definitions- tool_call: actions requiring an API or database call- complex_reasoning: multi-step analysis, synthesis, judgment calls- clarification_needed: ambiguous or incomplete requestsRequest: {request}Respond with JSON only:{{"route": "<category>", "confidence": <0.0-1.0>, "reason": "<one sentence>"}}"""def slm_router(state: AgentState) -> AgentState:    request = state["messages"][-1].content    prompt = ROUTER_PROMPT.format(request=request)    raw = slm.invoke(prompt)    try:        parsed = json.loads(raw.strip())        state["route"] = parsed["route"]        state["confidence"] = parsed["confidence"]    except (json.JSONDecodeError, KeyError):        # Parsing failure -> safe default is escalation, not silent wrong routing        state["route"] = "complex_reasoning"        state["confidence"] = 0.0    return statedef route_decision(state: AgentState) -> str:    # Low confidence escalates to LLM regardless of route label    if state["confidence"] < 0.7:        return "llm_handler"    route_map = {        "simple_lookup": "direct_response",        "tool_call": "tool_executor",        "complex_reasoning": "llm_handler",        "clarification_needed": "clarification_handler",    }    return route_map.get(state["route"], "llm_handler")

def build_routing_graph():    graph = StateGraph(AgentState)    graph.add_node("router", slm_router)    graph.add_node("direct_response", direct_response_node)    graph.add_node("tool_executor", tool_executor_node)    graph.add_node("llm_handler", llm_handler_node)    graph.add_node("clarification_handler", clarification_handler_node)    graph.set_entry_point("router")    graph.add_conditional_edges(        "router",        route_decision,        {            "direct_response": "direct_response",            "tool_executor": "tool_executor",            "llm_handler": "llm_handler",            "clarification_handler": "clarification_handler",        }    )    graph.add_edge("direct_response", END)    graph.add_edge("tool_executor", END)    graph.add_edge("llm_handler", END)    graph.add_edge("clarification_handler", END)    return graph.compile(checkpointer=MemorySaver())

SCHEMA_ENFORCER_PROMPT = """Extract structured data from the user request.Target schema:{schema}User request: {request}Rules:- Return valid JSON matching the schema exactly- Use null for missing optional fields- If a required field cannot be determined, set "extraction_failed": true- Do not invent values - only extract what is explicitly statedRespond with JSON only."""def make_schema_enforcer(schema: dict, schema_name: str):    """Factory that returns a schema enforcer node for a specific schema."""    def schema_enforcer_node(state: AgentState) -> AgentState:        request = state["messages"][-1].content        prompt = SCHEMA_ENFORCER_PROMPT.format(            schema=json.dumps(schema, indent=2),            request=request        )        raw = slm.invoke(prompt)        try:            extracted = json.loads(raw.strip())        except json.JSONDecodeError:            state["intent"] = {"extraction_failed": True, "raw": raw}            state["escalate"] = True            return state        if extracted.get("extraction_failed"):            state["intent"] = extracted            state["escalate"] = True        else:            state["intent"] = extracted            state["escalate"] = False        return state    schema_enforcer_node.__name__ = f"schema_enforcer_{schema_name}"    return schema_enforcer_node# Example: calendar event schemaCALENDAR_SCHEMA = {    "intent": "string (create_event | update_event | delete_event)",    "title": "string (required)",    "start_time": "ISO 8601 datetime (required)",    "end_time": "ISO 8601 datetime (optional)",    "attendees": "array of email strings (optional)",    "location": "string (optional)"}calendar_enforcer = make_schema_enforcer(CALENDAR_SCHEMA, "calendar")

def validate_calendar_intent(state: AgentState) -> AgentState:    intent = state.get("intent", {})    errors = []    # Required field check    if not intent.get("title"):        errors.append("title is required")    if not intent.get("start_time"):        errors.append("start_time is required")    # Type/format validation    if intent.get("start_time"):        try:            from datetime import datetime            datetime.fromisoformat(intent["start_time"])        except ValueError:            errors.append(f"start_time is not valid ISO 8601: {intent['start_time']}")    # Business rule validation    if intent.get("start_time") and intent.get("end_time"):        from datetime import datetime        start = datetime.fromisoformat(intent["start_time"])        end = datetime.fromisoformat(intent["end_time"])        if end <= start:            errors.append("end_time must be after start_time")    if errors:        state["last_action_result"] = f"Validation failed: {'; '.join(errors)}"        state["escalate"] = True    else:        state["escalate"] = False    return state

OUTPUT_VALIDATOR_PROMPT = """You are an output safety validator. Review the AI response below.Check for:1. PII exposure (names + contact info combined, SSNs, credit card numbers, passwords)2. Schema compliance: does the response match the expected format? Expected: {expected_format}3. Hallucination signals: confident specific claims without basis (exact numbers, dates, names not in context)4. Toxicity or policy violationsResponse to validate:{response}Return JSON:{{"pass": true/false, "issues": ["issue1", "issue2"], "severity": "low|medium|high|critical"}}If pass is false and severity is critical, the response must be blocked entirely."""def output_validator(state: AgentState, expected_format: str = "natural language") -> AgentState:    last_message = state["messages"][-1].content    prompt = OUTPUT_VALIDATOR_PROMPT.format(        response=last_message,        expected_format=expected_format    )    raw = slm.invoke(prompt)    try:        result = json.loads(raw.strip())    except json.JSONDecodeError:        # Validator itself failed - safe default is to flag for review        state["last_action_result"] = "validator_error"        state["escalate"] = True        return state    if not result.get("pass", True):        severity = result.get("severity", "medium")        if severity == "critical":            # Block the response entirely, return safe error message            state["messages"].append(AIMessage(                content="I encountered an issue generating a safe response. Please try rephrasing your request."            ))            state["escalate"] = True        elif severity in ("high", "medium"):            # Flag for human review, still return response            state["last_action_result"] = f"flagged:{severity}:{','.join(result.get('issues', []))}"    return state

LOOP_CONTROLLER_PROMPT = """You are a loop controller for an agentic system. Evaluate the current state.Task: {task}Loop iteration: {loop_count} of {max_loops}Last action result: {last_result}Conversation so far: {history}Decide the next action:- CONTINUE: task not complete, last action succeeded, safe to proceed- COMPLETE: task is finished, return result to user- REPLAN: task not complete but current approach is failing, need new strategy- ESCALATE: situation requires human judgment or LLM reasoning- ABORT: task cannot be completed, too many failures, or approaching loop limitRules:- If loop_count >= {max_loops} * 0.8, prefer COMPLETE or ABORT over CONTINUE- If last action failed twice in a row, prefer REPLAN or ESCALATE- Never CONTINUE if last_result contains an unhandled errorReturn JSON:{{"decision": "<CONTINUE|COMPLETE|REPLAN|ESCALATE|ABORT>",  "reason": "<one sentence>",  "confidence": <0.0-1.0>}}"""MAX_LOOPS = 10ESCALATE_THRESHOLD = 0.65def loop_controller(state: AgentState) -> AgentState:    loop_count = state.get("loop_count", 0) + 1    state["loop_count"] = loop_count    # Hard circuit breaker - never trust the SLM to self-limit    if loop_count >= MAX_LOOPS:        state["route"] = "ABORT"        return state    task = state["messages"][0].content    last_result = state.get("last_action_result", "no action yet")    history = "\n".join([        f"{m.type}: {m.content[:200]}"        for m in state["messages"][-6:]  # Last 3 turns    ])    prompt = LOOP_CONTROLLER_PROMPT.format(        task=task,        loop_count=loop_count,        max_loops=MAX_LOOPS,        last_result=last_result,        history=history    )    raw = slm.invoke(prompt)    try:        result = json.loads(raw.strip())        decision = result["decision"]        confidence = result.get("confidence", 0.5)    except (json.JSONDecodeError, KeyError):        # Controller parse failure - safe default is ESCALATE        state["route"] = "ESCALATE"        return state    # Low confidence decisions escalate regardless of label    if confidence < ESCALATE_THRESHOLD and decision not in ("COMPLETE", "ABORT"):        decision = "ESCALATE"    state["route"] = decision    return statedef loop_decision(state: AgentState) -> str:    decision_map = {        "CONTINUE": "action_executor",        "COMPLETE": "response_formatter",        "REPLAN": "llm_planner",        "ESCALATE": "llm_handler",        "ABORT": "abort_handler",    }    return decision_map.get(state["route"], "llm_handler")

def build_agentic_loop_graph():    graph = StateGraph(AgentState)    # Nodes    graph.add_node("loop_controller", loop_controller)    graph.add_node("action_executor", action_executor_node)    graph.add_node("response_formatter", response_formatter_node)    graph.add_node("llm_planner", llm_planner_node)    graph.add_node("llm_handler", llm_handler_node)    graph.add_node("abort_handler", abort_handler_node)    # Entry    graph.set_entry_point("loop_controller")    # Loop controller routes    graph.add_conditional_edges(        "loop_controller",        loop_decision,        {            "action_executor": "action_executor",            "response_formatter": "response_formatter",            "llm_planner": "llm_planner",            "llm_handler": "llm_handler",            "abort_handler": "abort_handler",        }    )    # Action executor feeds back into the loop controller    graph.add_edge("action_executor", "loop_controller")    # LLM planner feeds back into the loop controller with updated plan    graph.add_edge("llm_planner", "loop_controller")    # Terminal nodes    graph.add_edge("response_formatter", END)    graph.add_edge("llm_handler", END)    graph.add_edge("abort_handler", END)    return graph.compile(checkpointer=MemorySaver())

def action_executor_node(state: AgentState) -> AgentState:    """    Executes the next action in the agentic loop.    Replace this stub with your tool dispatch logic.    Contract:    - Read the current plan from state["messages"]    - Execute one action (one tool call, one API call, one write operation)    - Write the result to state["last_action_result"]    - Append result as AIMessage to state["messages"]    - Never execute more than one action per call - the loop controller decides whether to continue    """    # Example: extract action from last LLM plan message    plan = state["messages"][-1].content    # Replace with your actual tool dispatch    # e.g. result = your_tool_registry.execute(plan)    result = f"Executed action from plan: {plan[:100]}"    state["last_action_result"] = result    state["messages"] = [AIMessage(content=result)]    return state

from langgraph.checkpoint.memory import MemorySaverfrom langgraph.types import interruptdef human_approval_node(state: AgentState) -> AgentState:    """    Pauses the graph and surfaces the pending action for human review.    The graph resumes when the caller invokes    graph.invoke(Command(resume=...)) with the same thread_id.    On resume this node runs again from the top, and interrupt()    returns the resume value instead of pausing.    """    pending_action = state.get("intent", {})    blocked_reason = state.get("last_action_result", "requires approval")    # interrupt() pauses execution here and returns control to the caller.    # The payload below is what the caller sees in result["__interrupt__"].    # Nothing above this line may have side effects: it re-runs on resume.    human_decision = interrupt({        "pending_action": pending_action,        "reason": blocked_reason,        "message": "This action requires your approval before execution."    })    # human_decision is whatever the caller passes in Command(resume=...)    if human_decision.get("approved"):        state["escalate"] = False        state["last_action_result"] = "approved_by_human"    else:        state["escalate"] = True        state["last_action_result"] = f"rejected_by_human:{human_decision.get('reason', '')}"    return statedef build_agentic_loop_with_approval():    graph = StateGraph(AgentState)    graph.add_node("loop_controller", loop_controller)    graph.add_node("action_executor", action_executor_node)    graph.add_node("consequence_check", consequence_check_node)    graph.add_node("human_approval", human_approval_node)    graph.add_node("response_formatter", response_formatter_node)    graph.add_node("abort_handler", abort_handler_node)    graph.add_node("llm_handler", llm_handler_node)    graph.set_entry_point("loop_controller")    graph.add_conditional_edges(        "loop_controller",        loop_decision,        {            "action_executor": "consequence_check",  # Check before every execution            "response_formatter": "response_formatter",            "llm_planner": "llm_handler",            "llm_handler": "llm_handler",            "abort_handler": "abort_handler",        }    )    # Consequence check gates every action    graph.add_conditional_edges(        "consequence_check",        lambda s: "human_approval" if s.get("escalate") else "action_executor",        {            "human_approval": "human_approval",            "action_executor": "action_executor",        }    )    # After human decision, either execute or return to loop    graph.add_conditional_edges(        "human_approval",        lambda s: "action_executor" if not s.get("escalate") else "loop_controller",        {            "action_executor": "action_executor",            "loop_controller": "loop_controller",        }    )    graph.add_edge("action_executor", "loop_controller")    graph.add_edge("response_formatter", END)    graph.add_edge("llm_handler", END)    graph.add_edge("abort_handler", END)    # The interrupt() call inside human_approval_node does the pausing.    # A checkpointer is required: it is where the paused state lives.    return graph.compile(checkpointer=MemorySaver())

from langgraph.types import Commandconfig = {"configurable": {"thread_id": "task-001"}}graph = build_agentic_loop_with_approval()# Start the task - runs until interrupt() fires inside human_approvalresult = graph.invoke(initial_state, config=config)# Graph is now paused. The interrupt payload comes back in the result:pending = result["__interrupt__"][0].valueprint(f"Pending action requiring approval: {pending['pending_action']}")# While paused, graph.get_state(config).next == ("human_approval",)# Human approves - the dict becomes interrupt()'s return valueresult = graph.invoke(Command(resume={"approved": True}), config=config)# Or, human rejects - resume with a rejection reason instead:# result = graph.invoke(#     Command(resume={"approved": False, "reason": "Budget not authorized"}),#     config=config,# )

CONSEQUENCE_MODELER_PROMPT = """You are a consequence modeler for an agentic system.Classify the proposed action before it executes.Proposed action: {action}Action parameters: {parameters}System context: {context}Classification dimensions:1. Reversibility: can this action be undone? (reversible | partially_reversible | irreversible)2. Idempotency: is this action safe to retry? (idempotent | non_idempotent)3. Blast radius: what breaks if this fails or produces wrong output? (isolated | service_level | system_wide)4. Data sensitivity: does this action touch sensitive data? (none | low | high | critical)Return JSON:{{  "reversibility": "<reversible|partially_reversible|irreversible>",  "idempotency": "<idempotent|non_idempotent>",  "blast_radius": "<isolated|service_level|system_wide>",  "data_sensitivity": "<none|low|high|critical>",  "requires_approval": <true|false>,  "safe_to_execute": <true|false>,  "reason": "<one sentence>"}}"""def consequence_modeler(state: AgentState, action: str, parameters: dict) -> dict:    context = {        "loop_count": state.get("loop_count", 0),        "previous_actions": state.get("messages", [])[-4:],    }    prompt = CONSEQUENCE_MODELER_PROMPT.format(        action=action,        parameters=json.dumps(parameters),        context=json.dumps(context, default=str)    )    raw = slm.invoke(prompt)    try:        result = json.loads(raw.strip())    except json.JSONDecodeError:        # Consequence modeler failure - block the action, don't guess        return {            "safe_to_execute": False,            "requires_approval": True,            "reason": "Consequence modeler failed to parse - blocking for safety"        }    # Override: irreversible + non_idempotent always requires approval    if (result.get("reversibility") == "irreversible" and            result.get("idempotency") == "non_idempotent"):        result["requires_approval"] = True        result["safe_to_execute"] = False    # Override: system_wide blast radius always requires approval    if result.get("blast_radius") == "system_wide":        result["requires_approval"] = True    return result

import hashlibimport timedef execute_with_idempotency(    action_fn,    parameters: dict,    consequence: dict,    state: AgentState) -> dict:    """Execute an action with idempotency key if classified as non-idempotent."""    if consequence.get("idempotency") == "non_idempotent":        # Generate deterministic idempotency key from action + params + turn        key_source = json.dumps({            "params": parameters,            "loop_count": state.get("loop_count"),            "message_id": len(state.get("messages", []))        }, sort_keys=True)        idempotency_key = hashlib.sha256(key_source.encode()).hexdigest()[:32]        return action_fn(**parameters, idempotency_key=idempotency_key)    else:        return action_fn(**parameters)

from sentence_transformers import CrossEncoder# Dedicated cross-encoder reranker - faster and more accurate than# a general SLM for pure relevance scoringreranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")def context_pruner_node(state: AgentState, retrieved_chunks: list[str]) -> AgentState:    query = state["messages"][-1].content    if len(retrieved_chunks) <= 3:        # Small enough to pass through directly        state["intent"]["context"] = retrieved_chunks        return state    # Score all (query, chunk) pairs    pairs = [(query, chunk) for chunk in retrieved_chunks]    scores = reranker.predict(pairs)    # Rank and keep top-k    ranked = sorted(        zip(scores, retrieved_chunks),        key=lambda x: x[0],        reverse=True    )    TOP_K = 4    MIN_RELEVANCE_SCORE = 0.1  # Chunks below this are noise regardless of rank    filtered = [        chunk for score, chunk in ranked[:TOP_K]        if score > MIN_RELEVANCE_SCORE    ]    if not filtered:        # All chunks below threshold - better to have the LLM say        # "I don't have enough context" than to hallucinate from noise        state["intent"]["context"] = []        state["intent"]["context_insufficient"] = True    else:        state["intent"]["context"] = filtered        state["intent"]["context_insufficient"] = False    return state

SLM_PRUNER_PROMPT = """Given the query and retrieved chunks, produce a pruned context.Query: {query}Retrieved chunks:{chunks}Tasks:1. Remove chunks with no relevance to the query2. Remove duplicate information across chunks3. If two chunks say the same thing, keep the more specific one4. Preserve all unique facts relevant to the queryReturn a JSON array of the chunks to keep, in order of relevance.Do not modify chunk text. Return chunk indices only: [0, 3, 5]"""def slm_context_pruner_node(state: AgentState, retrieved_chunks: list[str]) -> AgentState:    query = state["messages"][-1].content    chunks_formatted = "\n\n".join([        f"[{i}] {chunk}" for i, chunk in enumerate(retrieved_chunks)    ])    prompt = SLM_PRUNER_PROMPT.format(        query=query,        chunks=chunks_formatted    )    raw = slm.invoke(prompt)    try:        indices = json.loads(raw.strip())        pruned = [retrieved_chunks[i] for i in indices if i < len(retrieved_chunks)]    except (json.JSONDecodeError, IndexError):        # Pruner failure - fall back to first 4 chunks rather than all chunks        pruned = retrieved_chunks[:4]    state["intent"]["context"] = pruned    return state

def direct_lookup_node(state: AgentState) -> AgentState:    """Handles simple factual lookups without LLM - stub for your lookup logic."""    query = state["messages"][-1].content    # Replace with your actual lookup logic (DB query, cache hit, etc.)    result = f"Direct lookup result for: {query}"    state["messages"] = [AIMessage(content=result)]    state["last_action_result"] = "success"    return statedef tool_executor_node(state: AgentState) -> AgentState:    """Executes a validated, consequence-checked tool call."""    intent = state.get("intent", {})    # Replace with your actual tool dispatch logic    state["last_action_result"] = f"Tool executed: {intent.get('intent', 'unknown')}"    state["messages"] = [AIMessage(content=f"Action completed: {state['last_action_result']}")]    return statedef llm_handler_node(state: AgentState) -> AgentState:    """Sends request to the frontier LLM for reasoning tasks."""    context = state.get("intent", {}).get("context", [])    context_str = "\n\n".join(context) if context else ""    messages = state["messages"]    if context_str:        messages = [SystemMessage(content=f"Context:\n{context_str}")] + messages    response = llm.invoke(messages)    state["messages"] = [response]    state["last_action_result"] = "llm_response"    return statedef llm_planner_node(state: AgentState) -> AgentState:    """Re-plans when the loop controller signals REPLAN."""    task = state["messages"][0].content    last_result = state.get("last_action_result", "")    plan_prompt = (        f"Task: {task}\n"        f"Previous approach failed: {last_result}\n"        "Devise a different approach. Be specific about the next action to take."    )    response = llm.invoke(plan_prompt)    state["messages"] = [response]    state["last_action_result"] = "replanned"    return statedef response_formatter_node(state: AgentState) -> AgentState:    """Formats the final response when the loop controller signals COMPLETE."""    # Last message already contains the result - clean it up if needed    state["last_action_result"] = "complete"    return statedef abort_handler_node(state: AgentState) -> AgentState:    """Returns a safe error when the loop hits the hard circuit breaker."""    loop_count = state.get("loop_count", 0)    state["messages"] = [AIMessage(        content=(            f"The task could not be completed after {loop_count} attempts. "            "Please try rephrasing your request or contact support."        )    )]    state["last_action_result"] = "aborted"    return statedef clarification_handler_node(state: AgentState) -> AgentState:    """Asks the user for clarification when intent is ambiguous."""    state["messages"] = [AIMessage(        content="I need a bit more information to help you. Could you clarify what you're trying to accomplish?"    )]    return statedef consequence_check_node(state: AgentState) -> AgentState:    """Wrapper that runs consequence_modeler against the current intent."""    intent = state.get("intent", {})    action = intent.get("intent", "unknown_action")    parameters = {k: v for k, v in intent.items() if k != "intent"}    consequence = consequence_modeler(state, action, parameters)    if not consequence.get("safe_to_execute", False) or consequence.get("requires_approval", False):        state["escalate"] = True        state["last_action_result"] = f"blocked:{consequence.get('reason', 'requires approval')}"    else:        state["escalate"] = False    return statedef context_pruner_wrapper(state: AgentState) -> AgentState:    """Retrieves and prunes context before passing to LLM handler."""    from langchain_chroma import Chroma    from langchain_openai import OpenAIEmbeddings    query = state["messages"][-1].content    # Concrete retrieval example using Chroma - swap for your vector DB    # (FAISS: use FAISS.load_local(...).similarity_search(query, k=10))    # (Pinecone: use index.query(vector=embed(query), top_k=10))    vectorstore = Chroma(        collection_name="knowledge_base",        embedding_function=OpenAIEmbeddings(),        persist_directory="./chroma_db"    )    docs = vectorstore.similarity_search(query, k=10)    retrieved_chunks = [doc.page_content for doc in docs]    return context_pruner_node(state, retrieved_chunks)

def build_support_pipeline():    graph = StateGraph(AgentState)    # Stage 1: Route the request    graph.add_node("router", slm_router)    # Stage 2a: Simple lookup path (no LLM)    graph.add_node("direct_lookup", direct_lookup_node)    # Stage 2b: Tool execution path    graph.add_node("schema_enforcer", calendar_enforcer)    graph.add_node("intent_validator", validate_calendar_intent)    graph.add_node("consequence_check", consequence_check_node)    graph.add_node("tool_executor", tool_executor_node)    # Stage 2c: Complex reasoning path    graph.add_node("context_pruner", context_pruner_wrapper)    graph.add_node("llm_handler", llm_handler_node)    # Stage 3: Output validation (all paths)    graph.add_node("output_validator", output_validator)    # Entry    graph.set_entry_point("router")    # Router branches    graph.add_conditional_edges(        "router",        route_decision,        {            "direct_response": "direct_lookup",            "tool_executor": "schema_enforcer",            "llm_handler": "context_pruner",            "clarification_handler": "output_validator",  # Return clarification request        }    )    # Tool path: enforce -> validate -> consequence check -> execute    graph.add_edge("schema_enforcer", "intent_validator")    graph.add_conditional_edges(        "intent_validator",        lambda s: "consequence_check" if not s.get("escalate") else "llm_handler",        {            "consequence_check": "consequence_check",            "llm_handler": "llm_handler",        }    )    graph.add_conditional_edges(        "consequence_check",        lambda s: "tool_executor" if not s.get("escalate") else "llm_handler",        {            "tool_executor": "tool_executor",            "llm_handler": "llm_handler",        }    )    # All paths converge at output validator    graph.add_edge("direct_lookup", "output_validator")    graph.add_edge("tool_executor", "output_validator")    graph.add_edge("llm_handler", "output_validator")    graph.add_edge("context_pruner", "llm_handler")    graph.add_edge("output_validator", END)    return graph.compile(checkpointer=MemorySaver())

import structlogfrom opentelemetry import tracelogger = structlog.get_logger()tracer = trace.get_tracer("slm-pipeline")def instrument_slm_node(node_fn, node_name: str):    """Decorator that adds structured logging and OpenTelemetry spans to any SLM node."""    def instrumented(state: AgentState) -> AgentState:        with tracer.start_as_current_span(f"slm.{node_name}") as span:            input_summary = state["messages"][-1].content[:100] if state.get("messages") else ""            span.set_attribute("slm.node", node_name)            span.set_attribute("slm.input_preview", input_summary)            span.set_attribute("slm.loop_count", state.get("loop_count", 0))            result_state = node_fn(state)            span.set_attribute("slm.route", result_state.get("route", ""))            span.set_attribute("slm.confidence", result_state.get("confidence", 0.0))            span.set_attribute("slm.escalate", result_state.get("escalate", False))            logger.info(                "slm_decision",                node=node_name,                route=result_state.get("route"),                confidence=result_state.get("confidence"),                escalate=result_state.get("escalate"),                loop_count=result_state.get("loop_count"),                input_preview=input_summary,            )            return result_state    return instrumented# Apply to routerinstrumented_router = instrument_slm_node(slm_router, "router")

import pytestfrom langchain_core.messages import HumanMessage# Router test suiteROUTER_TEST_CASES = [    # (input, expected_route, min_confidence)    ("What is my order status for #12345?",       "tool_call",          0.8),    ("Explain the difference between TCP and UDP", "complex_reasoning",  0.75),    ("What does refund mean?",                     "simple_lookup",      0.8),    ("do the thing",                               "clarification_needed", 0.6),    ("Book a meeting with Priya tomorrow at 3pm",  "tool_call",          0.8),    ("Why is my payment failing?",                 "complex_reasoning",  0.7),]@pytest.mark.parametrize("request,expected_route,min_confidence", ROUTER_TEST_CASES)def test_router_classification(request, expected_route, min_confidence):    state = AgentState(        messages=[HumanMessage(content=request)],        route="", confidence=0.0, intent={},        loop_count=0, last_action_result="", escalate=False    )    result = slm_router(state)    assert result["route"] == expected_route, (        f"Expected {expected_route}, got {result['route']} "        f"(confidence: {result['confidence']:.2f}) for: {request}"    )    assert result["confidence"] >= min_confidence, (        f"Confidence {result['confidence']:.2f} below threshold {min_confidence} for: {request}"    )# Validator test suiteVALIDATOR_TEST_CASES = [    # (intent_dict, should_pass)    ({"intent": "create_event", "title": "Standup", "start_time": "2026-04-01T09:00:00"}, True),    ({"intent": "create_event", "title": "", "start_time": "2026-04-01T09:00:00"},          False),  # Empty title    ({"intent": "create_event", "title": "Standup", "start_time": "not-a-date"},            False),  # Bad date    ({"intent": "create_event", "title": "Standup",      "start_time": "2026-04-01T10:00:00", "end_time": "2026-04-01T09:00:00"},              False),  # End before start]@pytest.mark.parametrize("intent,should_pass", VALIDATOR_TEST_CASES)def test_intent_validator(intent, should_pass):    state = AgentState(        messages=[], route="", confidence=0.9,        intent=intent, loop_count=0, last_action_result="", escalate=False    )    result = validate_calendar_intent(state)    if should_pass:        assert not result["escalate"], f"Validation failed unexpectedly for: {intent}"    else:        assert result["escalate"], f"Validation passed unexpectedly for: {intent}"