The Finish-to-Finish Agentic AI Pipeline

# notion.py
# Conditions: none past Python’s normal library
# Run: python notion.py

from dataclasses import dataclass, subject
from typing import Any
from enum import Enum
import json
from datetime import datetime, timezone

class InputSource(Enum):
USER_TEXT = “user_text”
WEBHOOK = “webhook”
FILE_UPLOAD = “file_upload”

@dataclass
class AgentInput:
“””
The normalized inner form each downstream part consumes,
no matter the place the uncooked enter really got here from. That is the
complete level of a notion layer: all the pieces previous this level
solely ever sees this one construction.
“””
supply: InputSource
content material: str
metadata: dict[str, Any] = subject(default_factory=dict)
received_at: str = subject(default_factory=lambda: datetime.now(timezone.utc).isoformat())

def perceive_user_text(raw_text: str) -> AgentInput:
“””Uncooked chat enter — the best case, but it surely nonetheless wants normalization.”””
return AgentInput(
supply=InputSource.USER_TEXT,
content material=raw_text.strip(),
metadata={“channel”: “chat”},
)

def perceive_webhook(raw_payload: str) -> AgentInput:
“””
A webhook delivers structured JSON, not plain textual content. Notion extracts
the half the agent ought to cause about and discards transport-level
noise like headers and signatures.
“””
payload = json.masses(raw_payload)
event_type = payload.get(“event_type”, “unknown”)
description = payload.get(“description”, “”)
return AgentInput(
supply=InputSource.WEBHOOK,
content material=f”Occasion ‘{event_type}’ acquired: {description}”,
metadata={“event_type”: event_type, “raw_payload”: payload},
)

def perceive_file_upload(filename: str, file_size_bytes: int, mime_type: str) -> AgentInput:
“””
A file add occasion has no natural-language content material in any respect — notion
has to assemble one thing the reasoning engine can really use.
“””
return AgentInput(
supply=InputSource.FILE_UPLOAD,
content material=f”Consumer uploaded file ‘{filename}’ ({mime_type}, {file_size_bytes} bytes)”,
metadata={“filename”: filename, “mime_type”: mime_type, “size_bytes”: file_size_bytes},
)


if __name__ == “__main__”:
text_input = perceive_user_text(” What is the standing of my refund? “)
webhook_input = perceive_webhook(json.dumps({
“event_type”: “payment_failed”,
“description”: “Card declined for order #4821”,
}))
file_input = perceive_file_upload(“invoice_q3.pdf”, 184320, “utility/pdf”)

for inp in [text_input, webhook_input, file_input]:
print(f”[{inp.source.value}] content material=”{inp.content material}””)
print(f” metadata keys: {record(inp.metadata.keys())}n”)

# reminiscence.py
# Conditions: none past Python’s normal library
# Run: python reminiscence.py

from dataclasses import dataclass, subject
from datetime import datetime, timezone

@dataclass
class WorkingMemory:
“””
Working reminiscence: the fast context for the CURRENT process solely.
Lives in-process, bounded by a flip restrict, and is gone the second
the session ends. That is what most demo code calls “reminiscence” –
but it surely’s just one piece of the actual image.
“””
max_turns: int = 10
turns: record[dict] = subject(default_factory=record)

def add_turn(self, position: str, content material: str) -> None:
self.turns.append({“position”: position, “content material”: content material})
if len(self.turns) > self.max_turns:
self.turns.pop(0) # Oldest flip drops off as soon as the restrict is hit

def as_context(self) -> str:
return “n”.be a part of(f”{t[‘role’]}: {t[‘content’]}” for t in self.turns)


@dataclass
class EpisodicMemoryEntry:
“””A single saved episode — what occurred, when, and its embedding for later recall.”””
timestamp: str
abstract: str
embedding: record[float] # In manufacturing this comes from an actual embedding mannequin

class EpisodicMemory:
“””
Episodic reminiscence: persists ACROSS periods, saved externally
(a vector retailer in manufacturing), and retrieved by semantic similarity
fairly than recency. That is what offers an agent “hindsight” — a
functionality working reminiscence structurally can’t have, because it’s gone
the moment the session ends.
“””
def __init__(self):
self._store: record[EpisodicMemoryEntry] = []

def record_episode(self, abstract: str, embedding: record[float]) -> None:
self._store.append(EpisodicMemoryEntry(
timestamp=datetime.now(timezone.utc).isoformat(),
abstract=abstract,
embedding=embedding,
))

def retrieve_similar(self, query_embedding: record[float], top_k: int = 2) -> record[EpisodicMemoryEntry]:
“””Actual implementations do cosine similarity in opposition to a vector index.”””
def dot(a, b): return sum(x * y for x, y in zip(a, b))
ranked = sorted(self._store, key=lambda e: dot(e.embedding, query_embedding), reverse=True)
return ranked[:top_k]


if __name__ == “__main__”:
wm = WorkingMemory(max_turns=3)
wm.add_turn(“consumer”, “What’s my refund standing?”)
wm.add_turn(“agent”, “Let me examine that for you.”)
wm.add_turn(“consumer”, “It is order 4821”)
wm.add_turn(“agent”, “Discovered it — refund is processing”) # pushes the primary prove

print(“Working reminiscence (bounded to final 3 turns):”)
print(wm.as_context())

em = EpisodicMemory()
em.record_episode(“Consumer requested about refund for order 4821, resolved efficiently”, [0.9, 0.1, 0.0])
em.record_episode(“Consumer requested about delivery delay for order 1190”, [0.1, 0.9, 0.0])
em.record_episode(“Consumer requested about refund eligibility for order 7734”, [0.85, 0.15, 0.0])

comparable = em.retrieve_similar([0.88, 0.12, 0.0], top_k=2)
print(“nEpisodic reminiscence — retrieved by similarity to a brand new refund question:”)
for entry in comparable:
print(f” {entry.abstract}”)

# planning.py
# Conditions: none past Python’s normal library
# Run: python planning.py

from dataclasses import dataclass, subject
import json

@dataclass
class PlanStep:
step_number: int
description: str
tool_tag: str # which class of instrument this step will want — execution decides HOW

@dataclass
class Plan:
objective: str
steps: record[PlanStep] = subject(default_factory=record)

def mock_llm_plan(objective: str) -> str:
“””
Mock LLM name standing in for an actual planning name. The purpose being
demonstrated is structural: planning PRODUCES a plan object — it does
not execute something. Execution is a separate part (subsequent part)
that the plan will get handed off to.
“””
if “refund” in objective.decrease():
return json.dumps({
“steps”: [
{“step_number”: 1, “description”: “Look up the order by ID”, “tool_tag”: “database_lookup”},
{“step_number”: 2, “description”: “Check refund eligibility against policy”, “tool_tag”: “policy_check”},
{“step_number”: 3, “description”: “Issue the refund if eligible”, “tool_tag”: “payment_api”},
{“step_number”: 4, “description”: “Notify the customer of the outcome”, “tool_tag”: “email”},
]
})
return json.dumps({“steps”: [
{“step_number”: 1, “description”: “Search knowledge base for the answer”, “tool_tag”: “search”},
]})

def create_plan(objective: str) -> Plan:
“””Take a objective, produce a structured plan. Nothing right here has a facet impact.”””
parsed = json.masses(mock_llm_plan(objective))
return Plan(objective=objective, steps=[PlanStep(**s) for s in parsed[“steps”]])


if __name__ == “__main__”:
for objective in [“Process a refund for order 4821”, “What are your business hours?”]:
plan = create_plan(objective)
print(f”Objective: {plan.objective}”)
for step in plan.steps:
print(f” Step {step.step_number}: {step.description} [tool_tag={step.tool_tag}]”)
print()

# tool_execution.py
# Conditions: none past Python’s normal library
# Run: python tool_execution.py

import time
import hashlib
from dataclasses import dataclass
from typing import Callable, Any, Elective

@dataclass
class ToolResult:
success: bool
output: Any = None
error: Elective[str] = None
idempotency_key: Elective[str] = None
from_cache: bool = False

class ToolExecutor:
“””
The non-negotiable fundamentals for instrument execution: validate inputs earlier than
calling something, implement a timeout so a sluggish name cannot hold the
complete run, and make side-effecting calls idempotent so a retry
would not double-charge a buyer or double-send an e-mail.
“””
def __init__(self, timeout_seconds: float = 5.0):
self.timeout_seconds = timeout_seconds
self._idempotency_cache: dict[str, ToolResult] = {}

def _make_idempotency_key(self, tool_name: str, args: dict) -> str:
uncooked = f”{tool_name}:{sorted(args.objects())}”
return hashlib.sha256(uncooked.encode()).hexdigest()[:16]

def execute(self, tool_name: str, tool_fn: Callable, args: dict,
required_args: record[str], idempotent: bool = False) -> ToolResult:
# 1. Validate earlier than touching something exterior
lacking = [a for a in required_args if a not in args]
if lacking:
return ToolResult(success=False, error=f”Lacking required args: {lacking}”)

# 2. Idempotency — a retried name with similar args returns the cached end result
idem_key = self._make_idempotency_key(tool_name, args) if idempotent else None
if idem_key and idem_key in self._idempotency_cache:
cached = self._idempotency_cache[idem_key]
return ToolResult(success=cached.success, output=cached.output,
idempotency_key=idem_key, from_cache=True)

# 3. Execute in opposition to a timeout funds
begin = time.monotonic()
attempt:
output = tool_fn(**args)
besides Exception as e:
return ToolResult(success=False, error=str(e), idempotency_key=idem_key)

if (time.monotonic() – begin) > self.timeout_seconds:
return ToolResult(success=False, error=f”Device exceeded {self.timeout_seconds}s timeout”,
idempotency_key=idem_key)

end result = ToolResult(success=True, output=output, idempotency_key=idem_key)
if idem_key:
self._idempotency_cache[idem_key] = end result
return end result


def issue_refund(order_id: str, quantity: float) -> str:
return f”Refunded ${quantity} for order {order_id}”


if __name__ == “__main__”:
executor = ToolExecutor(timeout_seconds=1.0)

# Lacking arg — caught earlier than execution
r1 = executor.execute(“issue_refund”, issue_refund, {“order_id”: “4821”}, [“order_id”, “amount”])
print(f”Lacking arg check: success={r1.success}, error=”{r1.error}””)

# First name executes, retry with similar args hits the idempotency cache
args = {“order_id”: “4821”, “quantity”: 49.99}
r2a = executor.execute(“issue_refund”, issue_refund, args, [“order_id”, “amount”], idempotent=True)
r2b = executor.execute(“issue_refund”, issue_refund, args, [“order_id”, “amount”], idempotent=True)
print(f”nFirst name: from_cache={r2a.from_cache}, output=”{r2a.output}””)
print(f”Retry, identical args: from_cache={r2b.from_cache}, output=”{r2b.output}””)

# orchestrator.py
# Conditions: none past Python’s normal library
# Run: python orchestrator.py

from dataclasses import dataclass, subject

@dataclass
class PlanStep:
step_number: int
description: str
tool_tag: str

@dataclass
class Plan:
objective: str
steps: record[PlanStep] = subject(default_factory=record)

@dataclass
class StepOutcome:
step_number: int
success: bool
output: str

# Stand-ins for the planning and tool-execution parts above –
# orchestration’s job is to CALL these in sequence, not do their work itself.
def mock_create_plan(objective: str) -> Plan:
return Plan(objective=objective, steps=[
PlanStep(1, “Look up order”, “database_lookup”),
PlanStep(2, “Check eligibility”, “policy_check”),
PlanStep(3, “Issue refund”, “payment_api”),
])

def mock_execute_tool(step: PlanStep) -> StepOutcome:
if step.tool_tag == “policy_check”: # simulate a failure to check cease logic
return StepOutcome(step.step_number, success=False, output=”Coverage examine failed: order too previous”)
return StepOutcome(step.step_number, success=True, output=f”Accomplished: {step.description}”)

class Orchestrator:
“””
Holds the sequence collectively throughout a number of steps, decides whether or not to
proceed or cease, and caps the run with an specific step restrict. It does
not plan and it doesn’t execute instruments — it calls the parts that do.
“””
def __init__(self, max_steps: int = 10):
self.max_steps = max_steps

def run(self, objective: str) -> record[StepOutcome]:
plan = mock_create_plan(objective)
outcomes: record[StepOutcome] = []

for step in plan.steps:
if len(outcomes) >= self.max_steps:
print(f” Step restrict ({self.max_steps}) reached — stopping.”)
break

print(f” Executing step {step.step_number}: {step.description}”)
final result = mock_execute_tool(step)
outcomes.append(final result)

if not final result.success:
print(f” Step {step.step_number} failed: {final result.output}”)
print(f” Stopping run — a failed step blocks the remainder of this plan.”)
break

return outcomes


if __name__ == “__main__”:
outcomes = Orchestrator(max_steps=10).run(“Course of a refund for order 4821″)
print(f”nTotal steps tried: {len(outcomes)}”)
print(f”Closing final result: success={outcomes[-1].success}, output=”{outcomes[-1].output}””)

Executing step 1: Lookup order
Executing step 2: Examine eligibility
Step 2 failed: Coverage examine failed: order too previous
Stopping run — a failed step blocks the remainder of this plan

Whole steps tried: 2
Closing final result: success=False, output=”Coverage examine failed: order too previous”

# guardrails.py
# Conditions: none past Python’s normal library
# Run: python guardrails.py

from dataclasses import dataclass
from enum import Enum

class GuardrailVerdict(Enum):
ALLOW = “enable”
DENY = “deny”
REQUIRE_APPROVAL = “require_approval”

@dataclass
class ProposedAction:
tool_name: str
args: dict
estimated_cost: float
irreversible: bool

@dataclass
class GuardrailResult:
verdict: GuardrailVerdict
cause: str

class GuardrailEngine:
“””
Sits between planning’s proposed motion and the instrument executor.
Enforces an allow-list of permitted instruments, a value ceiling per motion,
and obligatory human approval for something irreversible — regardless
of how assured the planner was that it was the fitting transfer.
“””
def __init__(self, allowed_tools: set[str], cost_ceiling: float):
self.allowed_tools = allowed_tools
self.cost_ceiling = cost_ceiling

def examine(self, motion: ProposedAction) -> GuardrailResult:
if motion.tool_name not in self.allowed_tools:
return GuardrailResult(GuardrailVerdict.DENY,
f”Device ‘{motion.tool_name}’ isn’t on the allow-list”)
if motion.estimated_cost > self.cost_ceiling:
return GuardrailResult(GuardrailVerdict.DENY,
f”Estimated value ${motion.estimated_cost} exceeds ceiling ${self.cost_ceiling}”)
if motion.irreversible:
return GuardrailResult(GuardrailVerdict.REQUIRE_APPROVAL,
“Motion is irreversible — requires human approval earlier than execution”)
return GuardrailResult(GuardrailVerdict.ALLOW, “Handed all guardrail checks”)


if __name__ == “__main__”:
guardrails = GuardrailEngine(
allowed_tools={“database_lookup”, “policy_check”, “payment_api”, “e-mail”},
cost_ceiling=100.0,
)

test_actions = [
ProposedAction(“database_lookup”, {“order_id”: “4821”}, 0.0, False),
ProposedAction(“delete_customer_account”, {“id”: “u_991”}, 0.0, True),
ProposedAction(“payment_api”, {“order_id”: “4821”, “amount”: 5000.0}, 5000.0, True),
ProposedAction(“payment_api”, {“order_id”: “4821”, “amount”: 49.99}, 49.99, True),
]

for motion in test_actions:
end result = guardrails.examine(motion)
print(f”{motion.tool_name} -> {end result.verdict.worth}: {end result.cause}”)

database_lookup -> enable: Handed all guardrail checks
delete_customer_account -> deny: Device ‘delete_customer_account’ isn’t on the allow-list
payment_api -> deny: Estimated value $5000.0 exceeds ceiling $100.0
payment_api -> require_approval: Motion is irreversible — requires human approval earlier than execution

# observability.py
# Conditions: none past Python’s normal library
# Run: python observability.py

from dataclasses import dataclass, subject
from datetime import datetime, timezone
from typing import Any
import json

@dataclass
class TraceEntry:
step_number: int
part: str
occasion: str
element: dict[str, Any]
timestamp: str = subject(default_factory=lambda: datetime.now(timezone.utc).isoformat())

class TraceLogger:
“””
Wraps every step of a run with a structured, timestamped log entry.
Intentionally separate from the orchestrator: the orchestrator decides
what occurs, the hint logger simply watches and data it — which is
what helps you to level on the actual step, part, and enter that
brought about a deviation, with out re-running something.
“””
def __init__(self, run_id: str):
self.run_id = run_id
self.entries: record[TraceEntry] = []
self._step_counter = 0

def log(self, part: str, occasion: str, **element) -> None:
self._step_counter += 1
self.entries.append(TraceEntry(self._step_counter, part, occasion, element))

def export_json(self) -> str:
return json.dumps({
“run_id”: self.run_id,
“hint”: [
{“step”: e.step_number, “component”: e.component, “event”: e.event,
“detail”: e.detail, “timestamp”: e.timestamp}
for e in self.entries
],
}, indent=2)

def find_failure_point(self) -> TraceEntry | None:
“””The only most helpful question on a hint: the place did it first go fallacious.”””
for entry in self.entries:
if entry.element.get(“success”) is False:
return entry
return None


if __name__ == “__main__”:
tracer = TraceLogger(run_id=”run_2026_06_20_001″)
tracer.log(“notion”, “input_normalized”, supply=”user_text”, content material=”Refund order 4821″)
tracer.log(“planning”, “plan_created”, step_count=3, objective=”Course of a refund for order 4821″)
tracer.log(“tool_execution”, “tool_called”, instrument=”database_lookup”, success=True, output=”Order discovered”)
tracer.log(“tool_execution”, “tool_called”, instrument=”policy_check”, success=False,
output=”Order too previous”, error=”Coverage examine failed: order too previous”)
tracer.log(“orchestrator”, “run_stopped”, cause=”step_failed”, step_number=4)

failure = tracer.find_failure_point()
print(f”Failure level: part=”{failure.part}”, step={failure.step_number}”)
print(f” element: {failure.element}”)