DEV Community: bredmond1019

The New Code: Why Specifications Will Replace Programming

bredmond1019 — Thu, 25 Jun 2026 15:38:28 +0000

The agents were doing exactly what I told them to. That was the problem.

I'd built a pipeline where AI agents could take a spec file, implement a feature, run the tests, review the result, and commit — without me writing a line of code. It mostly worked. Dozens of features shipped. But I kept reviewing the output and feeling like something was off. Not broken. Just subtly wrong in a way that was hard to name.

I spent a while blaming the models. Then the prompts. Then the validation steps. Eventually I had to sit with the obvious: the agents were implementing exactly what I'd written. My specs were underspecified. The bottleneck was always me, at the planning stage.

The thing most people throw away

There's something that feels right about vibe coding. You're operating at the level of intent — describing what you want and letting the model handle the mechanics. That part is genuinely useful.

But watch what most people do with the output:

Traditional development:
Source code  →  Compiler  →  Binary
(keep the source; regenerate binary anytime)

Vibe coding done wrong:
Prompt  →  LLM  →  Generated code
(delete the prompt; commit the code)

You've shredded the source and carefully version-controlled the binary.

The prompt — your structured description of what you wanted, why, and what "correct" meant — is the valuable artifact. The generated code is what compiles from it. When you discard the prompt and commit only the output, you've lost the thing that actually mattered.

The practical consequence shows up six months later: you're staring at code you wrote and spending twenty minutes reverse-engineering your own intent. The spec would have been a thirty-second read.

What a spec-driven pipeline is

I built what I call an SDLC (Software Development Lifecycle) harness — a system where instead of writing code directly, you write a spec describing what needs to be built, and AI agents handle the implementation, testing, review, and documentation.

The spec is the source. The code is what gets compiled from it.

Simple idea. The interesting part is figuring out how to run that pipeline efficiently. I made some expensive mistakes along the way.

The first version: expensive

My original design ran every task through the full pipeline — implement, test, review, document, wrap-up — independently, in an isolated environment, in parallel:

tasks.md
   │
   ├── Task 1 (isolated)  →  implement → test → review → doc → wrap  [~200k tokens]
   ├── Task 2 (isolated)  →  implement → test → review → doc → wrap  [~200k tokens]
   └── Task 3 (isolated)  →  implement → test → review → doc → wrap  [~200k tokens]
                                                          ────────────────────────────
                                                          Total: ~200k × N tasks

On paper: thorough. In practice: around 200,000 tokens per task. A five-task spec burned through a million tokens before I'd seen any integrated output.

The waste was structural. Setup phases that didn't vary between tasks ran N times. Per-task reviews could only see one task's changes — they missed integration problems anyway. Documentation ran in isolation before anything was assembled.

And the per-task review gave me false confidence: it caught issues within a task, but couldn't catch whether the integrated result actually worked. I was paying for isolation on tasks that were mostly sequential.

What I rebuilt

The redesign comes from a simpler question: what actually needs isolation, and what can be shared?

The implement step benefits from a fresh context per task — no bleed between unrelated changes. Everything else — setup, testing, review, documentation — can run once over the integrated result.

That produces a clean ladder of tools:

  /patch        trivial fix, no tests needed
      ↓
  /sdlc-task    one unit → implement → fast-test → fix → commit
      ↓
  /sdlc-run     one whole spec, full lifecycle, in-place
      ↓
  /sdlc-flow    one whole spec, worktree isolation, produces a PR
      ↓
  /sdlc-block   block-level orchestrator, branch train of PRs

Pick the rung that matches the scope. A trivial hotfix doesn't need a review stage. A whole feature spec does. Multiple independent feature blocks running in parallel need the orchestrator.

The biggest change is in the orchestrator. The original version orchestrated individual tasks. The rebuilt version operates at the block level — each block is a complete feature-sized unit of work that runs its own full pipeline in its own branch:

  master-plan.md
        │
  sdlc-block orchestrator
        │
    Phase 1  ─────────────────────── (parallel)
    ├── Block A  →  sdlc-flow  →  PR #1
    └── Block B  →  sdlc-flow  →  PR #2
        │
    (after Phase 1 merges)
        │
    Phase 2
    └── Block C  →  sdlc-flow  →  PR #3

And inside each sdlc-flow, the lean design runs one fresh implement agent per task, then one consolidated back-half over the integrated result:

  shared setup (once)
        │
    Task 1  →  fresh implement agent
    Task 2  →  fresh implement agent
    Task 3  →  fresh implement agent
        │
  test → review → fix → document → wrap-up
  (once, over the integrated result)

Per-task agent isolation, at roughly the cost of running it once.

But the pipeline redesign didn't fix my outputs

Faster and cheaper, yes. Better results, no.

What made the results better was fixing the specs.

I'd been optimizing execution while planning was still the bottleneck. The agents were reading underspecified tasks, making reasonable assumptions, and producing technically correct results that missed the actual intent. No amount of pipeline tuning changes that.

You can't fix a bad spec with a better pipeline.

What a planning artifact actually looks like

The most interesting thing I've done with this pipeline: I used it to redesign itself.

Here's a condensed version of the master plan I wrote before touching a line of code. The goal was to redesign all four SDLC engines — and the spec-driven pipeline was the tool I used to do it:

# SDLC Engines Redesign — Master Plan

## Goal

Four engines ship in the harness, but they no longer earn their place
as built. The most expensive one burns ~200k tokens per task running a
full review+test+document cycle that's mostly redundant when tasks are
sequential. Usage data: almost nobody runs it anymore — the simpler
single-spec runner has become the actual default.

This redesign gives each engine a distinct scope × ceremony tier,
gives every run a committed token-accounted state trail, and
rationalizes the planning commands that feed them.

## Architecture

| Engine       | Scope            | When to use                          |
|--------------|------------------|--------------------------------------|
| /patch       | Trivial hotfix   | No tests needed                      |
| /sdlc-task   | One small unit   | Implement → test → fix → commit      |
| /sdlc-run    | One whole spec   | Full lifecycle, in-place             |
| /sdlc-flow   | One whole spec   | Worktree isolation + PR              |
| /sdlc-block  | A full roadmap   | Block-level orchestration, PRs       |

---

## Phase 0 — Foundation

### Block A — Unified token telemetry
**What:** Every engine writes a committed state file with a token
usage block after each phase. Persists what was previously
render-only output that vanished when a run ended.

**Why:** Token costs were invisible between runs. You couldn't tell
which stage was expensive without watching the live output scroll by.

**Acceptance criteria:**
- sdlc-run writes a committed state file with a tokens roll-up
  after each phase
- sdlc-flow persists per-task token usage into its committed state
- No engine references a gitignored breadcrumb file
- node --check clean on both engines

### Block B — Lean single-unit engine
**What:** Rewrite the task-level engine into a lean single-unit
runner: implement → fast validation → fix loop (≤3 attempts) →
commit. Delete the heavy stages and coupling flags.

**Why:** Makes small work cheap enough to be worth a dedicated
engine, and gives the trivial /patch command an intermediate rung
instead of jumping straight to the full-spec runner.

**Acceptance criteria:**
- Engine runs only implement → test → fix → commit
- Coupling flags are gone (grep-clean)
- Committed state file with tokens is written
- node --check clean

---

## Phase 1 — The Headline Change

### Block A — Rewrite the orchestrator as a block-level engine
**What:** Replace the task-level wave machine with block-level
orchestration. Reads a master-plan file, computes dependency waves
at block granularity, fans out one sdlc-flow per block in its own
worktree branch, produces a PR per block by default. Delete the
legacy task-level execution engine and its orphaned schema file.

**Why:** The old design's merge-conflict failure mode was structural
— tasks sharing a worktree conflicting on shared files. Blocks are
independent by construction. This eliminates the failure mode
entirely, and reuses the proven single-spec runner as the inner
engine rather than duplicating its logic.

**Files:**
- Modified: sdlc-block.js — full rewrite (keep wave computation,
  config loader, traced-agent wrapper; add plan-file input,
  block-level fan-out, branch-train, two-level committed state)
- Deleted: execution-plan.schema.json — no remaining consumer
  after the task-level machine is removed

**Interfaces / shared surface:** Reads master-plan-format files
(the same format /generate-master-plan and /plan produce). Invokes
sdlc-flow as the inner engine via the workflow() primitive. The
committed block-orchestration-state.json — with child-flow token
roll-up — is the resume signal and the human review artifact.

**Out of scope:** The /review-PR and /merge-train commands for
human-gated branch-train merging (Block B). The per-block close-out
quality gate (Block C). The harness config schema rewrite for the
new block.* keys (Phase 3). This block may read provisional keys
and leave the schema update for Phase 3.

**Acceptance criteria:**
- Orchestrator reads a master-plan-format file and fans out
  sdlc-flow per independent block
- Default opens a PR per block; --auto-merge merges in
  dependency order
- Committed block-orchestration-state.json written with child
  token roll-up
- All legacy task-level code is gone (grep-clean for removed
  symbols: runTaskWorktree, runTaskInPlace, --from test,
  --verify-depth)
- node --check clean on the engine file

A few things worth noting about how this spec works.

The goal section captures why this is happening before describing what to build. "Usage data: almost nobody runs it" is load-bearing context — without that, an agent might preserve backward compatibility with a behavior nobody is actually using.

The Files field is a commitment, not a description. The agent knows exactly which files to touch and which to leave alone. Anything not listed is out of scope by default.

The Interfaces / shared surface field tells the agent what this block exposes to the blocks that come after it. That's the contract other blocks depend on — changing it mid-implementation breaks downstream work.

The Out of scope field does something most specs skip entirely: it names the things you are not building. This matters because an agent trying to be helpful will often implement adjacent things that seem obviously related. Explicit out-of-scope entries prevent that drift.

And the acceptance criteria are verifiable against the diff. Not "the engine works better" — "these specific symbols are gone, this file is committed, node --check passes." These become the exact checklist the review stage runs.

The practical test: Can you hand this spec to a smart engineer with zero prior context, and have them build the right thing? If not, the spec isn't done — and an AI agent will fail the same way.

Planning before generating

The flow that works for me now:

  Feature idea or requirement
          │
     /plan or /generate-master-plan
     (mini-roadmap: what, why, blocks, dependencies)
          │
     /generate-tasks
     (executable spec: tasks + acceptance criteria + decisions)
          │
     pick the right engine rung
          │
     AI agents implement

The planning step is where the hard questions happen: what does done actually look like, what are the real constraints, where are the edge cases. These are cheaper to answer before implementation starts than after.

When I rush through planning, I pay for it in review cycles. The root cause is almost always an underspecified task — and the fix is always "write a clearer spec," not "use a better model."

When I slow down at planning, implementations usually ship clean on the first pass.

The scarce skill

The scarce skill isn't writing code. It isn't prompting. It's writing specifications that fully capture intent — clear enough that a stateless agent can make the right call without asking follow-up questions.

This is hard. It requires thinking through your assumptions before you start. It requires distinguishing between "I have a vague sense of what I want" and "I can articulate what I want precisely enough for someone else to act on."

But it transfers. The same discipline that makes a good agent spec makes a better architecture decision record, a clearer PR description, a more useful design doc. It makes you a better collaborator regardless of whether the person on the other side is human or AI.

The tooling is changing fast. The underlying skill isn't.

Start with the spec.

If this was useful, I write about building production AI and agentic systems at learn-agentic-ai.com — including hands-on learning paths available in both English and Brazilian Portuguese. Come build something real.

Why Multi-Agent Systems Are a Trap (And What I Learned the Hard Way)

bredmond1019 — Tue, 23 Jun 2026 18:38:14 +0000

There's a moment in every ambitious AI engineering project where you convince yourself that more agents means more power. I hit that moment early in building my Python orchestration framework — and I spent several painful weeks learning exactly why that intuition is wrong.

The seductive pitch: decompose complex tasks into specialized sub-agents, run them in parallel, let them coordinate. What actually happened was a reliability nightmare that taught me more about agentic architecture than any framework documentation ever could.

The Problem I Actually Built

My Python orchestration system was designed to automate complex, multi-step workflows — the kind that require planning, research, code generation, and validation to happen in a coherent sequence. Early on, I structured it as a web of parallel agents: a planner, several workers, a validator, and a synthesizer, all exchanging structured messages.

On paper it was elegant. In practice, it had three failure modes I couldn't engineer away:

Context drift. Each agent only saw the slice of information it was handed. The worker writing one module couldn't see what the worker writing another module had decided. By the time the synthesizer tried to combine outputs, I had conflicting assumptions baked into the results — variable names that clashed, patterns that contradicted each other, interfaces that didn't align.

Cascading partial failures. When one agent produced ambiguous output, every downstream agent amplified the ambiguity. A planner that returned a slightly underspecified task description produced workers that each interpreted it differently. Nothing failed loudly. Everything just drifted, quietly, until the final output was incoherent.

Debugging opacity. When something went wrong in a parallel multi-agent system, tracing the failure was miserable. Was it the planner? One of the workers? The message-passing layer? I'd rebuilt the worst parts of distributed systems debugging inside a single Python process.

The Architecture Shift That Fixed It

The fix wasn't clever. It was humbling: I flattened the architecture.

Instead of parallel agents exchanging messages, I restructured the system around a single linear execution chain where each step receives the full accumulated context of everything that came before. The pattern looks like this:

import anthropic

client = anthropic.Anthropic()

def run_orchestrated_task(task: str) -> str:
    # Full context accumulates at each step
    context = {
        "task": task,
        "steps_completed": [],
        "artifacts": {}
    }

    # Each agent in the chain receives complete context
    for step_name, instruction in agent_steps:
        result = run_step(client, context, step_name, instruction)
        context["steps_completed"].append({
            "agent": step_name,
            "result": result
        })

    return context["artifacts"].get("final_output", "")

This single change eliminated context drift entirely. The implementer sees the planner's full reasoning. The validator sees every decision the implementer made. Nothing is hidden, nothing is re-interpreted from scratch.

Why Parallelism Is Usually the Wrong Bet

The appeal of parallel agents is speed. But in most agentic workflows, the bottleneck isn't compute — it's correctness. You're not waiting on CPU cycles; you're waiting on language model calls that need to produce coherent, consistent output.

Parallelism trades coherence for speed. For tasks where consistency matters — and in my experience, almost all production tasks require consistency — that's a bad trade.

The cases where I've found parallelism genuinely useful are narrow: independent sub-tasks with no shared state, clearly-bounded scope, and results that are combined by a human reviewer rather than automatically synthesized. Think: running the same analysis on ten separate documents and returning ten separate reports. Not: building one coherent system from ten parallel contributors.

The Context Engineering Insight

What changed my mental model was reframing the problem. I'd been thinking about agent architecture as a parallelism problem. It's actually a context problem.

The question isn't "how many agents can work at once?" It's "what does each agent need to know to make the right decision?"

When I started designing with that question first, the architecture became obvious. Every agent needs the full history of what was decided before it. That requirement naturally pushes you toward sequential execution — not because parallelism is impossible, but because the information dependencies make it impractical.

Here's the pattern I settled on for my orchestration framework:

import anthropic
from dataclasses import dataclass, field
from typing import Any

@dataclass
class ExecutionContext:
    original_task: str
    decisions: list[dict[str, Any]] = field(default_factory=list)
    artifacts: dict[str, str] = field(default_factory=dict)

    def record_decision(self, agent: str, decision: str, rationale: str):
        self.decisions.append({
            "agent": agent,
            "decision": decision,
            "rationale": rationale
        })

    def to_prompt_context(self) -> str:
        sections = [f"Task: {self.original_task}"]
        if self.decisions:
            sections.append("Decisions made so far:")
            for d in self.decisions:
                sections.append(
                    f"  [{d['agent']}] {d['decision']}\n"
                    f"    Rationale: {d['rationale']}"
                )
        return "\n\n".join(sections)


def run_step(
    ctx: ExecutionContext,
    agent_name: str,
    instruction: str
) -> str:
    client = anthropic.Anthropic()
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=ctx.to_prompt_context(),
        messages=[{"role": "user", "content": instruction}]
    )
    result = message.content[0].text
    ctx.record_decision(agent_name, result[:200], "see full output")
    return result

The key piece is to_prompt_context() — every agent's system prompt is rebuilt from the full decision history. Nothing is ever hidden.

What I'd Tell Myself Earlier

If I were starting the orchestration framework again, I'd establish these constraints from the beginning:

One writer at a time. Only one agent writes or modifies shared artifacts at a given moment. Research agents, planning agents, and analysis agents can run in parallel if their outputs are truly independent. But anything touching a shared artifact runs sequentially.

Explicit state over implicit message passing. Instead of agents exchanging messages, agents read from and write to an explicit execution context. The state is the contract. There's no ambiguity about what the next agent knows.

Fail loudly or not at all. If an agent can't complete its step with the context it has, it should raise an exception rather than produce ambiguous output. Silent degradation is the enemy of debuggable systems.

Sequential is not slow. The processing time for a ten-step agent chain is dominated by the LLM call latency, not the orchestration overhead. Parallelism rarely moves the needle in ways users notice.

The Reliability Payoff

After flattening the architecture, the system became boring in the best way. Failures were rare, and when they happened, they were traceable. The planner's output was in the log. The implementer's reasoning was in the log. The validator's critique was in the log. I could read the execution history like a transcript and immediately see where things went wrong.

That debugging experience is itself evidence for the architecture. If you can't read what your agents did and understand why, you've built something too complex to maintain.

Practical Next Steps

If you're building agentic systems and finding reliability hard to achieve:

Flatten before scaling. Get a single-chain architecture working reliably before adding any parallelism.
Serialize your context explicitly. Write the to_prompt_context() function and think carefully about what every agent actually needs to know.
Log decisions, not just outputs. The rationale behind each decision is what makes the system debuggable.
Test failure modes deliberately. Give agents ambiguous inputs and watch what cascades. That's where your architecture's assumptions are hiding.

The multi-agent dream is real, but it's further away than the frameworks suggest. The path there runs through reliable single-chain architectures that you actually understand — not through distributed complexity that looks impressive until it doesn't.

The 7 Building Blocks of Reliable AI Agents (Skip the Frameworks)

bredmond1019 — Tue, 23 Jun 2026 18:38:13 +0000

When I started building multi-agent systems, I made the mistake most engineers make: I went looking for the right framework. I evaluated LangChain, AutoGPT, and a dozen other options. I built prototypes in each. I spent weeks on tooling instead of solving the actual problem.

The moment things clicked was when I stripped everything back to bare API calls and asked: what does every agent I've shipped actually need? The answer wasn't a framework. It was seven building blocks — patterns that show up in every reliable agentic system I've built, from the Python orchestration framework that drives parallel task execution to the SDLC agentic harness that ships production code.

Here's what I've learned from building and operating these systems: the best AI agents aren't "agentic" in the way the demos make them look. They're mostly deterministic code with surgical, well-scoped calls to language models.

The Two Types of AI Systems (And Why It Matters)

Before the building blocks, this distinction matters enormously.

AI Assistants (Human-in-the-Loop): Users interact directly, can correct mistakes in real time, and expect exploration. Multiple LLM calls make sense. Think pair programming with Cursor, or a chat interface.

Background Automation (No Human Oversight): Errors compound silently. Latency and cost are magnified. Every LLM call needs justification. This is the domain of production agents.

My Python orchestration framework and the support-automation system I built both fall into the second category. They run without a human watching every step. That constraint shaped every architectural decision — and it's why these seven building blocks matter.

# Background automation: minimize and isolate LLM calls
def process_support_ticket(ticket_data: dict) -> dict:
    # Deterministic routing first — no LLM needed here
    if ticket_data["type"] == "refund" and ticket_data["amount"] < 50:
        return handle_refund_automatically(ticket_data)

    # Use LLM only where reasoning is genuinely required
    if requires_classification(ticket_data):
        context = build_classification_context(ticket_data)
        classification = classify_with_llm(context)
        return route_by_classification(classification, ticket_data)

    return standard_response_template(ticket_data)

Most of what I ship is the outer shell — deterministic orchestration code. The LLM handles the reasoning that can't be hard-coded.

Building Block 1: The Intelligence Layer

This is the only actual AI component. Everything else is software engineering.

I use the Anthropic SDK directly — no intermediate abstractions. The API surface is small and stable. Every layer on top is a maintenance liability.

import anthropic

client = anthropic.Anthropic()

def get_llm_response(prompt: str) -> str:
    """The atomic unit of intelligence: a focused, scoped LLM call."""
    message = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return message.content[0].text

The hard part isn't the API call. It's everything around it: what you put in the prompt, how you validate what comes back, and how you handle failure. The remaining six building blocks cover exactly that.

Building Block 2: Memory (State Management)

Language models are stateless. They know nothing about previous calls unless you tell them. Memory management is your responsibility entirely.

In the Python orchestration framework, I maintain context across multi-step pipelines explicitly:

def chat_with_memory(conversation_history: list, new_message: str) -> tuple[str, list]:
    """Maintain conversation state across orchestration steps."""
    conversation_history.append({"role": "user", "content": new_message})

    message = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1024,
        messages=conversation_history
    )

    assistant_message = message.content[0].text
    conversation_history.append({
        "role": "assistant",
        "content": assistant_message
    })

    # Keep context window bounded — trim oldest turns when needed
    if len(conversation_history) > 20:
        conversation_history = conversation_history[-20:]

    return assistant_message, conversation_history

In production, conversation history lives in the database, not in RAM. Each message carries a session ID and timestamp. Token counts are tracked explicitly so you can manage context windows before hitting limits — not after.

Note:
In the SDLC agentic harness, each task worktree maintains its own context: the spec file, a running implementation report, and git history. The "memory" is the files on disk, not an in-memory list. This makes every step auditable and resumable.

Building Block 3: Tools (External Integration)

Tools let an agent take action in the world: read a file, call an API, query a database. They're powerful — and expensive. Every tool call is at least two LLM round trips.

The support-automation system I built uses tools conservatively. The vast majority of tickets route deterministically. Tools activate only when the agent genuinely can't proceed without external information.

tools = [
    {
        "name": "lookup_account_status",
        "description": "Look up a customer account's current status and plan details.",
        "input_schema": {
            "type": "object",
            "properties": {
                "account_id": {
                    "type": "string",
                    "description": "The customer account identifier"
                }
            },
            "required": ["account_id"]
        }
    }
]

def agent_with_tools(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages
    )

    # Process tool use if the model requested it
    if response.stop_reason == "tool_use":
        tool_use = next(b for b in response.content if b.type == "tool_use")
        tool_result = execute_tool(tool_use.name, tool_use.input)

        messages.extend([
            {"role": "assistant", "content": response.content},
            {
                "role": "user",
                "content": [{"type": "tool_result",
                             "tool_use_id": tool_use.id,
                             "content": tool_result}]
            }
        ])

        final = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        return final.content[0].text

    return response.content[0].text

The rule I follow: if a tool call can be replaced with a pre-query before the LLM sees the prompt, do that instead. Cheaper, faster, and deterministic.

Building Block 4: Validation (Structured Output)

Unstructured LLM output is a liability in production systems. The fix is forcing structured responses and validating them before downstream code touches them.

Anthropic's API supports structured output natively via tool use patterns. I define schemas with Pydantic, which gives me type safety and validation logic in one place:

from pydantic import BaseModel
from typing import Literal
import json

class TicketClassification(BaseModel):
    category: Literal["billing", "technical", "account", "general"]
    urgency: Literal["high", "medium", "low"]
    requires_human: bool
    summary: str

def classify_ticket(ticket_text: str) -> TicketClassification:
    """Structured classification — always returns a validated object."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        system=(
            "You are a support ticket classifier. "
            "Return a JSON object with these exact fields: "
            "category (billing/technical/account/general), "
            "urgency (high/medium/low), "
            "requires_human (boolean), "
            "summary (one sentence). Return only JSON."
        ),
        messages=[{"role": "user", "content": ticket_text}]
    )

    raw = response.content[0].text
    data = json.loads(raw)
    return TicketClassification(**data)

In the SDLC harness, every implement step returns a structured report. The orchestrator reads that report as data, not as prose. When an agent produces output that doesn't match the expected shape, it fails fast and clearly — not silently downstream.

→
Structured output is the single highest-leverage reliability improvement I've made to agentic systems. It converts probabilistic text generation into deterministic data extraction that downstream code can depend on.

Building Block 5: Control Flow (Deterministic Routing)

This is the pattern that made the biggest difference when I productionized the support-automation system. Instead of handing control to the LLM, I use the LLM for classification and then let deterministic code do the routing.

def handle_support_request(ticket: dict) -> str:
    """One LLM call for classification; deterministic routing after."""
    classification = classify_ticket(ticket["body"])

    # Log the classification — this is invaluable for debugging
    print(f"Category: {classification.category}")
    print(f"Urgency: {classification.urgency}")
    print(f"Requires human: {classification.requires_human}")

    # Deterministic routing — no more LLM calls for most paths
    if classification.requires_human or classification.urgency == "high":
        return escalate_to_human_queue(ticket, classification)

    if classification.category == "billing":
        return handle_billing_ticket(ticket, classification)

    if classification.category == "technical":
        return handle_technical_ticket(ticket, classification)

    return generate_general_response(ticket, classification)

The key insight: the LLM's job is to understand the input, not to drive the execution. Once the intent is classified and the confidence is above threshold, the rest is a state machine.

In the SDLC harness, this pattern runs end-to-end: each pipeline stage classifies what the previous agent produced (PASS / FAIL / PARTIAL), then routes deterministically — implement, test, review, document, or retry. No agent decides its own next step.

Building Block 6: Recovery (Error Handling)

Production agents fail. APIs time out, rate limits hit, models return malformed output. The question isn't whether failures happen — it's whether your system degrades gracefully.

The Python orchestration framework uses a circuit-breaker pattern combined with exponential backoff. Individual task failures don't cascade:

import time
import logging
from typing import Optional

def resilient_llm_call(
    prompt: str,
    max_retries: int = 3,
    backoff_base: float = 2.0
) -> Optional[str]:
    """Production-ready call with retry and graceful fallback."""
    for attempt in range(max_retries):
        try:
            message = client.messages.create(
                model="claude-haiku-4-5-20251001",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            )
            result = message.content[0].text
            if not result or len(result.strip()) < 10:
                raise ValueError("Response too short — likely an error state")
            return result

        except Exception as e:
            logging.warning(
                f"LLM call failed (attempt {attempt + 1}/{max_retries}): {e}"
            )
            if attempt < max_retries - 1:
                time.sleep(backoff_base ** attempt)
            else:
                logging.error(f"All {max_retries} attempts exhausted")
                return None

    return None

def process_with_fallback(input_data: dict) -> str:
    """Graceful degradation when the LLM path fails."""
    ai_result = resilient_llm_call(build_prompt(input_data))

    if ai_result:
        return ai_result

    # Rule-based fallback — always have one before you ship
    logging.info("Falling back to rule-based processing")
    return rule_based_response(input_data)

Warning:
Build the fallback path before you build the AI path. If you can't describe what your system does when the LLM fails, you're not ready to ship it.

Building Block 7: Human Oversight

Not every decision should be fully automated. For operations with significant consequences — sending a message to a customer, modifying critical data, triggering an irreversible action — a human checkpoint is worth the latency.

The SDLC harness implements this at the review stage: after an agent implements and tests a task, a review step evaluates the result against acceptance criteria. Clean passes continue automatically; ambiguous or failed results either retry (automated) or escalate to a human. Automation handles scale; humans handle accountability.

from enum import Enum
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"

@dataclass
class ApprovalRequest:
    id: str
    content: str
    context: dict
    status: ApprovalStatus = ApprovalStatus.PENDING
    created_at: datetime = field(default_factory=datetime.now)

def generate_with_approval(
    prompt: str,
    require_approval: bool = True
) -> Optional[str]:
    """Generate content with an optional human approval gate."""
    draft = resilient_llm_call(prompt)

    if not draft:
        return None

    if not require_approval:
        return draft

    # In production: store the request, notify via webhook/Slack, await response
    request = ApprovalRequest(
        id=generate_id(),
        content=draft,
        context={"prompt": prompt}
    )
    store_approval_request(request)
    notify_approver(request)

    return poll_for_approval(request.id, timeout_seconds=300)

In the support-automation system, high-urgency escalations feed into a human review queue. The agent drafts a response; a team member approves or edits before it sends. The automation provides velocity; the human provides accountability where it matters.

Putting It All Together

These building blocks combine into a complete production loop:

class ReliableAgent:
    def __init__(self):
        self.conversation_memory: dict[str, list] = {}
        self.error_counts: dict[str, int] = {}

    def process(self, session_id: str, input_data: dict) -> str:
        try:
            # Block 2: Load conversation context
            history = self.conversation_memory.get(session_id, [])

            # Block 4 + 5: Classify with structured output, route deterministically
            classification = classify_ticket(input_data["body"])

            if classification.requires_human:
                # Block 7: Human oversight for critical paths
                return self.escalate_with_draft(input_data, classification)

            if classification.urgency == "high":
                # Block 3: Tool use only when needed
                account_info = fetch_account_details(input_data["account_id"])
                response = self.generate_with_context(
                    input_data, classification, account_info, history
                )
            else:
                # Block 1: Direct LLM call for routine responses
                response = resilient_llm_call(
                    build_prompt(input_data, classification, history)
                )

            # Block 2: Update memory
            history.append({"role": "user", "content": input_data["body"]})
            history.append({"role": "assistant", "content": response})
            self.conversation_memory[session_id] = history[-10:]

            self.error_counts[session_id] = 0
            return response

        except Exception as e:
            # Block 6: Recovery
            count = self.error_counts.get(session_id, 0) + 1
            self.error_counts[session_id] = count

            if count > 5:
                return "System temporarily unavailable. A team member will follow up."

            return rule_based_response(input_data)

What This Architecture Makes Possible

The support-automation system I built processes a large volume of daily requests with this pattern. What I can tell you from operating it: the deterministic routing layer handles the majority of requests without any LLM call at all. Only the ones requiring genuine language understanding — nuanced complaints, ambiguous multi-part questions, edge cases the rules don't cover — involve a model call.

That constraint — use LLM calls only where they earn their cost — forced better system design across the board. The classification layer got sharper because I had to define what "requires reasoning" actually means. The fallback paths got more complete because I had to think about every failure mode. The observability layer became essential because I needed to see which paths were being taken and why.

The SDLC agentic harness runs the same way. Each pipeline stage is mostly deterministic orchestration code. The agents are scoped to specific reasoning tasks: write implementation, evaluate test results, review against criteria. Everything else is file I/O, git operations, and control flow.

The Seven Building Blocks

To build a reliable production agent, you need all seven:

Intelligence: Direct API calls to the language model — scoped, purposeful, minimal
Memory: Explicit state management — conversation history, session context, persistent storage
Tools: External integrations — used sparingly, only where deterministic pre-fetching won't do
Validation: Structured output with schema enforcement — convert text generation into data extraction
Control Flow: LLM for classification, deterministic code for routing — never let the model drive execution
Recovery: Retry logic, exponential backoff, graceful degradation — always have a fallback
Human Oversight: Approval checkpoints for high-consequence operations — automation handles scale, humans handle accountability

You don't need a framework. You need these patterns, implemented cleanly, with the Anthropic SDK directly.

Start with one workflow. Map the classification logic. Build the deterministic routing. Add an LLM call only where you genuinely can't route without it. The framework will emerge from the patterns — and it will be one you understand completely.

→
If you're evaluating frameworks right now, pause. Pick one real workflow in your system and write it using just these seven building blocks with direct SDK calls. The clarity you get from that exercise will shape every architectural decision after it.

Multi-Agent Observability: See Everything Your AI Agents Do

bredmond1019 — Tue, 23 Jun 2026 18:38:11 +0000

Once I had three agents running in parallel, I lost the thread. I couldn't tell which one was waiting on me, which had stalled on a bad tool call, or why the final output came back missing a piece.

The problem wasn't the agents — it was that I had no visibility into what any of them were actually doing. Each one was a black box unless I stopped everything and read its terminal.

Here's the setup I built to fix that: Claude Code hooks feeding a minimal event server, so you can see what every agent is doing in real time — across 3, 5, 10 instances at once.

The Problem: Too Many Agents, Too Little Visibility

When I'm running the SDLC harness with tasks in parallel, the setup looks something like this:

One agent implementing a new module
Another reviewing the previous task's output
A third running the validation gates
Two more doing research on different parts of the codebase

Without observability, you're flying blind. Which agent needs your input? What are they actually doing? When something goes wrong, how do you trace it back?

The Solution: Real-Time Multi-Agent Observability

Here's what we're building:

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Claude Agent 1 │     │  Claude Agent 2 │     │  Claude Agent 3 │
│   (App: CRM)    │     │ (App: API Docs) │     │ (App: Testing)  │
└────────┬────────┘     └────────┬────────┘     └────────┬────────┘
         │                       │                       │
         │    Claude Hooks       │    Claude Hooks       │
         └───────────┬───────────┴───────────┬──────────┘
                     │                       │
                     ▼                       ▼
              ┌─────────────────────────────────┐
              │        BUN SERVER               │
              │   • Store to SQLite             │
              │   • Broadcast via WebSocket     │
              └──────────────┬──────────────────┘
                             │
                             ▼
              ┌─────────────────────────────────┐
              │     REAL-TIME DASHBOARD         │
              │   • Live Activity Pulse         │
              │   • Event Stream                │
              │   • AI Summaries               │
              └─────────────────────────────────┘

Key Features:

Live Activity Pulse: Visual representation of all agent activities
Event Stream: Every tool call, hook, and decision
AI-Powered Summaries: Understand at a glance what each agent is doing
Session Tracking: Color-coded agents for easy identification

Building the Observability System

Step 1: Enhanced Hook Configuration

First, we upgrade our hooks to send comprehensive event data:

#!/usr/bin/env python3
# ~/.claude/hooks/send-event.py

import sys
import json
import requests
import os
from datetime import datetime

def summarize_with_ai(event_data, event_type):
    """Use a small, fast model to summarize the event"""
    if event_type not in ['pre-tool-use', 'post-tool-use']:
        return None

    try:
        # Use Haiku for ultra-fast summaries
        import anthropic
        client = anthropic.Client()

        prompt = f"Summarize in 10 words what this {event_type} event does: {json.dumps(event_data)}"

        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=30,
            messages=[{"role": "user", "content": prompt}]
        )

        return response.content[0].text
    except:
        return None

def send_event(app_name, event_type, summarize=True):
    """Send event to observability server"""
    event = json.loads(sys.stdin.read())

    # Add metadata
    event['app_name'] = app_name
    event['event_type'] = event_type
    event['timestamp'] = datetime.now().isoformat()
    event['session_id'] = os.environ.get('CLAUDE_SESSION_ID', 'unknown')

    # Add AI summary if requested
    if summarize:
        event['summary'] = summarize_with_ai(event, event_type)

    # Send to server
    try:
        requests.post('http://localhost:3000/events', 
                     json=event, 
                     timeout=1)
    except:
        pass  # Don't block on failures

    sys.exit(0)

# Usage: send-event.py <app-name> <event-type> [--summarize]
if __name__ == "__main__":
    app_name = sys.argv[1]
    event_type = sys.argv[2]
    summarize = '--summarize' in sys.argv
    send_event(app_name, event_type, summarize)

Step 2: Configure Hooks for All Events

# ~/.claude/settings.yaml
hooks:
  pre-prompt:
    - ~/.claude/hooks/send-event.py "my-app" "pre-prompt"

  pre-tool-use:
    - ~/.claude/hooks/security-check.py
    - ~/.claude/hooks/send-event.py "my-app" "pre-tool-use" --summarize

  post-tool-use:
    - ~/.claude/hooks/send-event.py "my-app" "post-tool-use" --summarize

  notification:
    - ~/.claude/hooks/send-event.py "my-app" "notification"
    - ~/.claude/hooks/play-sound.py

  stop:
    - ~/.claude/hooks/send-event.py "my-app" "stop"
    - ~/.claude/hooks/voice-summary.py

  pre-compact:
    - ~/.claude/hooks/send-event.py "my-app" "pre-compact"

Step 3: Build the Event Server

// server.ts
import { serve } from 'bun';
import { Database } from 'bun:sqlite';

const db = new Database('events.db');
db.run(`
  CREATE TABLE IF NOT EXISTS events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp TEXT,
    app_name TEXT,
    session_id TEXT,
    event_type TEXT,
    data TEXT
  )
`);

const clients = new Set<WebSocket>();

serve({
  port: 3000,

  async fetch(req, server) {
    const url = new URL(req.url);

    // Handle event ingestion
    if (url.pathname === '/events' && req.method === 'POST') {
      const event = await req.json();

      // Store in database
      db.run(
        'INSERT INTO events (timestamp, app_name, session_id, event_type, data) VALUES (?, ?, ?, ?, ?)',
        event.timestamp,
        event.app_name,
        event.session_id,
        event.event_type,
        JSON.stringify(event)
      );

      // Broadcast to all connected clients
      const message = JSON.stringify({
        type: 'event',
        data: event
      });

      clients.forEach(client => {
        if (client.readyState === WebSocket.OPEN) {
          client.send(message);
        }
      });

      return new Response('OK');
    }

    // Upgrade to WebSocket for real-time updates
    if (url.pathname === '/ws') {
      if (server.upgrade(req)) {
        return;
      }
    }

    return new Response('Not found', { status: 404 });
  },

  websocket: {
    open(ws) {
      clients.add(ws);
      console.log('Client connected');
    },

    close(ws) {
      clients.delete(ws);
      console.log('Client disconnected');
    },

    message(ws, message) {
      // Handle client messages if needed
    }
  }
});

console.log('Observability server running on http://localhost:3000');

Step 4: Create the Real-Time Dashboard

// useWebSocketEvents.ts
import { ref, onMounted, onUnmounted } from 'vue';

export interface AgentEvent {
  timestamp: string;
  app_name: string;
  session_id: string;
  event_type: string;
  summary?: string;
  data: any;
}

export function useWebSocketEvents() {
  const events = ref<AgentEvent[]>([]);
  const isConnected = ref(false);
  let ws: WebSocket | null = null;

  const connect = () => {
    ws = new WebSocket('ws://localhost:3000/ws');

    ws.onopen = () => {
      isConnected.value = true;
      console.log('Connected to observability server');
    };

    ws.onmessage = (event) => {
      const message = JSON.parse(event.data);
      if (message.type === 'event') {
        // Add to events array (limit to last 1000)
        events.value = [message.data, ...events.value].slice(0, 1000);
      }
    };

    ws.onclose = () => {
      isConnected.value = false;
      // Reconnect after 1 second
      setTimeout(connect, 1000);
    };
  };

  onMounted(connect);

  onUnmounted(() => {
    ws?.close();
  });

  return {
    events,
    isConnected
  };
}

Advanced Observability Features

1. Live Activity Pulse

Visualize agent activity over time with a pulse chart that shows activity intensity and which agents are most active.

2. Smart Event Filtering

Filter events by:

Application name
Event type (pre-tool-use, post-tool-use, etc.)
Session ID
Time range
Search query

3. Session-Based Color Coding

Each agent session gets a unique color based on its session ID, making it easy to track individual agents visually.

Practical Patterns

Pattern 1: Agent Health Monitoring

Detect when agents get stuck or stop responding by tracking the time since their last event.

Pattern 2: Cross-Agent Coordination

Track when multiple agents are working on the same files to prevent conflicts.

Pattern 3: Performance Analytics

Measure agent performance with metrics like:

Total events per session
Tools used
Average response time
Error rate
AI-generated summaries

Note:
Pro Tip: Use small, fast models like claude-3-haiku-20240307 for event summarization. The summaries are generated quickly and stay out of the critical path, so agent throughput stays high.

Scaling Considerations

As you scale from 3 agents to 30:

1. Event Sampling

For high-frequency events, sample rather than log everything:

if (Math.random() < 0.1) { // 10% sampling
  sendEvent(data);
}

2. Batch Processing

Send events in batches to reduce network overhead:

const eventBatch = [];
const flushBatch = () => {
  if (eventBatch.length > 0) {
    sendBatch(eventBatch);
    eventBatch.length = 0;
  }
};
setInterval(flushBatch, 1000); // Flush every second

3. Data Retention

Implement automatic cleanup:

DELETE FROM events WHERE timestamp < datetime('now', '-7 days');

The Power of Visibility

With multi-agent observability in place, you can:

Scale Confidently: Run 10+ agents without losing track
Debug Quickly: Trace issues back to specific agents and actions
Optimize Workflows: Identify bottlenecks and inefficiencies
Prevent Conflicts: Detect when agents step on each other's toes
Measure Impact: Quantify what your agents actually accomplish

→
Real Impact: Once observability is in place, you can confidently hand off work to multiple agents in parallel — you trust them because you can see everything they're doing, not because you're hoping for the best.

Getting Started

Start Simple: Begin with basic event logging to a file
Add Real-Time: Implement WebSocket broadcasting
Build the Dashboard: Start with a simple event list, add visualizations
Scale Gradually: Add more agents as your observability improves

Remember: If you don't measure it, you can't improve it. If you don't monitor it, how will you know what's actually happening?

The future of engineering is multi-agent systems. The key to multi-agent systems is observability. Build it once, scale it forever.

Your agents are working hard. It's time you could see everything they do.

Navigating Claude Code Rate Limits: The Art of Model Selection and Strategic Diversification

bredmond1019 — Tue, 05 Aug 2025 13:14:36 +0000

The notification hit like a cold shower: "New rate limits coming August 28th." For those of us who've been pushing Claude Code to its limits with 24/7 agent workflows and parallel processing, this was our wake-up call. We'd become "clawholics" – addicted to the most powerful agentic coding tool available. But with great power comes... rate limits.

The Three-Model Revolution

Claude Code just shipped one of its most strategically important features: sub-agent model selection. We now have access to three distinct models for our agents:

Haiku 3.5: The sprinter – fast, cheap, perfect for simple tasks
Sonnet 4: The workhorse – balanced performance for everyday coding
Opus 4: The powerhouse – maximum intelligence for critical work

This isn't just about having options; it's about solving two fundamental problems that plague AI-assisted development:

Model Overkill: Wasting expensive Opus tokens on tasks that Haiku could handle
Model Underperformance: Using weak models for complex tasks and having to redo work

The Performance-Speed-Cost Triangle

Let me share a real example. I recently ran a crypto research system with 12 parallel agents – 4 different analysis types, each running on all three model tiers. The results were illuminating:

Task: Generate formatted crypto analysis report

Haiku 3.5:
- Time: 8 seconds
- Tokens: ~2,000
- Format Accuracy: 60%
- Cost: $0.03

Sonnet 4:
- Time: 45 seconds  
- Tokens: ~8,000
- Format Accuracy: 85%
- Cost: $0.24

Opus 4:
- Time: 2 minutes
- Tokens: ~15,000
- Format Accuracy: 95%
- Cost: $1.20

The lesson? Not every nail needs a sledgehammer.

Smart Model Selection Patterns

Here's my framework for choosing the right model:

Use Haiku When:

Generating file names or simple identifiers
Basic summarization tasks
Quick file operations or migrations
Any task with clear, deterministic outcomes

Use Sonnet When:

General coding tasks
Refactoring existing code
Building standard features
Tasks requiring good-but-not-perfect accuracy

Use Opus (+ Thinking) When:

Architecting complex systems
Production-critical code
Tasks requiring deep reasoning
When accuracy trumps everything else

The ABC Testing Pattern

One powerful pattern I've discovered is ABC testing your prompts across model tiers. Here's how:

Create a dedicated agent prompt file
Reference it in three different agents (Haiku, Sonnet, Opus)
Run the same task across all three
Analyze results for the optimal model choice

# crypto_analyzer_agent_prompt.md
You are a cryptocurrency analysis expert...
[prompt details]

# In your slash command:
agents:
  - name: "crypto-haiku"
    model: "haiku-3.5"
    prompt: "@agent_prompts/crypto_analyzer_agent_prompt.md"

  - name: "crypto-sonnet"
    model: "sonnet-4"
    prompt: "@agent_prompts/crypto_analyzer_agent_prompt.md"

  - name: "crypto-opus"
    model: "opus-4"
    prompt: "@agent_prompts/crypto_analyzer_agent_prompt.md"

The Diversification Imperative

Here's the uncomfortable truth: we're overexposed to Claude Code. Like investors with too much capital in a single stock, we face concentration risk. The rate limits aren't unfair – they're a reality check.

Consider these alternatives entering the arena:

Qwen3 Coder: 480 billion parameters of open-source power
Gemini CLI: Google's rapidly improving alternative
Open-source options: A growing ecosystem of tools

This isn't about abandoning Claude Code – it's about being prepared. The landscape is evolving rapidly, and the winners will be those who master principles over tools.

Practical Rate Limit Strategies

With the new limits affecting the top 5% of users (that's probably you if you're reading this), here's how to adapt:

Reserve Opus for Critical Path: Only use maximum compute when it directly impacts outcomes
Implement Caching: Don't regenerate what you've already computed
Batch Operations: Group similar tasks to minimize overhead
Time-Shift Workloads: Run heavy operations during off-peak hours
Monitor Usage: Track your consumption patterns to avoid surprises

The Hidden Fourth Dimension

Beyond performance, speed, and cost, there's a fourth dimension: availability. Rate limits introduce scarcity to what felt like infinite compute. This changes everything about how we architect our systems.

Instead of throwing Opus at every problem, we need to be strategic:

Use Haiku for exploration and prototyping
Graduate to Sonnet for implementation
Reserve Opus for final validation and critical decisions

Looking Ahead

Anthropic's mention of "supporting long-running use cases through other options" hints at what's coming. My prediction? Dedicated compute tiers for serious users, similar to reserved instances in cloud computing.

But until then, we adapt. The principles remain constant:

Context: What information does the model need?
Model: What level of intelligence does the task require?
Prompt: How clearly can we communicate the task?

Master these, and you'll thrive regardless of rate limits or tool changes.

Action Steps

Audit Your Current Usage: Which tasks are consuming the most Opus tokens unnecessarily?
Create Model Selection Guidelines: Document which models to use for different task types
Build Fallback Systems: What happens when you hit rate limits mid-workflow?
Explore Alternatives: Spend 20% of your time with competing tools
Focus on Principles: Tools will change; principles endure

The Bottom Line

Rate limits aren't the enemy – they're a forcing function for better engineering. By matching model capabilities to task requirements, we can do more with less. The era of unlimited compute was always temporary; the era of intelligent compute allocation is just beginning.

Yes, we're addicted to Claude Code. But like any good investor, it's time to diversify our portfolio. The future belongs to engineers who can leverage the best tool for each job, not those who depend on a single solution.

Stay focused. Keep building. But build smarter.

Remember: In the attention economy, compute is currency. Spend it wisely.

Watch the Full Video

For a deep dive into these concepts with live demonstrations and detailed examples, check out the original video by IndyDevDan:

📹 I'm ADDICTED to Claude Code: RATE LIMITS, Agent Models, and CC Alternatives

In this 24-minute video, Dan demonstrates:

Running 12 parallel agents across Haiku, Sonnet, and Opus models
Real-world crypto research system implementation
Live examples of model performance differences
Detailed rate limit strategies and workarounds
Future predictions for agentic coding tools

This article was originally published on learn-agentic-ai.com

Multi-Agent Orchestration: Running 10+ Claude Instances in Parallel (Part 3)

bredmond1019 — Sat, 02 Aug 2025 22:06:46 +0000

Last Tuesday at 3 AM, I watched 12 Claude agents rebuild my entire frontend while I slept. One agent refactored components, another wrote tests, a third updated documentation, and a fourth optimized performance.

By morning, I had a pull request with 10,000+ lines of perfectly coordinated changes.

This isn't science fiction. This is multi-agent orchestration with Claude Code, and it's changing how we build software at scale.

The Multi-Agent Revolution

In Parts 1 and 2, we explored Claude's capabilities and hook system. Now, let's tackle the ultimate productivity multiplier: running multiple Claude instances in parallel.

But first, a warning: This is where things get complex. Multiple agents mean:

Resource contention
File conflicts
Coordination challenges
Observability nightmares

Get it wrong, and you'll have chaos. Get it right, and you'll achieve superhuman productivity.

The Architecture That Makes It Possible

Here's the system architecture I use for multi-agent orchestration:

┌─────────────────────────────────────────────┐
│            Orchestrator (Meta-Agent)         │
│         Decides what needs to be done        │
└──────────────────┬──────────────────────────┘
                   │ Creates tasks
                   ▼
┌─────────────────────────────────────────────┐
│              Task Queue (Redis)              │
│         Stores and distributes work          │
└─────┬───────┬───────┬───────┬──────────────┘
      │       │       │       │
      ▼       ▼       ▼       ▼
┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent 1 │ │ Agent 2 │ │ Agent 3 │ │ Agent N │
│Frontend │ │ Backend │ │  Tests  │ │  Docs   │
└─────────┘ └─────────┘ └─────────┘ └─────────┘
      │       │       │       │
      └───────┴───────┴───────┘
                   │
                   ▼
        ┌──────────────────┐
        │  Observability   │
        │    Dashboard     │
        └──────────────────┘

Step 1: The Meta-Agent Orchestrator

The meta-agent is Claude running in a special mode where it doesn't write code - it manages other agents:

# orchestrator.py
import json
import redis
import subprocess
from typing import List, Dict

class MetaAgent:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379, db=0)
        self.task_queue = 'claude_tasks'

    def analyze_project(self, requirements: str) -> List[Dict]:
        """Use Claude to break down requirements into parallel tasks"""

        prompt = f"""
        Analyze these requirements and break them into independent tasks
        that can be executed in parallel by specialized agents:

        {requirements}

        Return a JSON array of tasks with:
        - id: unique identifier
        - type: frontend|backend|testing|docs|refactor
        - description: what needs to be done
        - dependencies: array of task IDs that must complete first
        - files: array of files this task will modify
        """

        # Call Claude API
        response = self.call_claude(prompt)
        return json.loads(response)

    def distribute_tasks(self, tasks: List[Dict]):
        """Queue tasks for worker agents"""

        # Sort by dependencies
        sorted_tasks = self.topological_sort(tasks)

        for task in sorted_tasks:
            # Check dependencies
            if self.dependencies_complete(task):
                self.redis.lpush(self.task_queue, json.dumps(task))
            else:
                # Queue for later
                self.redis.lpush(f"{self.task_queue}:pending", json.dumps(task))

    def spawn_worker_agents(self, count: int):
        """Launch Claude worker agents"""

        for i in range(count):
            subprocess.Popen([
                'claude-code',
                '--mode', 'worker',
                '--id', f'agent-{i}',
                '--config', 'worker-config.json'
            ])

Step 2: Specialized Worker Agents

Each worker agent has a specific role and configuration:

# worker_agent.py
import os
import json
import redis
import time

class WorkerAgent:
    def __init__(self, agent_id: str, specialization: str):
        self.id = agent_id
        self.specialization = specialization
        self.redis = redis.Redis(host='localhost', port=6379, db=0)

    def run(self):
        """Main worker loop"""

        while True:
            # Get task from queue
            task_data = self.redis.brpop('claude_tasks', timeout=5)

            if task_data:
                task = json.loads(task_data[1])

                # Check if this agent can handle the task
                if self.can_handle(task):
                    self.execute_task(task)
                else:
                    # Put it back for another agent
                    self.redis.lpush('claude_tasks', task_data[1])
                    time.sleep(1)

    def execute_task(self, task: Dict):
        """Execute a task with Claude"""

        # Acquire file locks
        locked_files = self.acquire_locks(task['files'])

        try:
            # Set up Claude context
            prompt = self.build_prompt(task)

            # Execute with Claude
            os.environ['CLAUDE_SESSION_ID'] = f"{self.id}-{task['id']}"
            result = self.run_claude(prompt)

            # Report completion
            self.redis.hset(f"task:{task['id']}", 'status', 'complete')
            self.redis.hset(f"task:{task['id']}", 'result', result)

            # Trigger dependent tasks
            self.trigger_dependencies(task['id'])

        finally:
            # Release locks
            self.release_locks(locked_files)

    def acquire_locks(self, files: List[str]) -> List[str]:
        """Acquire exclusive locks on files"""

        locked = []
        for file_path in files:
            lock_key = f"lock:{file_path}"

            # Try to acquire lock with timeout
            if self.redis.set(lock_key, self.id, nx=True, ex=300):
                locked.append(file_path)
            else:
                # Couldn't get lock, release all and retry
                self.release_locks(locked)
                time.sleep(2)
                return self.acquire_locks(files)

        return locked

Step 3: Real-Time Observability

With multiple agents running, observability becomes critical. Here's my monitoring dashboard:

<!DOCTYPE html>
<html>
<head>
    <title>Claude Multi-Agent Command Center</title>
    <script src="https://cdn.jsdelivr.net/npm/vue@3"></script>
    <style>
        .agent-grid {
            display: grid;
            grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
            gap: 20px;
            padding: 20px;
        }
        .agent-card {
            border: 2px solid #3498db;
            border-radius: 8px;
            padding: 15px;
            position: relative;
        }
        .agent-card.active {
            border-color: #2ecc71;
            box-shadow: 0 0 10px rgba(46, 204, 113, 0.3);
        }
        .agent-status {
            position: absolute;
            top: 10px;
            right: 10px;
            width: 12px;
            height: 12px;
            border-radius: 50%;
            background: #95a5a6;
        }
        .agent-status.active { background: #2ecc71; }
        .agent-status.busy { background: #f39c12; }
        .agent-status.error { background: #e74c3c; }

        .task-progress {
            margin-top: 10px;
            height: 20px;
            background: #ecf0f1;
            border-radius: 10px;
            overflow: hidden;
        }
        .task-progress-bar {
            height: 100%;
            background: linear-gradient(90deg, #3498db, #2ecc71);
            transition: width 0.3s;
        }

        .conflict-alert {
            background: #e74c3c;
            color: white;
            padding: 10px;
            border-radius: 5px;
            margin: 10px;
        }
    </style>
</head>
<body>
    <div id="app">
        <h1>Claude Multi-Agent Command Center</h1>

        <!-- Overall Stats -->
        <div class="stats">
            <h2>Mission Status</h2>
            <p>Active Agents: {{ activeAgents.length }}</p>
            <p>Tasks Completed: {{ completedTasks }} / {{ totalTasks }}</p>
            <p>Files Modified: {{ modifiedFiles.size }}</p>
            <p>Conflicts Detected: {{ conflicts.length }}</p>
        </div>

        <!-- Conflict Alerts -->
        <div v-if="conflicts.length > 0" class="conflict-alert">
            ⚠️ File Conflicts Detected:
            <ul>
                <li v-for="conflict in conflicts" :key="conflict.file">
                    {{ conflict.file }} - {{ conflict.agents.join(' vs ') }}
                </li>
            </ul>
        </div>

        <!-- Agent Grid -->
        <div class="agent-grid">
            <div v-for="agent in agents" 
                 :key="agent.id" 
                 :class="['agent-card', { active: agent.status === 'active' }]">

                <div :class="['agent-status', agent.status]"></div>

                <h3>{{ agent.id }}</h3>
                <p>Type: {{ agent.specialization }}</p>
                <p>Current Task: {{ agent.currentTask || 'Idle' }}</p>

                <div v-if="agent.currentTask" class="task-progress">
                    <div class="task-progress-bar" 
                         :style="{ width: agent.progress + '%' }"></div>
                </div>

                <p>Files: {{ agent.workingFiles.join(', ') || 'None' }}</p>
                <p>Tasks Completed: {{ agent.completedCount }}</p>
            </div>
        </div>

        <!-- Activity Stream -->
        <div class="activity-stream">
            <h2>Live Activity</h2>
            <div v-for="event in recentEvents" :key="event.id" class="event">
                <span class="timestamp">{{ formatTime(event.timestamp) }}</span>
                <span class="agent">{{ event.agentId }}:</span>
                <span class="action">{{ event.action }}</span>
            </div>
        </div>
    </div>

    <script>
        const { createApp } = Vue;

        createApp({
            data() {
                return {
                    agents: [],
                    conflicts: [],
                    recentEvents: [],
                    totalTasks: 0,
                    completedTasks: 0,
                    modifiedFiles: new Set(),
                    ws: null
                };
            },

            computed: {
                activeAgents() {
                    return this.agents.filter(a => a.status === 'active');
                }
            },

            methods: {
                connect() {
                    this.ws = new WebSocket('ws://localhost:3001/agents');

                    this.ws.onmessage = (event) => {
                        const data = JSON.parse(event.data);

                        switch(data.type) {
                            case 'agent_update':
                                this.updateAgent(data.agent);
                                break;
                            case 'conflict':
                                this.conflicts.push(data.conflict);
                                break;
                            case 'task_complete':
                                this.completedTasks++;
                                break;
                            case 'event':
                                this.recentEvents.unshift(data.event);
                                this.recentEvents = this.recentEvents.slice(0, 50);
                                break;
                        }
                    };
                },

                updateAgent(agentData) {
                    const index = this.agents.findIndex(a => a.id === agentData.id);
                    if (index >= 0) {
                        this.agents[index] = agentData;
                    } else {
                        this.agents.push(agentData);
                    }

                    // Track modified files
                    if (agentData.workingFiles) {
                        agentData.workingFiles.forEach(f => this.modifiedFiles.add(f));
                    }
                },

                formatTime(timestamp) {
                    return new Date(timestamp).toLocaleTimeString();
                }
            },

            mounted() {
                this.connect();
            }
        }).mount('#app');
    </script>
</body>
</html>

Real-World Example: The Frontend Refactor

Last week, I needed to refactor my entire component library from class components to functional components with hooks. Here's how multi-agent orchestration handled it:

The Meta-Agent's Plan:

[
  {
    "id": "analyze-1",
    "type": "analysis",
    "description": "Scan all components and create refactoring plan",
    "dependencies": [],
    "files": []
  },
  {
    "id": "refactor-buttons",
    "type": "frontend",
    "description": "Convert all Button components to functional",
    "dependencies": ["analyze-1"],
    "files": ["components/Button/*.tsx"]
  },
  {
    "id": "refactor-forms",
    "type": "frontend", 
    "description": "Convert all Form components to functional",
    "dependencies": ["analyze-1"],
    "files": ["components/Form/*.tsx"]
  },
  {
    "id": "update-tests-buttons",
    "type": "testing",
    "description": "Update Button component tests",
    "dependencies": ["refactor-buttons"],
    "files": ["__tests__/Button/*.test.tsx"]
  },
  {
    "id": "update-tests-forms",
    "type": "testing",
    "description": "Update Form component tests",
    "dependencies": ["refactor-forms"],
    "files": ["__tests__/Form/*.test.tsx"]
  },
  {
    "id": "update-docs",
    "type": "docs",
    "description": "Update component documentation",
    "dependencies": ["refactor-buttons", "refactor-forms"],
    "files": ["docs/components/*.md"]
  }
]

The Execution:

Agent-1 and Agent-2 worked on different component folders in parallel
Agent-3 and Agent-4 updated tests as components were completed
Agent-5 regenerated documentation after all refactoring was done
Agent-6 ran performance benchmarks on the new components

Total time: 2 hours (vs estimated 2 days manual work)
Lines changed: 12,000+
Tests passing: 100%
Conflicts: 0

Handling the Complexity

Challenge 1: Resource Management

Running 10+ Claude instances will max out your system. Here's my resource manager:

# resource_manager.py
import psutil
import docker

class ResourceManager:
    def __init__(self, max_agents=10):
        self.max_agents = max_agents
        self.docker = docker.from_env()

    def can_spawn_agent(self) -> bool:
        # Check CPU usage
        if psutil.cpu_percent(interval=1) > 80:
            return False

        # Check memory
        if psutil.virtual_memory().percent > 85:
            return False

        # Check active containers
        active = len([c for c in self.docker.containers.list() 
                     if 'claude-agent' in c.name])

        return active < self.max_agents

    def spawn_agent_container(self, agent_config):
        """Spawn agent in Docker container for isolation"""

        container = self.docker.containers.run(
            'claude-agent:latest',
            environment=agent_config,
            detach=True,
            name=f"claude-agent-{agent_config['id']}",
            volumes={
                '/project': {'bind': '/workspace', 'mode': 'rw'}
            },
            cpu_quota=50000,  # Limit CPU usage
            mem_limit='2g'     # Limit memory
        )

        return container

Challenge 2: Coordination Without Conflicts

The key is smart task distribution and file locking:

# conflict_prevention.py
class ConflictPrevention:
    def __init__(self):
        self.file_graph = self.build_dependency_graph()

    def build_dependency_graph(self):
        """Map file dependencies to prevent conflicts"""

        # Analyze imports and exports
        graph = {}
        for file in glob.glob('**/*.ts', recursive=True):
            imports = self.extract_imports(file)
            graph[file] = imports

        return graph

    def can_modify_simultaneously(self, file1: str, file2: str) -> bool:
        """Check if two files can be modified in parallel"""

        # Check if files import each other
        if file2 in self.file_graph.get(file1, []):
            return False
        if file1 in self.file_graph.get(file2, []):
            return False

        # Check if they share common dependencies
        deps1 = set(self.file_graph.get(file1, []))
        deps2 = set(self.file_graph.get(file2, []))

        shared = deps1.intersection(deps2)

        # Allow if no shared critical dependencies
        return len(shared) == 0 or all(
            not self.is_critical(dep) for dep in shared
        )

Challenge 3: Quality Control

With multiple agents, quality control becomes critical:

# quality_gate.py
class QualityGate:
    def __init__(self):
        self.checks = [
            self.check_tests_pass,
            self.check_type_safety,
            self.check_no_conflicts,
            self.check_performance,
            self.check_security
        ]

    def validate_agent_work(self, agent_id: str, changes: Dict):
        """Validate agent's changes before merging"""

        results = []
        for check in self.checks:
            result = check(changes)
            results.append(result)

            if not result['passed']:
                # Revert changes and reassign task
                self.revert_changes(changes)
                self.reassign_task(agent_id, result['reason'])
                return False

        return True

    def check_tests_pass(self, changes):
        """Run tests on changed files"""

        affected_tests = self.find_affected_tests(changes['files'])

        result = subprocess.run(
            ['npm', 'test'] + affected_tests,
            capture_output=True
        )

        return {
            'passed': result.returncode == 0,
            'reason': result.stderr.decode() if result.returncode != 0 else None
        }

The Economics of Multi-Agent Development

Let's talk ROI. Running 10 Claude agents costs approximately:

API costs: ~$50/day at heavy usage
Infrastructure: ~$20/day for cloud resources

But the productivity gains:

10x faster development on parallelizable tasks
24/7 operation (agents don't sleep)
Consistent quality (no fatigue)
Comprehensive testing (every change, every time)

For a team of 5 developers, this replaces roughly $50,000/month in engineering time for $2,000/month in compute costs.

Getting Started with Multi-Agent

Start small:

Two agents: One for code, one for tests
Add observability: You need to see what's happening
Implement safety: File locks and conflict detection
Scale gradually: Add agents as you understand the patterns

The Future is Distributed

We're entering an era where a single developer can orchestrate an entire team of AI agents. The bottleneck isn't coding speed anymore - it's our ability to coordinate and direct these agents effectively.

Next week, I'm experimenting with 50+ agents working on a complete application rewrite. The meta-agent will manage sub-orchestrators, each controlling their own team of specialized agents.

It's turtles all the way down, and it's beautiful.

🚀 Take Your AI Engineering to the Next Level

🌐 Visit learn-agentic-ai.com - Your Hub for Advanced AI Development

🎓 Complete Learning Paths:

🎯 Claude Code Mastery - 7 modules from basics to multi-agent orchestration
🔧 AI Engineering Fundamentals - Build unstoppable foundations
🏗️ Production AI Systems - Enterprise-ready patterns and practices

About the Author:
I'm Brandon J. Redmond, AI Engineer & Agentic Systems Architect. I've built and deployed multi-agent systems processing millions of requests. Let's connect on LinkedIn or explore more at learn-agentic-ai.com.

Have you experimented with multiple AI agents? What challenges did you face? Let's discuss in the comments!

Previous Articles in This Series:

Part 1: The Claude Code Revolution
Part 2: Mastering Claude Hooks

Ready to build your own agent swarm? Start with the Claude Code Mastery learning path - from zero to orchestrating multiple agents in 7 comprehensive modules.

Mastering Claude Hooks: Building Observable AI Systems (Part 2)

bredmond1019 — Sat, 02 Aug 2025 22:06:43 +0000

Picture this: It's 6 AM, you're deep in flow with Claude Code, and suddenly your AI assistant decides to be helpful by running rm -rf / on your production server.

Sound terrifying? This exact scenario is what Claude hooks were designed to prevent.

In Part 1, we explored how Claude Code is revolutionizing software development. Now, let's dive into the power tool that transforms Claude from a coding assistant into a proactive development partner: Hooks.

What Are Claude Hooks?

Hooks are event-driven scripts that fire at specific points in Claude's lifecycle. Think of them as middleware for your AI assistant - intercepting, monitoring, and enhancing every interaction.

Five hook events give you complete control:

PreToolUse - Your safety guardian
PostToolUse - The observer
Notification - Interactive moments
Stop - Session complete
SubagentStop - Parallel processing

The Safety Net That Saved My Project

Last week, Claude suggested a "cleanup" operation that would've deleted my entire node_modules across all projects. Here's the hook that saved me:

#!/usr/bin/env python3
# scripts/safety_check.py

import os
import sys

DANGEROUS_PATTERNS = [
    'rm -rf /',
    'rm -rf ~',
    'sudo rm',
    ':(){:|:&};:',  # Fork bomb
    'dd if=/dev/zero',
    'chmod -R 777'
]

def check_safety():
    command = os.environ.get('TOOL_COMMAND', '')

    for pattern in DANGEROUS_PATTERNS:
        if pattern in command:
            print(f"🛑 BLOCKED: Dangerous command detected: {pattern}")
            print(f"Command was: {command}")
            sys.exit(1)  # This stops Claude from executing

    # Check for mass deletions
    if 'rm' in command and command.count('*') > 2:
        print("⚠️  Multiple wildcards in deletion command. Blocking for safety.")
        sys.exit(1)

if __name__ == "__main__":
    check_safety()

Hook configuration:

{
  "hooks": {
    "PreToolUse": [{
      "matcher": ".*",
      "hooks": [{
        "type": "command",
        "command": "python scripts/safety_check.py"
      }]
    }]
  }
}

This simple hook has prevented disasters multiple times. But safety is just the beginning.

Automation: Never Run Tests Manually Again

Tired of reminding Claude to run tests after changes? I built this automation chain:

#!/usr/bin/env python3
# scripts/auto_test.py

import os
import subprocess
import json

def handle_file_change():
    tool_name = os.environ.get('TOOL_NAME', '')
    file_path = os.environ.get('TOOL_FILE_PATH', '')

    if tool_name not in ['Edit', 'Write']:
        return

    # TypeScript file changed? Run type checking
    if file_path.endswith('.ts') or file_path.endswith('.tsx'):
        print("🔍 Running TypeScript checks...")
        subprocess.run(['npm', 'run', 'typecheck'], capture_output=True)

    # Test file changed? Run related tests
    if 'test' in file_path or 'spec' in file_path:
        print("🧪 Running tests...")
        subprocess.run(['npm', 'test', file_path], capture_output=True)

    # Component changed? Run component tests
    if '/components/' in file_path:
        test_file = file_path.replace('.tsx', '.test.tsx')
        if os.path.exists(test_file):
            print(f"🎯 Running component tests for {os.path.basename(file_path)}")
            subprocess.run(['npm', 'test', test_file])

if __name__ == "__main__":
    handle_file_change()

Now every code change triggers appropriate tests automatically. No reminders needed.

Building Real-Time Observability

When you're running multiple Claude agents in parallel, observability isn't optional - it's survival. Here's my real-time monitoring system:

The Event Streaming Hook

#!/usr/bin/env python3
# scripts/stream_event.py

import sys
import json
import requests
import os
from datetime import datetime

def stream_event():
    event_type = sys.argv[1] if len(sys.argv) > 1 else "unknown"

    # Capture all environment variables
    event = {
        'timestamp': datetime.now().isoformat(),
        'event_type': event_type,
        'session_id': os.environ.get('CLAUDE_SESSION_ID', 'unknown'),
        'tool': os.environ.get('TOOL_NAME'),
        'file_path': os.environ.get('TOOL_FILE_PATH'),
        'command': os.environ.get('TOOL_COMMAND'),
    }

    # Send to observability server (non-blocking)
    try:
        requests.post(
            'http://localhost:3000/events',
            json=event,
            timeout=0.5
        )
    except:
        pass  # Fail silently to not disrupt workflow

if __name__ == "__main__":
    stream_event()

The Real-Time Dashboard

// Simple Node.js server with WebSockets
const WebSocket = require('ws');
const express = require('express');
const app = express();
const server = require('http').createServer(app);
const wss = new WebSocket.Server({ server });

const events = [];
const clients = new Set();

app.post('/events', express.json(), (req, res) => {
    const event = req.body;
    events.push(event);

    // Broadcast to all connected clients
    const message = JSON.stringify({
        type: 'event',
        data: event
    });

    clients.forEach(client => {
        if (client.readyState === WebSocket.OPEN) {
            client.send(message);
        }
    });

    res.send('OK');
});

wss.on('connection', (ws) => {
    clients.add(ws);

    // Send recent history
    ws.send(JSON.stringify({
        type: 'history',
        data: events.slice(-100)
    }));

    ws.on('close', () => clients.delete(ws));
});

server.listen(3000);

The result? A real-time view of everything Claude does:

(Dashboard showing multiple Claude agents working in parallel with activity pulse visualization)

Voice Notifications: Stay in Flow

My favorite productivity hack? Voice notifications for long-running tasks:

#!/usr/bin/env python3
# scripts/voice_notify.py

import os
import pyttsx3
from datetime import datetime

def notify():
    event_type = os.environ.get('HOOK_EVENT', '')

    engine = pyttsx3.init()
    engine.setProperty('rate', 180)

    if event_type == 'Stop':
        duration = calculate_session_duration()
        engine.say(f"Claude session complete. Duration: {duration} minutes")
    elif event_type == 'Notification':
        engine.say("Claude needs your attention")
    elif 'test' in os.environ.get('TOOL_COMMAND', ''):
        engine.say("Tests complete. Check results.")

    engine.runAndWait()

if __name__ == "__main__":
    notify()

Now I can step away from my desk, and Claude tells me when tests finish or when input is needed.

Advanced Pattern: Multi-Agent Coordination

Running multiple Claude instances? Here's how to coordinate them:

#!/usr/bin/env python3
# scripts/agent_coordinator.py

import os
import json
import redis
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, db=0)

def coordinate_agents():
    session_id = os.environ.get('CLAUDE_SESSION_ID')
    event_type = os.environ.get('HOOK_EVENT')

    if event_type == 'PreToolUse':
        # Check if another agent is working on the same file
        file_path = os.environ.get('TOOL_FILE_PATH')
        if file_path:
            current_owner = r.get(f"file_lock:{file_path}")
            if current_owner and current_owner.decode() != session_id:
                print(f"⚠️  File is being edited by another agent: {current_owner.decode()}")
                print("Waiting for lock release...")
                sys.exit(1)
            else:
                # Acquire lock
                r.setex(f"file_lock:{file_path}", 30, session_id)

    elif event_type == 'PostToolUse':
        # Release file lock
        file_path = os.environ.get('TOOL_FILE_PATH')
        if file_path:
            r.delete(f"file_lock:{file_path}")

    elif event_type == 'SubagentStop':
        # Track completion
        r.hincrby('agent_stats', 'completed_tasks', 1)

        # Notify other agents
        message = {
            'session_id': session_id,
            'status': 'completed',
            'timestamp': datetime.now().isoformat()
        }
        r.publish('agent_updates', json.dumps(message))

if __name__ == "__main__":
    coordinate_agents()

The Complete Power Setup

Here's my production hook configuration that combines everything:

{
  "hooks": {
    "PreToolUse": [{
      "matcher": ".*",
      "hooks": [
        {
          "type": "command",
          "command": "python scripts/safety_check.py"
        },
        {
          "type": "command",
          "command": "python scripts/agent_coordinator.py"
        }
      ]
    }],
    "PostToolUse": [{
      "matcher": ".*",
      "hooks": [
        {
          "type": "command",
          "command": "python scripts/stream_event.py PostToolUse"
        },
        {
          "type": "command",
          "command": "python scripts/auto_test.py"
        },
        {
          "type": "command",
          "command": "python scripts/agent_coordinator.py"
        }
      ]
    }],
    "Notification": [{
      "matcher": ".*",
      "hooks": [
        {
          "type": "command",
          "command": "python scripts/voice_notify.py"
        },
        {
          "type": "command",
          "command": "python scripts/stream_event.py Notification"
        }
      ]
    }],
    "Stop": [{
      "matcher": ".*",
      "hooks": [
        {
          "type": "command",
          "command": "python scripts/save_session.py"
        },
        {
          "type": "command",
          "command": "python scripts/voice_notify.py"
        },
        {
          "type": "command",
          "command": "python scripts/generate_summary.py"
        }
      ]
    }],
    "SubagentStop": [{
      "matcher": ".*",
      "hooks": [
        {
          "type": "command",
          "command": "python scripts/agent_coordinator.py"
        },
        {
          "type": "command",
          "command": "python scripts/stream_event.py SubagentStop"
        }
      ]
    }]
  }
}

The Results Are Mind-Blowing

After implementing this hook system:

Zero destructive commands executed (12 blocked in the last week)
100% test coverage maintained automatically
Real-time visibility into all agent operations
45% reduction in context switching thanks to voice notifications
Parallel agent conflicts eliminated with coordination

Performance Considerations

Keep your hooks fast! They run synchronously and can slow Claude down:

# Good: Fast, non-blocking
try:
    requests.post(url, json=data, timeout=0.5)
except:
    pass

# Bad: Slow, blocking
response = requests.post(url, json=data)
response.raise_for_status()
process_response(response.json())

Aim for <100ms execution time per hook.

Your Turn: Start Small, Think Big

You don't need this entire system on day one. Start with:

A simple safety hook to prevent disasters
One automation hook for your most repetitive task
Basic logging to understand Claude's patterns

Then gradually expand based on your needs.

The Future is Observable

We're moving from black-box AI to fully observable, controllable systems. Hooks are the bridge between human intent and AI execution.

In Part 3, we'll explore multi-agent orchestration - how to run 10+ Claude instances in parallel for massive productivity gains.

But don't wait - implement your first hook today. Your future self will thank you.

📚 Master AI Development at learn-agentic-ai.com

🎓 Featured Learning Paths:

🚀 Claude Code Mastery - Complete 7-module course including advanced hooks
🛠️ AI Engineering Fundamentals - Build your foundation

📖 Related Deep-Dive Articles:

Mastering Claude Hooks: Complete Technical Guide - Extended version with advanced patterns
Claude Hooks: Automate Your AI Development Workflow - Practical automation recipes
Multi-Agent Observability: See Everything Your AI Agents Do - Build complete observability systems
Why Multi-Agent Systems Are a Trap - Critical analysis and solutions
Self-Building AI: Meta-Agents and Sub-Agent Architecture - Advanced orchestration patterns

🔧 Practical Resources:

GitHub: Claude Hooks Examples - Ready-to-use hook templates
Free AI Development Newsletter - Weekly insights and tutorials

About the Author:
I'm Brandon J. Redmond, an AI Engineer specializing in observable AI systems. I've helped dozens of teams implement production-ready AI workflows. Connect on LinkedIn or visit learn-agentic-ai.com.

What repetitive tasks would you automate with hooks? Share your ideas below!

Previous: Part 1 - The Claude Code Revolution
Next: Part 3 - Multi-Agent Orchestration

The Claude Code Revolution: How AI Transformed Software Engineering (Part 1)

bredmond1019 — Sat, 02 Aug 2025 22:06:41 +0000

Two months ago, I was the guy rolling my eyes at "AI will change everything" posts. Another autocomplete tool, I thought. Another overhyped assistant that writes buggy code.

Then I watched Boris from Anthropic demonstrate something that made my jaw drop: Their engineering onboarding time went from 2-3 weeks to 2 days. Not a typo. Two. Days.

That's when I realized we're not looking at an incremental improvement. We're witnessing a fundamental shift in how software gets built.

From Skeptic to Evangelist

Let me be brutally honest: I spent years perfecting my craft. Learning design patterns, mastering frameworks, optimizing my workflow. The idea that an AI could suddenly make all that less relevant? It was terrifying.

But here's what I discovered after two months of intensive Claude Code usage:

Week 1: "This is just fancy autocomplete"

Week 2: "Okay, it understands context better than I expected"

Week 4: "Wait, did I just build in 2 hours what usually takes 2 days?"

Week 8: "I can't imagine coding without this anymore"

The Paradigm Shift Nobody's Talking About

Everyone focuses on "AI writes code faster." That's missing the point entirely.

The real revolution? We're shifting from writing HOW to expressing WHAT.

Traditional Development:

Idea → Design → Architecture → Implementation → Testing → Debugging
        ↑                           ↑                      ↑
     (2 days)                  (5 days)              (3 days)

With Claude Code:

Idea → Express Intent → Claude Implements → Validate → Ship
        ↑                     ↑                  ↑
     (1 hour)            (30 minutes)       (30 minutes)

It's not about typing speed. It's about eliminating the translation layer between human intent and machine execution.

Real Examples That Changed Everything

Example 1: The Multi-Agent Observability System

Traditional approach: 2-3 weeks

Research WebSocket libraries
Design event architecture
Implement server
Build dashboard
Add real-time updates
Debug connection issues
Optimize performance

With Claude Code: 4 hours

Me: "Build a real-time observability system for monitoring multiple Claude agents. 
     Use WebSockets, show activity pulses, track sessions, and highlight conflicts."

Claude: [Implements complete system with server, dashboard, and visualizations]

Me: "Add AI-powered event summarization using Haiku"

Claude: [Integrates Anthropic API with intelligent batching]

The result? A production-ready system handling 10+ parallel agents with real-time monitoring.

Example 2: The Learning Platform

I needed a complete learning management system with:

Module progression tracking
MDX content rendering
Quiz functionality
Progress persistence
Internationalization

Time without Claude: 3-4 weeks (conservative estimate)

Time with Claude: 2 days

But here's the kicker - the Claude version was better. It suggested architectural patterns I hadn't considered, handled edge cases I would've missed, and included accessibility features I might have forgotten.

Why Claude Code is Different

1. It's Unopinionated (And That's Its Superpower)

While other tools force their patterns on you, Claude adapts to YOUR codebase. It reads your conventions, understands your architecture, and writes code that looks like you wrote it.

2. It Understands Context at Scale

This isn't about completing a single function. Claude can:

Understand entire codebases
Track complex requirements across files
Maintain consistency across thousands of lines
Remember architectural decisions from earlier in the conversation

3. It Learns Your Intent, Not Just Your Syntax

Example from last week:

Me: "Make it more resilient"

Claude: [Adds retry logic, error boundaries, graceful degradation, 
         timeout handling, and circuit breakers - because it understood
         the production context from our conversation]

The Uncomfortable Truth

Here's what keeps me up at night: The gap between developers using AI and those who aren't is growing exponentially.

While someone spends a week implementing authentication, an AI-assisted developer ships the entire feature set. While someone debugs a WebSocket connection, an AI-assisted developer has moved on to the next project.

This isn't about being "replaced by AI." It's about the developers who embrace these tools becoming 10x more valuable than those who don't.

What This Means For You

If you're reading this, you have a choice:

Dismiss it as hype and continue as before
Explore it casually and get marginal improvements
Master it and fundamentally transform how you build software

I chose option 3, and it's been the most exciting two months of my career.

Your Next Steps

Start Today: Don't wait for the "perfect" project. Start with your current work.
Think Differently: Stop thinking about HOW to code. Start thinking about WHAT you want to build.
Push Boundaries: Claude's capabilities will surprise you. Test the limits.
Share Your Journey: The community is figuring this out together.

The Bottom Line

We're not just writing code faster. We're fundamentally changing what it means to be a software engineer. The question isn't whether to adopt AI-assisted development - it's how quickly you can master it.

In the next part of this series, I'll show you exactly how to set up Claude Code and transform your development workflow. But don't wait for that - start exploring today.

The future isn't coming. It's already here. And it's incredible.

📚 Continue Your AI Engineering Journey

🎓 Learn More at learn-agentic-ai.com

Featured Learning Resources:

🚀 Claude Code Mastery Learning Path - 7 comprehensive modules from basics to advanced multi-agent systems
📖 AI Engineering Fundamentals - Master the core concepts

Related Deep-Dive Articles:

The Claude Code Revolution: Full Analysis - Extended version with technical details
Claude Code: Evolution of Programming with AI - Historical context and future implications
Building Intelligent AI Agents with Memory - Essential for multi-agent systems
Agent Architecture Patterns: Production Guide - Battle-tested patterns
The 7 Building Blocks of Reliable AI Agents - Skip the frameworks, understand the fundamentals

About the Author:
I'm Brandon J. Redmond, an AI Engineer & Agentic Systems Architect. I help organizations transform their development workflows with AI. Connect with me on LinkedIn or explore more resources at learn-agentic-ai.com.

What's been your experience with AI coding assistants? Share in the comments - I'd love to hear your perspective.

Next in Series: Part 2 - Mastering Claude Hooks: Building Observable AI Systems

Building Intelligent AI Agents with Memory: A Complete Guide

bredmond1019 — Tue, 08 Jul 2025 11:38:52 +0000

Have you ever wondered why ChatGPT forgets your name after every conversation? Or why your AI assistant can't remember that you prefer concise answers? The answer lies in a fundamental limitation: most AI applications today are stateless.

But what if your AI agents could remember? What if they could build relationships, learn from failures, and adapt to user preferences over time? This is the promise of agent memory systems, and it's about to transform how we build AI applications.

Why Memory Matters: The Intelligence Connection

As Richmond Alake points out in his brilliant talk, if AI is meant to mimic human intelligence, and human intelligence is fundamentally tied to memory, then it's a "no-brainer" that our agents need memory too.

Think about the most intelligent people you know. What makes them stand out? It's their ability to:

Recall relevant information at the right time
Learn from past experiences
Build upon previous knowledge
Maintain context across interactions

These are exactly the capabilities we need in our AI agents.

The Memory Spectrum: From Chatbots to Autonomous Agents

The evolution of AI applications has been rapid:

Chatbots (2022): Simple Q&A interfaces
RAG Systems (2023): Domain-specific knowledge integration
Tool-Using Agents (2024): LLMs with function calling
Memory-Enabled Agents (Now): Stateful, relationship-building systems

Each evolution has added capabilities, but memory is the key to unlocking true agent intelligence.

Understanding Agent Memory Types

Let's dive deep into the different types of memory your agents need, with practical implementation examples for each.

1. Conversational Memory

The most basic form - remembering what was said in a conversation.

interface ConversationMemory {
  conversationId: string;
  messages: Message[];
  timestamp: Date;
  metadata: {
    userId: string;
    sessionDuration: number;
    topicsSummary: string[];
  };
}

class ConversationMemoryManager {
  async store(message: Message, conversationId: string) {
    // Store in MongoDB with automatic indexing
    await this.db.collection('conversations').updateOne(
      { conversationId },
      { 
        $push: { messages: message },
        $set: { lastUpdated: new Date() },
        $inc: { messageCount: 1 }
      },
      { upsert: true }
    );
  }

  async retrieve(conversationId: string, limit: number = 10) {
    // Retrieve recent messages with context
    return await this.db.collection('conversations').findOne(
      { conversationId },
      { 
        messages: { $slice: -limit },
        projection: { messages: 1, metadata: 1 }
      }
    );
  }
}

2. Entity Memory

Tracking information about people, objects, and concepts mentioned in conversations.

interface EntityMemory {
  entityId: string;
  type: 'person' | 'organization' | 'location' | 'concept';
  attributes: Record<string, any>;
  relationships: Relationship[];
  mentions: EntityMention[];
  lastUpdated: Date;
}

class EntityMemoryManager {
  async updateEntity(entityData: Partial<EntityMemory>) {
    // Use MongoDB's flexible schema to store diverse entity types
    const entity = await this.db.collection('entities').findOneAndUpdate(
      { entityId: entityData.entityId },
      {
        $set: { 
          ...entityData,
          lastUpdated: new Date()
        },
        $addToSet: { 
          mentions: { $each: entityData.mentions || [] }
        }
      },
      { upsert: true, returnDocument: 'after' }
    );

    // Update vector embeddings for semantic search
    if (entity) {
      await this.updateEntityEmbedding(entity);
    }

    return entity;
  }

  async findRelatedEntities(entityId: string, relationshipType?: string) {
    // Graph-like queries in MongoDB
    return await this.db.collection('entities').find({
      'relationships.targetId': entityId,
      ...(relationshipType && { 'relationships.type': relationshipType })
    }).toArray();
  }
}

3. Episodic Memory

Remembering sequences of events and experiences.

interface Episode {
  episodeId: string;
  startTime: Date;
  endTime?: Date;
  events: Event[];
  outcome: 'success' | 'failure' | 'partial';
  learnings: string[];
  context: Record<string, any>;
}

class EpisodicMemoryManager {
  async recordEpisode(episode: Episode) {
    // Store complete interaction sequences
    await this.db.collection('episodes').insertOne({
      ...episode,
      // Add searchable summary
      summary: await this.generateEpisodeSummary(episode),
      // Vector embedding for similarity search
      embedding: await this.generateEmbedding(episode)
    });
  }

  async findSimilarEpisodes(currentContext: any, limit: number = 5) {
    // Use vector search to find similar past experiences
    const embedding = await this.generateEmbedding(currentContext);

    return await this.db.collection('episodes').aggregate([
      {
        $vectorSearch: {
          index: 'episode_embeddings',
          path: 'embedding',
          queryVector: embedding,
          numCandidates: 100,
          limit: limit
        }
      },
      {
        $project: {
          episodeId: 1,
          summary: 1,
          learnings: 1,
          score: { $meta: 'vectorSearchScore' }
        }
      }
    ]).toArray();
  }
}

4. Procedural Memory

Storing learned procedures and skills - like how the cerebellum stores motor skills.

interface Procedure {
  procedureId: string;
  name: string;
  steps: ProcedureStep[];
  conditions: Condition[];
  successRate: number;
  executionCount: number;
  averageExecutionTime: number;
  optimizations: Optimization[];
}

interface ProcedureStep {
  action: string;
  parameters: Record<string, any>;
  expectedOutcome: any;
  fallbackActions: string[];
}

class ProceduralMemoryManager {
  async learnProcedure(executionTrace: ExecutionTrace) {
    const procedurePattern = this.extractPattern(executionTrace);

    await this.db.collection('procedures').updateOne(
      { 'pattern.hash': procedurePattern.hash },
      {
        $inc: { 
          executionCount: 1,
          totalExecutionTime: executionTrace.duration
        },
        $push: { 
          executions: {
            $each: [executionTrace],
            $slice: -100 // Keep last 100 executions
          }
        },
        $set: {
          successRate: { 
            $divide: [
              { $size: { $filter: { 
                input: '$executions',
                cond: { $eq: ['$$this.outcome', 'success'] }
              }}},
              { $size: '$executions' }
            ]
          }
        }
      },
      { upsert: true }
    );
  }

  async selectBestProcedure(goal: string, context: any) {
    // Find procedures that match the goal and context
    const candidates = await this.db.collection('procedures').find({
      'applicableGoals': goal,
      'conditions': { $all: this.extractConditions(context) }
    }).sort({ successRate: -1, executionCount: -1 }).limit(5).toArray();

    // Return the most reliable procedure
    return candidates[0];
  }
}

5. Semantic Memory

General knowledge and facts about the world.

interface SemanticFact {
  factId: string;
  subject: string;
  predicate: string;
  object: string;
  confidence: number;
  sources: Source[];
  contradictions: Contradiction[];
  lastVerified: Date;
}

class SemanticMemoryManager {
  async storeFact(fact: SemanticFact) {
    // Check for contradictions before storing
    const contradictions = await this.findContradictions(fact);

    if (contradictions.length > 0) {
      fact.contradictions = contradictions;
      // Resolve contradictions based on source reliability
      fact = await this.resolveContradictions(fact, contradictions);
    }

    await this.db.collection('semantic_facts').insertOne({
      ...fact,
      // Create knowledge graph connections
      tripleHash: this.hashTriple(fact.subject, fact.predicate, fact.object),
      embedding: await this.generateFactEmbedding(fact)
    });
  }

  async queryKnowledge(query: string) {
    // Natural language query to structured search
    const structuredQuery = await this.parseQuery(query);

    return await this.db.collection('semantic_facts').find({
      $or: [
        { subject: { $regex: structuredQuery.subject, $options: 'i' } },
        { object: { $regex: structuredQuery.object, $options: 'i' } },
        { $text: { $search: query } }
      ],
      confidence: { $gte: 0.7 }
    }).toArray();
  }
}

6. Working Memory

Short-term memory for current task execution.

interface WorkingMemory {
  sessionId: string;
  currentGoal: string;
  activeContext: Record<string, any>;
  stack: TaskStack[];
  scratchpad: Record<string, any>;
  attentionWeights: AttentionWeight[];
  expiresAt: Date;
}

class WorkingMemoryManager {
  private cache: Map<string, WorkingMemory> = new Map();

  async maintain(sessionId: string) {
    const memory = this.cache.get(sessionId) || await this.load(sessionId);

    // Implement attention mechanism
    memory.attentionWeights = this.calculateAttention(memory);

    // Prune irrelevant information
    memory.activeContext = this.pruneContext(
      memory.activeContext,
      memory.attentionWeights
    );

    // Persist important information to long-term memory
    if (this.shouldPersist(memory)) {
      await this.persistToLongTerm(memory);
    }

    this.cache.set(sessionId, memory);
  }

  calculateAttention(memory: WorkingMemory): AttentionWeight[] {
    // Implement recency, relevance, and frequency based attention
    return Object.entries(memory.activeContext).map(([key, value]) => ({
      key,
      weight: this.calculateWeight(key, value, memory.currentGoal),
      lastAccessed: new Date()
    }));
  }
}

7. Persona Memory

Agent personality and behavioral patterns.

interface PersonaMemory {
  agentId: string;
  personality: PersonalityTraits;
  communicationStyle: CommunicationPreferences;
  expertise: string[];
  values: Value[];
  behavioralPatterns: Pattern[];
  adaptations: Adaptation[];
}

class PersonaMemoryManager {
  async initializePersona(agentConfig: AgentConfig) {
    const persona: PersonaMemory = {
      agentId: agentConfig.id,
      personality: {
        openness: agentConfig.personality?.openness || 0.7,
        conscientiousness: agentConfig.personality?.conscientiousness || 0.8,
        extraversion: agentConfig.personality?.extraversion || 0.6,
        agreeableness: agentConfig.personality?.agreeableness || 0.8,
        emotionalStability: agentConfig.personality?.emotionalStability || 0.9
      },
      communicationStyle: {
        formality: agentConfig.style?.formality || 'professional',
        verbosity: agentConfig.style?.verbosity || 'balanced',
        humor: agentConfig.style?.humor || 'occasional',
        empathy: agentConfig.style?.empathy || 'high'
      },
      expertise: agentConfig.expertise || [],
      values: agentConfig.values || [],
      behavioralPatterns: [],
      adaptations: []
    };

    await this.db.collection('personas').insertOne(persona);
    return persona;
  }

  async adaptPersona(agentId: string, feedback: UserFeedback) {
    // Learn from user preferences
    const adaptation = this.analyzeAdaptation(feedback);

    await this.db.collection('personas').updateOne(
      { agentId },
      {
        $push: { 
          adaptations: {
            ...adaptation,
            timestamp: new Date(),
            confidence: this.calculateConfidence(adaptation)
          }
        },
        $set: {
          // Gradually adjust personality based on successful interactions
          'personality.openness': { 
            $avg: ['$personality.openness', adaptation.suggestedOpenness] 
          }
        }
      }
    );
  }
}

8. Toolbox Memory

Dynamic tool discovery and selection.

interface Tool {
  toolId: string;
  name: string;
  description: string;
  inputSchema: JSONSchema;
  outputSchema: JSONSchema;
  examples: Example[];
  performanceMetrics: PerformanceMetrics;
  tags: string[];
  capabilities: string[];
}

class ToolboxMemoryManager {
  async registerTool(tool: Tool) {
    // Store tool with searchable metadata
    await this.db.collection('tools').insertOne({
      ...tool,
      embedding: await this.generateToolEmbedding(tool),
      searchableText: `${tool.name} ${tool.description} ${tool.tags.join(' ')}`
    });
  }

  async selectTools(task: string, context: any, maxTools: number = 5) {
    // Intelligent tool selection based on task requirements
    const taskEmbedding = await this.generateEmbedding(task);

    const relevantTools = await this.db.collection('tools').aggregate([
      {
        $vectorSearch: {
          index: 'tool_embeddings',
          path: 'embedding',
          queryVector: taskEmbedding,
          numCandidates: 50,
          limit: maxTools * 2
        }
      },
      {
        $match: {
          // Filter by performance metrics
          'performanceMetrics.successRate': { $gte: 0.8 },
          'performanceMetrics.averageExecutionTime': { $lte: 5000 }
        }
      },
      {
        $sort: {
          'performanceMetrics.successRate': -1,
          score: -1
        }
      },
      {
        $limit: maxTools
      }
    ]).toArray();

    return relevantTools;
  }

  async updateToolPerformance(toolId: string, execution: ToolExecution) {
    // Track tool performance for better selection
    await this.db.collection('tools').updateOne(
      { toolId },
      {
        $inc: {
          'performanceMetrics.totalExecutions': 1,
          'performanceMetrics.successfulExecutions': execution.success ? 1 : 0,
          'performanceMetrics.totalExecutionTime': execution.duration
        },
        $set: {
          'performanceMetrics.successRate': {
            $divide: [
              '$performanceMetrics.successfulExecutions',
              '$performanceMetrics.totalExecutions'
            ]
          },
          'performanceMetrics.averageExecutionTime': {
            $divide: [
              '$performanceMetrics.totalExecutionTime',
              '$performanceMetrics.totalExecutions'
            ]
          }
        }
      }
    );
  }
}

Memory Management: The Core System

Building on these memory types, we need a comprehensive memory management system that handles the lifecycle of memories.

interface MemoryManager {
  generate(input: any): Promise<Memory>;
  store(memory: Memory): Promise<void>;
  retrieve(query: Query): Promise<Memory[]>;
  integrate(memories: Memory[]): Promise<IntegratedMemory>;
  update(memoryId: string, updates: Partial<Memory>): Promise<void>;
  forget(criteria: ForgetCriteria): Promise<void>;
}

class UnifiedMemoryManager implements MemoryManager {
  private memoryTypes: Map<string, BaseMemoryManager>;

  constructor(private db: MongoDB) {
    this.memoryTypes = new Map([
      ['conversation', new ConversationMemoryManager(db)],
      ['entity', new EntityMemoryManager(db)],
      ['episodic', new EpisodicMemoryManager(db)],
      ['procedural', new ProceduralMemoryManager(db)],
      ['semantic', new SemanticMemoryManager(db)],
      ['working', new WorkingMemoryManager(db)],
      ['persona', new PersonaMemoryManager(db)],
      ['toolbox', new ToolboxMemoryManager(db)]
    ]);
  }

  async generate(input: any): Promise<Memory> {
    // Analyze input to determine memory type
    const memoryType = await this.classifyMemoryType(input);
    const manager = this.memoryTypes.get(memoryType);

    // Extract relevant information
    const extractedInfo = await manager.extract(input);

    // Generate memory with metadata
    return {
      id: generateId(),
      type: memoryType,
      content: extractedInfo,
      metadata: {
        createdAt: new Date(),
        source: input.source,
        confidence: this.calculateConfidence(extractedInfo),
        tags: await this.generateTags(extractedInfo)
      }
    };
  }

  async retrieve(query: Query): Promise<Memory[]> {
    // Multi-modal retrieval across memory types
    const retrievalStrategies = [
      this.vectorSearch(query),
      this.keywordSearch(query),
      this.graphSearch(query),
      this.temporalSearch(query)
    ];

    const results = await Promise.all(retrievalStrategies);

    // Merge and rank results
    return this.rankMemories(
      this.mergeResults(results),
      query.context
    );
  }

  async integrate(memories: Memory[]): Promise<IntegratedMemory> {
    // Combine multiple memories into coherent context
    const integrated: IntegratedMemory = {
      memories: memories,
      summary: await this.summarizeMemories(memories),
      conflicts: this.detectConflicts(memories),
      gaps: this.identifyGaps(memories),
      recommendations: await this.generateRecommendations(memories)
    };

    // Resolve conflicts if any
    if (integrated.conflicts.length > 0) {
      integrated.resolvedConflicts = await this.resolveConflicts(
        integrated.conflicts
      );
    }

    return integrated;
  }

  async forget(criteria: ForgetCriteria): Promise<void> {
    // Implement forgetting mechanism (not deletion)
    const memoriesToForget = await this.db.collection('memories').find(
      criteria.filter
    ).toArray();

    for (const memory of memoriesToForget) {
      await this.db.collection('memories').updateOne(
        { _id: memory._id },
        {
          $inc: { 
            forgettingScore: criteria.forgettingRate 
          },
          $set: {
            lastForgettingUpdate: new Date()
          }
        }
      );
    }

    // Archive memories that exceed forgetting threshold
    await this.archiveHighlyForgottenMemories();
  }

  private async archiveHighlyForgottenMemories() {
    const threshold = 0.9;

    const forgottenMemories = await this.db.collection('memories').find({
      forgettingScore: { $gte: threshold }
    }).toArray();

    if (forgottenMemories.length > 0) {
      // Move to archive collection
      await this.db.collection('archived_memories').insertMany(
        forgottenMemories.map(m => ({
          ...m,
          archivedAt: new Date(),
          archiveReason: 'forgetting_threshold_exceeded'
        }))
      );

      // Remove from active memories
      await this.db.collection('memories').deleteMany({
        _id: { $in: forgottenMemories.map(m => m._id) }
      });
    }
  }
}

Implementing Memory Signals

Richmond mentioned implementing memory signals like recall and recency. Here's how to build that:

interface MemorySignal {
  signalType: 'recency' | 'frequency' | 'importance' | 'emotion' | 'surprise';
  value: number;
  decay: DecayFunction;
}

class MemorySignalProcessor {
  calculateRecency(memory: Memory): number {
    const hoursSinceAccess = 
      (Date.now() - memory.lastAccessed.getTime()) / (1000 * 60 * 60);

    // Exponential decay with half-life of 24 hours
    return Math.exp(-0.693 * hoursSinceAccess / 24);
  }

  calculateFrequency(memory: Memory): number {
    // Logarithmic scaling to prevent over-weighting
    return Math.log2(memory.accessCount + 1) / 10;
  }

  calculateImportance(memory: Memory, context: any): number {
    // Context-dependent importance
    const baseImportance = memory.metadata.importance || 0.5;
    const contextRelevance = this.calculateContextRelevance(memory, context);
    const userPreference = this.getUserPreferenceScore(memory);

    return (baseImportance + contextRelevance + userPreference) / 3;
  }

  calculateEmotionalWeight(memory: Memory): number {
    // Memories with emotional content are better retained
    const emotions = memory.metadata.emotions || [];
    const emotionIntensity = emotions.reduce(
      (sum, emotion) => sum + emotion.intensity, 
      0
    ) / emotions.length;

    return Math.min(emotionIntensity * 1.5, 1);
  }

  calculateSurprise(memory: Memory, expectations: any): number {
    // Unexpected information is more memorable
    const expectedValue = this.getExpectedValue(memory.type, expectations);
    const actualValue = memory.content;

    const surprise = this.calculateKLDivergence(expectedValue, actualValue);
    return Math.tanh(surprise); // Normalize to [0, 1]
  }

  combineSignals(signals: MemorySignal[]): number {
    // Weighted combination of all signals
    const weights = {
      recency: 0.3,
      frequency: 0.2,
      importance: 0.25,
      emotion: 0.15,
      surprise: 0.1
    };

    return signals.reduce((total, signal) => {
      return total + (signal.value * weights[signal.signalType]);
    }, 0);
  }
}

Production Considerations

1. Scalability

class ScalableMemoryArchitecture {
  constructor(
    private primaryDB: MongoDB,
    private cache: RedisCache,
    private vectorDB: VectorDatabase
  ) {}

  async setupSharding() {
    // Shard by agent ID for horizontal scaling
    await this.primaryDB.admin().command({
      shardCollection: 'memories.conversations',
      key: { agentId: 'hashed' }
    });

    // Shard by timestamp for time-series data
    await this.primaryDB.admin().command({
      shardCollection: 'memories.episodes',
      key: { timestamp: 1 }
    });
  }

  async setupIndexes() {
    // Compound indexes for common queries
    await this.primaryDB.collection('memories').createIndexes([
      { agentId: 1, timestamp: -1 },
      { userId: 1, memoryType: 1 },
      { 'metadata.tags': 1 },
      { 'content.text': 'text' } // Full-text search
    ]);

    // TTL index for automatic cleanup
    await this.primaryDB.collection('working_memory').createIndex(
      { expiresAt: 1 },
      { expireAfterSeconds: 0 }
    );
  }

  async implementCaching() {
    // Multi-layer caching strategy
    return {
      l1Cache: new InMemoryLRU({ maxSize: 1000 }), // Hot memories
      l2Cache: this.cache, // Distributed cache
      l3Storage: this.primaryDB // Persistent storage
    };
  }
}

2. Privacy and Security

class SecureMemoryManager {
  async encryptSensitiveMemories(memory: Memory): Promise<EncryptedMemory> {
    // Identify and encrypt PII
    const piiFields = this.identifyPII(memory);

    for (const field of piiFields) {
      memory[field] = await this.encrypt(memory[field]);
    }

    return {
      ...memory,
      encrypted: true,
      encryptedFields: piiFields
    };
  }

  async implementAccessControl(agentId: string, userId: string) {
    // Role-based memory access
    const permissions = await this.getPermissions(agentId, userId);

    return {
      canRead: permissions.includes('read'),
      canWrite: permissions.includes('write'),
      canDelete: permissions.includes('delete'),
      memoryFilter: this.buildMemoryFilter(permissions)
    };
  }

  async auditMemoryAccess(access: MemoryAccess) {
    // Compliance and audit trail
    await this.db.collection('memory_audit').insertOne({
      ...access,
      timestamp: new Date(),
      ipAddress: access.request.ip,
      userAgent: access.request.userAgent
    });
  }
}

3. Performance Optimization

class OptimizedMemoryRetrieval {
  async hybridSearch(query: string, context: any): Promise<Memory[]> {
    // Parallel execution of different search strategies
    const [
      vectorResults,
      keywordResults,
      graphResults
    ] = await Promise.all([
      this.vectorSearch(query, { limit: 20 }),
      this.keywordSearch(query, { fuzzy: true }),
      this.graphTraversal(context.entities, { depth: 2 })
    ]);

    // Intelligent result fusion
    return this.fuseResults(
      vectorResults,
      keywordResults,
      graphResults,
      { weights: [0.5, 0.3, 0.2] }
    );
  }

  async precomputeEmbeddings() {
    // Batch embedding generation
    const unprocessedMemories = await this.db.collection('memories').find({
      embedding: { $exists: false }
    }).limit(1000).toArray();

    const embeddings = await this.batchGenerateEmbeddings(
      unprocessedMemories.map(m => m.content)
    );

    // Bulk update
    const operations = unprocessedMemories.map((memory, index) => ({
      updateOne: {
        filter: { _id: memory._id },
        update: { $set: { embedding: embeddings[index] } }
      }
    }));

    await this.db.collection('memories').bulkWrite(operations);
  }
}

Real-World Implementation: Customer Service Agent

Let's put it all together with a practical example:

class CustomerServiceAgent {
  private memoryManager: UnifiedMemoryManager;
  private llm: LLMProvider;

  async handleCustomerInteraction(
    message: string,
    customerId: string,
    sessionId: string
  ) {
    // 1. Load relevant memories
    const memories = await this.loadCustomerContext(customerId, sessionId);

    // 2. Update working memory
    await this.updateWorkingMemory(sessionId, {
      currentMessage: message,
      customerId,
      timestamp: new Date()
    });

    // 3. Extract entities and update entity memory
    const entities = await this.extractEntities(message);
    await this.updateEntityMemories(entities, customerId);

    // 4. Check for similar past interactions
    const similarEpisodes = await this.findSimilarInteractions(
      message,
      memories.episodic
    );

    // 5. Generate response with full context
    const response = await this.generateContextualResponse(
      message,
      memories,
      similarEpisodes
    );

    // 6. Store the interaction
    await this.storeInteraction({
      customerId,
      sessionId,
      message,
      response,
      entities,
      sentiment: await this.analyzeSentiment(message),
      resolution: response.resolved ? 'resolved' : 'ongoing'
    });

    // 7. Update customer profile
    await this.updateCustomerProfile(customerId, {
      lastInteraction: new Date(),
      interactionCount: { $inc: 1 },
      topics: entities.map(e => e.type),
      satisfactionTrend: this.calculateSatisfactionTrend(memories)
    });

    return response;
  }

  async loadCustomerContext(customerId: string, sessionId: string) {
    // Parallel load all relevant memory types
    const [
      conversational,
      entity,
      episodic,
      semantic,
      preferences
    ] = await Promise.all([
      this.memoryManager.retrieve({
        type: 'conversation',
        filter: { customerId },
        limit: 10
      }),
      this.memoryManager.retrieve({
        type: 'entity',
        filter: { 'mentions.customerId': customerId }
      }),
      this.memoryManager.retrieve({
        type: 'episodic',
        filter: { customerId },
        limit: 5
      }),
      this.memoryManager.retrieve({
        type: 'semantic',
        filter: { 'metadata.customerId': customerId }
      }),
      this.loadCustomerPreferences(customerId)
    ]);

    return {
      conversational,
      entity,
      episodic,
      semantic,
      preferences,
      summary: await this.summarizeContext({
        conversational,
        entity,
        episodic,
        semantic,
        preferences
      })
    };
  }

  async generateContextualResponse(
    message: string,
    memories: any,
    similarEpisodes: Episode[]
  ) {
    const prompt = `
You are a helpful customer service agent with access to the following context:

CUSTOMER HISTORY:
${memories.summary}

SIMILAR PAST INTERACTIONS:
${similarEpisodes.map(e => `- ${e.summary}: ${e.outcome}`).join('\n')}

CUSTOMER PREFERENCES:
${JSON.stringify(memories.preferences, null, 2)}

CURRENT MESSAGE:
${message}

Provide a helpful, personalized response that:
1. Acknowledges any past interactions if relevant
2. Uses the customer's preferred communication style
3. Resolves their issue effectively
4. Maintains consistency with previous interactions
`;

    const response = await this.llm.generate(prompt);

    return {
      text: response,
      resolved: this.checkIfResolved(response),
      nextSteps: this.extractNextSteps(response),
      sentiment: await this.analyzeSentiment(response)
    };
  }
}

The Future: Neuroscience-Inspired Architectures

Richmond's talk mentioned the collaboration between neuroscientists and engineers. Here's a glimpse of what's coming:

class NeuroscienceInspiredMemory {
  // Implement consolidation like human sleep cycles
  async consolidateMemories() {
    // Transfer important short-term memories to long-term storage
    const importantMemories = await this.selectForConsolidation();

    for (const memory of importantMemories) {
      // Strengthen connections (like synaptic strengthening)
      await this.strengthenMemoryConnections(memory);

      // Create abstract representations
      await this.createAbstractRepresentation(memory);

      // Update memory networks
      await this.updateMemoryNetworks(memory);
    }
  }

  // Implement replay mechanisms
  async replayMemories(context: any) {
    // Like hippocampal replay during rest
    const relevantMemories = await this.selectRelevantMemories(context);

    // Simulate replay to strengthen patterns
    for (const memory of relevantMemories) {
      await this.simulateReplay(memory);
      await this.updatePatternStrength(memory);
    }
  }

  // Implement predictive coding
  async predictiveProcessing(input: any) {
    // Generate predictions based on memory
    const predictions = await this.generatePredictions(input);

    // Compare with actual input
    const predictionError = this.calculatePredictionError(predictions, input);

    // Update models based on error
    if (predictionError > this.threshold) {
      await this.updatePredictiveModels(predictionError);
    }

    return {
      predictions,
      error: predictionError,
      surprisingElements: this.identifySurprises(predictions, input)
    };
  }
}

Conclusion: Memory as the Foundation of Intelligence

As we've explored, memory isn't just a nice-to-have feature for AI agents—it's the foundation of intelligence itself. By implementing comprehensive memory systems, we can transform our stateless AI applications into intelligent agents that:

Build genuine relationships with users
Learn from experience and improve over time
Maintain context across interactions
Adapt to individual preferences
Make better decisions based on past outcomes

The tools and patterns we've covered—from MongoDB's flexible document model to vector search capabilities—provide everything you need to build production-ready memory systems today.

Remember Richmond's key insight: we're not just building AI, we're architecting intelligence. And intelligence without memory is like trying to navigate life with permanent amnesia.

Start small—implement conversational memory first. Then gradually add other memory types as your agents grow more sophisticated. Your users will notice the difference, and your agents will finally be able to build the lasting relationships that make AI truly valuable.

📺 Watch the Original Talk

This post is based on Richmond Alake's excellent presentation on "Architecting Agent Memory" at MongoDB.

Watch on YouTube →

Richmond works at MongoDB and created the open-source Memoripy library mentioned in the talk. Connect with him on LinkedIn for more insights on building memory systems for AI agents.

The AI UX Crisis: How Tech Giants Are Breaking Software Design Principles

bredmond1019 — Tue, 08 Jul 2025 11:37:49 +0000

The Sixth Button That Broke My Workflow

Last week, I opened Microsoft Excel and discovered something that made me question everything I know about software design: a floating Copilot button that follows my cursor around, covering content, lagging behind my clicks, and generally making my life miserable. This wasn't just any AI button—it was the sixth way Microsoft had found to inject Copilot into my workflow.

Let me count them:

The floating button that hovers over cells
The giant Copilot button in the ribbon
The context menu when right-clicking
The Copilot app pinned to my taskbar (by default)
The "click-to-do" feature on Copilot Plus PCs
A physical Copilot key on my keyboard

And I'm betting it won't be the last.

⚠️ Warning: This isn't just a Microsoft problem. The same day, I opened Notion and found four different AI buttons on my screen simultaneously—including one placed by Firefox, my web browser, just one pixel away from my cursor.

We've Been Here Before: The Ghost of Clippy Returns

If you're experiencing déjà vu, you're not alone. This aggressive AI integration reminds me of two dark chapters in computing history:

1. Clippy: The Original Sin of Intrusive Assistance

Remember Microsoft's animated paperclip? That overeager assistant who'd pop up asking "It looks like you're writing a letter. Would you like help?" every time you typed "Dear"? We collectively hated Clippy so much that it became a cautionary tale in UX design textbooks.

Yet here we are in 2025, with Notion sporting an animated AI face custom-designed to distract, and Microsoft cramming AI suggestions into every possible interaction.

2. Browser Toolbar Hell

For those who remember the early 2000s, companies would sneakily install browser toolbars that "added functionality" but really just promoted their services. Sound familiar? Today's AI buttons feel exactly like those toolbars—unwanted additions that companies inject into our workflows because they have a product to sell, not because we asked for them.

The Current State: A Tour of AI UX Disasters

Let's examine how major tech companies are breaking fundamental design principles:

Microsoft Office: The Poster Child of AI Overreach

Microsoft Office AI Integration Count

Word:
- Ribbon button
- Floating side button
- Context menu option
- Alt+I keyboard shortcut reminder
- 4 text boxes on startup (30-40% screen real estate lost)

Outlook:
- Full-screen popup on launch (desktop & mobile)
- Email summary bar
- Reply/summarize button
- "Add zing to your mailbox" with AI themes

Excel:
- Floating cell button (with lag and fade effects)
- Ribbon integration
- Context menu

The Outlook example is particularly egregious. It summarized an automated OneDrive email about a shared file as: "Maya shared a document titled document with you via a link that works for everyone." After 19 seconds of processing. For an email it sent itself.

Google Workspace: Nine Buttons to Nowhere

One user reported finding nine Gemini buttons on a single Google Drive page. When clicked? "Gemini is still learning and can't help with that." If your AI can't actually do anything, maybe don't add nine buttons for it?

Meta: The Accidental AI Chat

WhatsApp, Instagram, and Facebook users report accidentally triggering AI chats they never wanted. I discovered I apparently had a conversation with Meta AI about "Sya Uno lottery results" from Colombia—something I have zero memory of ever asking about. This perfectly encapsulates Meta AI: a feature you can't disable that you trigger by accident.

Xiaomi: Breaking Core Functionality

Perhaps the most user-hostile example: Xiaomi moved the copy button—one of the most fundamental text operations—into a submenu to make room for an AI button in the primary position. Imagine breaking copy-paste to promote your AI assistant.

Why This Is Happening: The Two Driving Forces

1. The Shareholder Panic

Every tech executive I know tells the same story:

Shareholders heard about AI being "the next big thing"
They panicked about missing out
Board mandates trickled down: "AI in everything, NOW!"
Product teams are now measured by how much AI they can cram in

This inverts the entire product development process. Instead of identifying user needs and selecting appropriate technology, teams are handed a technology (AI) and told to find problems for it to solve.

2. The Platform Wars

There's a genuine race happening across the software stack:

Application level: Notion wants its AI to be your writing assistant
Browser level: Firefox wants to be your web AI companion
OS level: Microsoft wants Copilot to rule them all

Each layer is fighting to establish itself as the AI interface before user habits solidify. Those popup notifications and prominent buttons? They're not accidents—they're territory markers in a platform war.

The Real Cost: Productivity and Trust

❌ The Irony: These "productivity enhancing" AI features are actively destroying productivity by:

Covering content we're trying to read

Adding lag to basic operations

Distracting with animations and popups

Requiring us to hunt for settings to disable them

Breaking muscle memory by moving familiar buttons

But there's a deeper cost: trust. When software actively works against user intentions, when it prioritizes corporate goals over user needs, it breaks the fundamental contract between developer and user.

A Better Way: Principles for Ethical AI Integration

Not all AI integration is bad. Here are examples of AI done right:

Good AI Integration

Adobe's voice enhancement: Saved unusable audio recordings
DaVinci Resolve's magic masks: Enables previously impossible video editing
Transcription services: Assists with accessibility and note-taking

What makes these good?

They solve real user problems
They're invoked intentionally by the user
They don't interrupt existing workflows
They provide clear value

Design Principles for AI Features

// AI Integration Decision Framework

interface AIFeatureDecision {
  // Ask these questions before adding any AI feature

  userNeed: {
    identified: boolean;
    validated: boolean;
    alternativeSolutions: string[];
  };

  implementation: {
    isUserInitiated: boolean;
    disrupsExistingWorkflow: boolean;
    canBeDisabled: boolean;
    providesUniqueValue: boolean;
  };

  ethicalConsiderations: {
    respectsUserAutonomy: boolean;
    transparentAboutLimitations: boolean;
    protectsUserData: boolean;
  };
}

// Only proceed if all key criteria are met
function shouldImplementAIFeature(decision: AIFeatureDecision): boolean {
  return (
    decision.userNeed.identified &&
    decision.userNeed.validated &&
    decision.implementation.isUserInitiated &&
    !decision.implementation.disrupsExistingWorkflow &&
    decision.implementation.canBeDisabled &&
    decision.implementation.providesUniqueValue &&
    decision.ethicalConsiderations.respectsUserAutonomy
  );
}

The Path Forward: Reclaiming User-Centered Design

As AI engineers and product developers, we have a responsibility to push back against this trend. Here's how:

For Developers

Champion user needs over feature mandates
Measure success by user satisfaction, not AI adoption metrics
Design for intent: AI should enhance, not interrupt
Provide control: Every AI feature must be easily disabled
Respect the workflow: Don't break existing patterns

For Users

Vote with your feet: Support software that respects your workflow
Provide feedback: Tell companies when AI integration hurts productivity
Explore alternatives: Linux, LibreOffice, and other options exist
Share experiences: Public pressure can drive change

For Companies

Start with problems, not solutions
Test with real users in real workflows
Measure actual productivity, not engagement metrics
Provide granular controls for AI features
Remember Clippy: Learn from history

Conclusion: The Future We Choose

We're at a crossroads. We can continue down the path of aggressive, user-hostile AI integration—creating a new generation of Clippy-like disasters that users will mock for decades. Or we can choose a different path: thoughtful, user-centered AI that genuinely enhances our capabilities without destroying our workflows.

The technology isn't the problem. AI can be transformative when applied thoughtfully. The problem is prioritizing corporate metrics over user needs, platform wars over productivity, and feature checkboxes over genuine utility.

As someone deeply invested in AI's potential, I believe we can do better. We must do better. Because if we don't, we risk poisoning the well for AI adoption entirely—creating a generation of users who reflexively reject AI assistance because their first experiences were so frustratingly bad.

The choice is ours. Let's choose wisely.

Source: Inspired by "Software is evolving backwards" by TechAltar

Video: https://www.youtube.com/watch?v=oXtvAQ-e0iE

Creating AI Agents That Use Your MCP Server (Part 4/5)

bredmond1019 — Fri, 04 Jul 2025 18:59:13 +0000

Now comes the magic! In Part 3, we built an MCP server that can collect and analyze feedback. Today, we'll create an AI agent that uses this server to intelligently handle customer interactions. By the end, you'll have a working AI system!

What We're Building

We'll create an AI agent that:

Connects to our MCP server
Uses an LLM to understand customer needs
Automatically collects and analyzes feedback
Generates intelligent reports
Suggests business improvements

Step 1: Setting Up LLM Integration

First, let's install what we need:

# Make sure your virtual environment is activated
pip install openai python-dotenv aiohttp

Create a .env file for your API keys:

# .env file
OPENAI_API_KEY=your-api-key-here
# Get your key from: https://platform.openai.com/api-keys

Step 2: Building Our AI Agent

Create feedback_agent.py:

#!/usr/bin/env python3
"""
AI Agent for Customer Feedback System
Intelligently handles feedback collection and analysis
"""

import os
import json
import asyncio
from typing import Dict, Any, List
from datetime import datetime
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# For LLM integration
from openai import AsyncOpenAI

# For MCP client
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class FeedbackAgent:
    def __init__(self):
        self.name = "Feedback Assistant"
        self.llm = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.mcp_session = None
        self.server_params = StdioServerParameters(
            command="python",
            args=["feedback_server.py"]
        )

    async def connect_to_server(self):
        """Connect to our MCP server"""
        print(f"🔌 Connecting to feedback server...")

        async with stdio_client(self.server_params) as (read, write):
            async with ClientSession(read, write) as session:
                self.mcp_session = session

                # Initialize connection
                await session.initialize()

                # Get available tools and resources
                tools = await session.list_tools()
                resources = await session.list_resources()

                print(f"✅ Connected! Found {len(tools)} tools and {len(resources)} resources")

                # Keep the session active
                await self.run_agent_loop()

    async def think(self, task: str) -> str:
        """Use LLM to understand and plan actions"""
        response = await self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {
                    "role": "system",
                    "content": f"""You are {self.name}, an AI assistant for a café.
                    You help collect and analyze customer feedback.
                    Be friendly, professional, and insightful."""
                },
                {
                    "role": "user",
                    "content": task
                }
            ],
            temperature=0.7
        )

        return response.choices[0].message.content

    async def collect_customer_feedback(self, conversation: str) -> Dict[str, Any]:
        """Intelligently extract feedback from conversation"""

        # Use LLM to extract information
        extraction_prompt = f"""
        Extract the following from this customer conversation:
        1. Customer name (if mentioned)
        2. The main feedback points
        3. Overall rating (1-5)
        4. Key topics mentioned

        Conversation:
        {conversation}

        Return as JSON format.
        """

        response = await self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "Extract information and return valid JSON only."},
                {"role": "user", "content": extraction_prompt}
            ],
            temperature=0.3,
            response_format={"type": "json_object"}
        )

        extracted = json.loads(response.choices[0].message.content)

        # Use MCP tool to collect feedback
        result = await self.mcp_session.call_tool(
            "collect_feedback",
            arguments={
                "customer_name": extracted.get("customer_name", "Anonymous"),
                "feedback": extracted.get("feedback", conversation),
                "rating": extracted.get("rating", 3)
            }
        )

        return {
            "status": "collected",
            "details": extracted,
            "server_response": result
        }

    async def analyze_feedback_trends(self) -> str:
        """Analyze all feedback and generate insights"""

        # Get recent feedback from MCP server
        recent_feedback = await self.mcp_session.read_resource("feedback://recent")
        summary_data = await self.mcp_session.read_resource("feedback://summary")

        # Use LLM to generate insights
        analysis_prompt = f"""
        Analyze this customer feedback data and provide:
        1. Key themes and patterns
        2. Areas needing immediate attention
        3. Positive aspects to maintain
        4. Specific recommendations for improvement

        Recent Feedback:
        {recent_feedback}

        Summary Statistics:
        {summary_data}
        """

        response = await self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {
                    "role": "system", 
                    "content": "You are a business analyst. Provide actionable insights."
                },
                {"role": "user", "content": analysis_prompt}
            ],
            temperature=0.5
        )

        return response.choices[0].message.content

    async def handle_customer_interaction(self, message: str) -> str:
        """Main interaction handler"""

        # Determine intent
        intent_prompt = f"""
        Classify this message intent:
        - 'give_feedback': Customer wants to share feedback
        - 'check_status': Customer asking about their previous feedback
        - 'general_question': Other questions

        Message: {message}

        Return only the intent classification.
        """

        intent_response = await self.llm.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "Classify intent. Return only the classification."},
                {"role": "user", "content": intent_prompt}
            ],
            temperature=0.1
        )

        intent = intent_response.choices[0].message.content.strip().lower()

        if "give_feedback" in intent:
            # Collect feedback
            feedback_result = await self.collect_customer_feedback(message)

            # Generate friendly response
            response = await self.think(
                f"Customer gave feedback. Details: {feedback_result}. "
                "Thank them and mention any immediate actions we'll take."
            )

            return response

        elif "check_status" in intent:
            # Get feedback summary
            summary = await self.mcp_session.read_resource("feedback://summary")

            response = await self.think(
                f"Customer asking about feedback status. Our summary: {summary}. "
                "Provide a helpful update."
            )

            return response

        else:
            # General response
            return await self.think(f"Respond helpfully to: {message}")

    async def run_agent_loop(self):
        """Main agent loop for demo purposes"""
        print("\n🤖 Feedback Agent is ready!")
        print("Commands: 'analyze' for trends, 'quit' to exit, or chat naturally\n")

        while True:
            try:
                user_input = input("You: ").strip()

                if user_input.lower() == 'quit':
                    break

                elif user_input.lower() == 'analyze':
                    print("\n📊 Analyzing feedback trends...")
                    analysis = await self.analyze_feedback_trends()
                    print(f"\nAnalysis:\n{analysis}")

                else:
                    # Handle as customer interaction
                    response = await self.handle_customer_interaction(user_input)
                    print(f"\nAgent: {response}")

                print("\n" + "-"*50 + "\n")

            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"❌ Error: {e}")

# Main entry point
async def main():
    agent = FeedbackAgent()
    await agent.connect_to_server()

if __name__ == "__main__":
    print("🚀 Starting Feedback Agent...")
    asyncio.run(main())

Step 3: Creating Intelligent Workflows

Now let's add automated workflows. Create workflow_engine.py:

#!/usr/bin/env python3
"""
Workflow Engine for Automated Feedback Processing
"""

import asyncio
from datetime import datetime, timedelta
from typing import List, Dict, Any

class FeedbackWorkflow:
    def __init__(self, agent):
        self.agent = agent
        self.workflows = {
            "daily_analysis": self.daily_analysis_workflow,
            "negative_feedback_alert": self.negative_feedback_workflow,
            "weekly_report": self.weekly_report_workflow
        }

    async def daily_analysis_workflow(self):
        """Run daily analysis of feedback"""
        print(f"\n📅 Running daily analysis - {datetime.now().strftime('%Y-%m-%d %H:%M')}")

        # Get and analyze feedback
        analysis = await self.agent.analyze_feedback_trends()

        # Check for critical issues
        if "immediate attention" in analysis.lower():
            print("🚨 Critical issues detected!")
            # In production: send alerts, create tickets, etc.

        print(f"✅ Daily analysis complete")
        return analysis

    async def negative_feedback_workflow(self, feedback_data: Dict[str, Any]):
        """Handle negative feedback immediately"""
        print(f"\n⚠️ Negative feedback detected from {feedback_data['customer_name']}")

        # Generate personalized response plan
        response_plan = await self.agent.think(
            f"Create action plan for this negative feedback: {feedback_data['feedback']}. "
            "Include: immediate response, follow-up actions, and prevention measures."
        )

        print(f"📋 Action Plan:\n{response_plan}")

        # In production: 
        # - Send alert to manager
        # - Create follow-up task
        # - Track resolution

        return response_plan

    async def weekly_report_workflow(self):
        """Generate comprehensive weekly report"""
        print(f"\n📊 Generating weekly report...")

        # Get all data
        summary = await self.agent.mcp_session.read_resource("feedback://summary")
        recent = await self.agent.mcp_session.read_resource("feedback://recent")

        # Generate executive summary
        report = await self.agent.think(
            f"Create executive summary for this week's feedback. "
            f"Data: {summary}\nRecent examples: {recent}\n"
            "Include: key metrics, trends, recommendations, and success stories."
        )

        print(f"📄 Weekly Report:\n{report}")
        return report

    async def run_workflow(self, workflow_name: str, *args, **kwargs):
        """Execute a specific workflow"""
        if workflow_name in self.workflows:
            return await self.workflows[workflow_name](*args, **kwargs)
        else:
            raise ValueError(f"Unknown workflow: {workflow_name}")

class WorkflowScheduler:
    def __init__(self, workflow_engine):
        self.engine = workflow_engine
        self.scheduled_tasks = []

    def schedule_daily(self, hour: int, minute: int, workflow: str):
        """Schedule daily workflow"""
        self.scheduled_tasks.append({
            "type": "daily",
            "time": {"hour": hour, "minute": minute},
            "workflow": workflow
        })

    def schedule_on_event(self, event_type: str, workflow: str):
        """Schedule event-triggered workflow"""
        self.scheduled_tasks.append({
            "type": "event",
            "trigger": event_type,
            "workflow": workflow
        })

    async def run(self):
        """Run the scheduler"""
        print("⏰ Workflow scheduler started")

        # For demo: simulate some workflows
        await asyncio.sleep(2)
        await self.engine.run_workflow("daily_analysis")

        # Simulate negative feedback
        await asyncio.sleep(3)
        await self.engine.run_workflow("negative_feedback_alert", {
            "customer_name": "Demo User",
            "feedback": "The coffee was cold and service was slow",
            "rating": 2
        })

Step 4: Putting It All Together

Create run_system.py to run the complete system:

#!/usr/bin/env python3
"""
Run the Complete AI Feedback System
"""

import asyncio
import subprocess
import time
from feedback_agent import FeedbackAgent
from workflow_engine import FeedbackWorkflow, WorkflowScheduler

async def run_complete_system():
    """Run server and agent together"""

    # Start the MCP server in background
    print("🚀 Starting MCP Feedback Server...")
    server_process = subprocess.Popen(
        ["python", "feedback_server.py"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    # Give server time to start
    time.sleep(2)

    try:
        # Create and connect agent
        print("🤖 Starting AI Agent...")
        agent = FeedbackAgent()

        # Create workflow engine
        workflow = FeedbackWorkflow(agent)
        scheduler = WorkflowScheduler(workflow)

        # Schedule some workflows
        scheduler.schedule_daily(9, 0, "daily_analysis")
        scheduler.schedule_on_event("negative_feedback", "negative_feedback_alert")
        scheduler.schedule_daily(17, 0, "weekly_report")

        # Run the agent
        await agent.connect_to_server()

    finally:
        # Clean up
        server_process.terminate()
        print("\n👋 System shutdown complete")

if __name__ == "__main__":
    print("🎯 Complete AI Feedback System Starting...\n")
    asyncio.run(run_complete_system())

Step 5: Testing Your AI System

Let's test the complete system:

# Make sure your .env file has your OpenAI API key
python run_system.py

Try these interactions:

Give feedback:
- "Hi, I'm Sarah. I loved the new latte recipe! 5 stars!"
- "The service was terrible today. My order took 30 minutes."
Check analytics:
- Type: analyze
Ask questions:
- "What's the overall feedback trend?"
- "How many people gave feedback today?"

Key Concepts We've Implemented

1. Agent-Server Communication

The agent connects to the MCP server and can:

Read resources (feedback data)
Call tools (collect feedback, analyze sentiment)
Process results intelligently

2. LLM Integration

We use GPT-3.5 to:

Understand natural language
Extract information from conversations
Generate insights and recommendations
Create personalized responses

3. Automated Workflows

Our system can:

Run scheduled analyses
Respond to events (negative feedback)
Generate reports automatically

4. Intelligent Decision Making

The agent:

Classifies user intent
Chooses appropriate actions
Provides contextual responses

What's Next?

In our final part (Part 5), we'll:

Add data persistence with a real database
Implement advanced features (multi-language support, voice input)
Deploy the system for production use
Explore scaling strategies

You've now built a functional AI system! How would you extend this for your business needs? What other workflows would be helpful?

Ready to go deeper? Access the complete tutorial with additional examples at learn-agentic-ai.com/learn/paths/ai-systems-intro.

Challenge: Extend Your System

Try adding these features:

Email notifications for negative feedback
Sentiment trend visualization
Customer response templates
Multi-location support

Share your extensions in the comments!

The Building Blocks - Understanding MCP, Agents, and LLMs (Part 2/5)

bredmond1019 — Fri, 04 Jul 2025 18:58:36 +0000

In Part 1, we learned that AI systems are like smart orchestra conductors. Now, let's meet the musicians in our orchestra and understand how they work together. By the end of this article, you'll understand the core components we'll use to build our AI system.

The Four Essential Components

1. Workflow Engines: The Conductor

A workflow engine orchestrates the entire process. Think of it as a recipe that says:

First, collect the customer feedback
Then, analyze the sentiment
Next, categorize the issues
Finally, generate a summary report

Popular workflow engines include:

n8n: Visual workflow automation
Temporal: Code-based workflows
Apache Airflow: Data pipeline orchestration

2. AI Agents: The Specialized Workers

AI agents are like skilled employees, each with a specific job:

Research Agent: Gathers information from various sources
Analysis Agent: Processes and understands data
Writing Agent: Creates reports and summaries
Action Agent: Performs tasks like sending emails

Here's a simple conceptual example:

class FeedbackAnalysisAgent:
    def __init__(self, llm):
        self.llm = llm
        self.name = "Feedback Analyzer"

    def analyze(self, feedback_text):
        prompt = f"Analyze this feedback and identify key themes: {feedback_text}"
        return self.llm.complete(prompt)

3. Large Language Models: The Brains

LLMs are the intelligence behind AI agents. They:

Understand natural language
Generate human-like responses
Analyze patterns and sentiment
Make contextual decisions

Popular options:

OpenAI GPT-4: Powerful and versatile
Anthropic Claude: Great for analysis and coding
Local models: Privacy-focused alternatives

4. MCP Servers: The Universal Connectors

Here's where things get exciting. Model Context Protocol (MCP) is like a universal adapter that lets AI systems connect to any tool or service.

Imagine you have:

A filing cabinet (your database)
A telephone (communication system)
A calculator (processing tool)

MCP creates a standard way for AI to use all these tools, regardless of their original design.

How MCP Works: A Simple Analogy

Think of MCP like a restaurant menu:

Resources = Menu Sections
- "Today's Specials" (dynamic data)
- "Regular Menu" (static information)
Tools = Things You Can Order
- "Get me the soup of the day"
- "I'll have the daily special"
- "Can you modify this dish?"
Server = The Waiter
- Takes your requests
- Communicates with the kitchen
- Brings back results

Why MCP Changes Everything

Before MCP, connecting AI to different tools was like needing a different translator for each country you visit. With MCP, it's like everyone suddenly speaks the same language.

Benefits:

Standardization: One protocol for all integrations
Flexibility: Easy to add new capabilities
Security: Clear boundaries and permissions
Simplicity: Less code, more functionality

A Sneak Peek: Our MCP Server Structure

Here's what we'll build in the upcoming parts:

from mcp.server import Server, Resource, Tool

class CustomerFeedbackServer:
    def __init__(self):
        self.server = Server("feedback-analyzer")
        self.feedback_store = []

    def setup(self):
        # Define what data we expose
        self.server.add_resource(Resource(
            uri="feedback://recent",
            name="Recent Feedback",
            description="Last 10 customer feedback entries"
        ))

        # Define what actions we can take
        self.server.add_tool(Tool(
            name="analyze_sentiment",
            description="Analyze emotional tone of feedback",
            input_schema={
                "type": "object",
                "properties": {
                    "text": {"type": "string"}
                }
            }
        ))

Putting It All Together

When these components work together:

Workflow Engine says: "Time to analyze today's feedback"
AI Agent receives the task and thinks: "I need to get the feedback first"
MCP Server provides access to the feedback data
LLM analyzes the sentiment and extracts insights
Results flow back through the system to generate reports

What's Next?

In Part 3, we'll roll up our sleeves and set up our development environment. You'll:

Install Python and the MCP SDK
Set up your first MCP server
Test basic connectivity
Prepare for building our feedback analysis system

The exciting part? Once you understand these building blocks, you can apply them to ANY business problem.

Ready to dive deeper? Follow along with the complete tutorial at learn-agentic-ai.com/learn/paths/ai-systems-intro.

What tools or services would you want to connect to an AI system? Drop a comment below - your idea might become our next example!