待翻译：The End-to-End Agentic AI Pipeline

AI 服务暂时不可用，以下为来源正文，待恢复后补全翻译。

The End-to-End Agentic AI Pipeline - MachineLearningMastery.com The End-to-End Agentic AI Pipeline - MachineLearningMastery.com In this article, you will learn the seven architectural components that separate a production-grade agentic AI system from a demo script, and how each one fits into the agent’s core feedback loop. Topics we will cover include: What each of the seven components — perception, memory, reasoning and planning, tool execution, orchestration, guardrails, and observability — is specifically responsible for. Where each component tends to break in real systems, and why that component must be kept separate from the others. Focused, runnable Python code illustrating the responsibility of each component in isolation. Introduction Most “build an AI agent” tutorials show a 40-line script that calls an LLM in a loop and calls it done. That script works fine for a demo. It does not survive a second concurrent user, a flaky third-party API, or a task that turns out to need twelve steps instead of two. The gap between the demo and the production system isn’t clever prompting. It’s architecture. Production agentic systems are built from a consistent set of interconnected components: perception, reasoning, planning, memory, tool execution, orchestration, and guardrails. That same component breakdown shows up across nearly every serious architecture writeup, survey paper, and production postmortem published in the last year, regardless of which framework or vendor is doing the writing. The loop underneath all of it is consistent: Goal → Perception → Reasoning → Planning → Action → Observation → Memory Update → back to Reasoning, repeating until the goal is met, a stop condition fires, or the agent decides it needs a human. This article walks through each piece of that loop as its own component — what it’s responsible for, where it tends to break, and a focused code excerpt that makes the responsibility concrete. Nothing here is wired into one running pipeline. Each piece is shown in isolation, which is also how you should reason about your own system when deciding what it needs. The Seven Components, at a Glance Architectural surveys converge on the same core set: Perception, Memory, Reasoning/Planning, Tool Execution, and Orchestration form a closed feedback loop — the cycle that actually runs, step after step. Guardrails and Observability wrap around that entire loop as cross-cutting concerns rather than steps inside the sequence. You don’t “do” guardrails at step 4; guardrails sit between every proposed action and the world, watching every step. That distinction shapes the rest of this article. The first five sections walk through the loop in the order data actually flows through it. The last two sections cover the wrapper layers that make the loop survivable once real money, real customers, and real side effects are involved. Turning Raw Input Into Something the Agent Can Reason About Perception’s job is to transform raw inputs — text, voice, API payloads, sensor data, and file uploads — into a structured representation that the reasoning engine can actually work with. This is the component most tutorials skip entirely, because in a demo, “the user just types text” and there’s nothing to normalize. Real systems take input from webhooks, structured API calls, file uploads, and multiple channels simultaneously, and every one of those needs to land in the same shape before anything downstream can trust it. 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 # perception.py # Prerequisites: none beyond Python's standard library # Run: python perception.py from dataclasses import dataclass, field 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 internal shape every downstream component consumes, regardless of where the raw input actually came from. This is the entire point of a perception layer: everything past this point only ever sees this one structure. """ source: InputSource content: str metadata: dict[str, Any] = field(default_factory=dict) received_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) def perceive_user_text(raw_text: str) -> AgentInput: """Raw chat input -- the simplest case, but it still needs normalization.""" return AgentInput( source=InputSource.USER_TEXT, content=raw_text.strip(), metadata={"channel": "chat"}, ) def perceive_webhook(raw_payload: str) -> AgentInput: """ A webhook delivers structured JSON, not plain text. Perception extracts the part the agent should reason about and discards transport-level noise like headers and signatures. """ payload = json.loads(raw_payload) event_type = payload.get("event_type", "unknown") description = payload.get("description", "") return AgentInput( source=InputSource.WEBHOOK, content=f"Event '{event_type}' received: {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 upload event has no natural-language content at all -- perception has to construct something the reasoning engine can actually use. """ return AgentInput( source=InputSource.FILE_UPLOAD, content=f"User 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's the status 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, "application/pdf") for inp in [text_input, webhook_input, file_input]: print(f"[{inp.source.value}] content='{inp.content}'") print(f" metadata keys: {list(inp.metadata.keys())}\n") How to run: python perception.py, no dependencies required. Three completely different raw shapes — plain text, a webhook JSON payload, and a file-upload event — all collapse into the same AgentInput structure. The reasoning component downstream never needs to know or care which channel something arrived through. That’s the entire value of treating perception as its own component rather than inlining ad hoc parsing wherever input happens to enter the system. Working Context vs. What Actually Persists This is the component with the most nuance, and the one demo code gets wrong most often by treating “memory” as just “the conversation so far.” Production memory architecture separates working memory — the immediate context window for the current task — from long-term memory, which itself splits into episodic memory (what happened), semantic memory (facts learned), and procedural memory (skills and how-to knowledge). Short-term memory lives in-context and is essentially free; long-term memory typically lives in a vector store, indexed for semantic retrieval rather than exact match. The operational distinction matters: working memory is fast and disposable — it evaporates the moment the session ends. Episodic memory gives the agent something working memory structurally cannot provide: hindsight across sessions, the ability to recall “we handled something like this before, and here’s what happened.” 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 # memory.py # Prerequisites: none beyond Python's standard library # Run: python memory.py from dataclasses import dataclass, field from datetime import datetime, timezone @dataclass class WorkingMemory: """ Working memory: the immediate context for the CURRENT task only. Lives in-process, bounded by a turn limit, and is gone the moment the session ends. This is what most demo code calls "memory" -- but it's only one piece of the real picture. """ max_turns: int = 10 turns: list[dict] = field(default_factory=list) def add_turn(self, role: str, content: str) -> None: self.turns.append({"role": role, "content": content}) if len(self.turns) > self.max_turns: self.turns.pop(0) # Oldest turn drops off once the limit is hit def as_context(self) -> str: return "\n".join(f"{t['role']}: {t['content']}" for t in self.turns) @dataclass class EpisodicMemoryEntry: """A single stored episode -- what happened, when, and its embedding for later recall.""" timestamp: str summary: str embedding: list[float] # In production this comes from a real embedding model class EpisodicMemory: """ Episodic memory: persists ACROSS sessions, stored externally (a vector store in production), and retrieved by semantic similarity rather than recency. This is what gives an agent "hindsight" -- a capability working memory structurally cannot have, since it's gone the instant the session ends. """ def init(self): self._store: list[EpisodicMemoryEntry] = [] def record_episode(self, summary: str, embedding: list[float]) -> None: self._store.append(EpisodicMemoryEntry( timestamp=datetime.now(timezone.utc).isoformat(), summary=summary, embedding=embedding, )) def retrieve_similar(self, query_embedding: list[float], top_k: int = 2) -> list[EpisodicMemoryEntry]: """Real implementations do cosine similarity against 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("user", "What's my refund status?") wm.add_turn("agent", "Let me check that for you.") wm.add_turn("user", "It's order 4821") wm.add_turn("agent", "Found it -- refund is processing") # pushes the first turn out print("Working memory (bounded to last 3 turns):") print(wm.as_context()) em = EpisodicMemory() em.record_episode("User asked about refund for order 4821, resolved successfully", [0.9, 0.1, 0.0]) em.record_episode("User asked about shipping delay for order 1190", [0.1, 0.9, 0.0]) em.record_episode("User asked about refund eligibility for order 7734", [0.85, 0.15, 0.0]) similar = em.retrieve_similar([0.88, 0.12, 0.0], top_k=2) print("\nEpisodic memory -- retrieved by similarity to a new refund query:") for entry in similar: print(f" {entry.summary}") How to run: python memory.py, no dependencies required. Working memory drops its oldest turn once the limit is hit, and the first exchange about checking the status is gone by the end of the session. Episodic memory does the opposite: it surfaces the two refund-related episodes out of three stored entries, ranked by meaning, not by when they happened. That’s the structural line between the two — one is a sliding window, the other is a searchable archive. Reasoning and Planning (Deciding What to Do Next) Reasoning and planning take the current goal, the perceived input, and whatever memory was retrieved, and produce a plan — sometimes a single next action, sometimes a multi-step decomposition. This is the agent’s cognitive core, consulting memory and knowledge resources to synthesize action plans that get handed off to the execution module. The critical design point, easy to miss: planning’s responsibility ends at producing the plan. It does not call a tool, touch an API, or hav [truncated for AI cost control]