File size: 3,307 Bytes
b22c324 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | import json
import chromadb
from agentic_workflow.config import LLM_MODEL
from langchain_ollama import ChatOllama
from dataclasses import dataclass, field
llm = ChatOllama(model=LLM_MODEL, temperature=0)
@dataclass
class PlanStep:
name: str # e.g. "retrieval", "pii_check"
enabled: bool # whether this step should run
reason: str # why it was enabled or skipped
@dataclass
class Plan:
steps: list[PlanStep] = field(default_factory=list)
def is_enabled(self, step_name: str) -> bool:
"""Quick lookup — used by the pipeline to check if a step should run."""
return any(s.name == step_name and s.enabled for s in self.steps)
def summary(self) -> None:
"""Prints a readable breakdown of the plan."""
print("\n── Execution Plan ──────────────────────────")
for s in self.steps:
status = "RUN " if s.enabled else "SKIP"
print(f" [{status}] {s.name:<20} → {s.reason}")
print("────────────────────────────────────────────\n")
def create_plan(intent_result: dict) -> Plan:
"""
Reads the intent analysis output and decides which steps
the pipeline should execute, with a reason for each decision.
Args:
intent_result: dict returned by analyse_intent(), e.g.
{
"intent": "factual",
"needs_context": True,
"needs_history": False
}
Returns:
Plan object with a PlanStep for each possible action.
"""
intent = intent_result.get("intent", "factual")
needs_context = intent_result.get("needs_context", True)
needs_history = intent_result.get("needs_history", False)
steps = []
# PII check
# Always runs — sensitive queries must be gated regardless of intent
steps.append(PlanStep(
name="pii_check",
enabled=True,
reason="Always runs to protect against personal data leakage"
))
# Vector retrieval
# Only runs for factual queries that need external knowledge
steps.append(PlanStep(
name="retrieval",
enabled=needs_context,
reason=(
"Query requires knowledge base context"
if needs_context
else f"Skipped — '{intent}' intent does not need retrieval"
)
))
# Conversation history
# Only runs when the query references a prior turn
steps.append(PlanStep(
name="history",
enabled=needs_history,
reason=(
"Query appears to reference a prior turn"
if needs_history
else "Skipped — query is self-contained"
)
))
# LLM generation
# Always runs — we always need a response
steps.append(PlanStep(
name="llm_generation",
enabled=True,
reason="Always runs to generate the final response"
))
# Quality evaluation
# Always runs — every response should be checked before returning
steps.append(PlanStep(
name="quality_eval",
enabled=True,
reason="Always runs to verify response quality before returning"
))
return Plan(steps=steps)
|