← Back to news

Building an Advanced Agentic Harness

data4sci.com|63 points|29 comments|by Anon84|Aug 5, 2026

Building an Advanced Agentic Harness

From a single pilot to an air campaign: planning, parallelism, memory, verification, and observability for production-shaped agents.

Date: 2026-07-15 | Source: Data For Science Resources: GitHub Companion Code


The Philosophy: Beyond the Basic Loop

A lone pilot in a high-performance jet might win a dogfight, but you don't win a war that way. Real-world air campaigns require a massive support structure:

  • Mission Planners to strategize sorties before takeoff.
  • Squadrons to execute independent missions in parallel.
  • Fuel Budgets and "bingo calls" to ensure aircraft return before running dry.
  • Flight Recorders for post-mission reconstruction.
  • After-Action Reviews (AAR) to verify if the objective was actually met.

This structure ensures the system is fast, safe, debuggable, and measurable. Modern production agents—like Claude Code, Devin, Cursor, and Hermes—apply this same logic to the basic LLM loop.

The Core Challenge: How do you transform a solitary LLM call into a resilient system capable of planning, acting, recovering, and proving its own correctness?

From Naive to Production

We move away from a "naive" loop by building small, testable primitives. The following table maps specific agent failures to their architectural solutions:

Naive Failure ModeProduction-Grade Solution
LLMs hallucinate invalid tool argumentsTyped Tools with Pydantic validation
Everything runs sequentially (slow)Plan DAG & Parallel Execution
Context window fills with "junk"Tiered Memory under a retrieval budget
Errors propagate silentlyVerification Hierarchy
One prompt tries to do everythingRole Splitting (Planner \rightarrow Worker \rightarrow Critic)
API costs spiral out of controlMulti-dimensional Budgeting

The Running Example: City Comparison Agent

To demonstrate these primitives, we build an agent that compares a list of cities based on population, timezone, and a narrative summary.

While this seems simple, it is a perfect stress test because:

  1. Parallelism: A request for three cities requires nine independent lookups.
  2. Dependencies: The final report cannot be written until all lookups are complete.
  3. Verification: We can programmatically check if every requested city is present in the final output.
  4. Cost Variance: Dictionary reads (population/timezone) are "cheap," while LLM summaries are "expensive."

Basic Loop Approach \rightarrow Advanced Harness Approach

  • Define Tool Schemas
  • Generate Execution DAG
  • Execute Parallel Workers
  • Verify Results via Critic
  • Aggregate Final Report

Primitive 1: The Pluggable Brain

To avoid vendor lock-in and enable testing, we don't hard-wire the SDK. Instead, we use an abstraction layer.

class LLMProvider:
    """Shared interface. Subclass to plug in a different backend."""
    def complete(self, system: str, user: str, role: str = "default") -> str:
        raise NotImplementedError

    async def acomplete(self, system: str, user: str, role: str = "default") -> str:
        # Wrap sync call in a thread to maintain async compatibility
        return await asyncio.to_thread(self.complete, system, user, role)

We also implement a MockProvider. This returns deterministic responses (e.g., a canned plan or a rule-based pass/fail), allowing us to distinguish between orchestration errors and model reasoning errors.


Primitive 2: Typed Tools

Manual validation of tool arguments is a recipe for failure. Instead, we use Pydantic to drive the schema, the validation, and the documentation.

@dataclass
class TypedTool:
    name: str
    description: str
    args_model: type[BaseModel] # Pydantic model for schema
    fn: Callable[..., Any]
    cost_hint: float = 0.0 # Used for budget accounting

    def schema(self) -> dict:
        """Returns the shape expected by Anthropic/OpenAI APIs."""
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.args_model.model_json_schema(),
        }

    def run(self, raw_args: dict) -> Any:
        args, err = self.validate_args(raw_args)
        if err is not None:
            raise ValueError(err)
        return self.fn(**args.model_dump())

The Agentic Workflow

The interaction flow can be visualized as follows:


Budgeting and Costs

In a production system, we must track costs. We can represent the total cost of a mission using a simple summation:

TotalCost=i=1n(ToolCosti×Frequencyi)+OverheadLLMTotalCost = \sum_{i=1}^{n} (ToolCost_i \times Frequency_i) + \text{Overhead}_{LLM}

By assigning a cost_hint to each TypedTool, the orchestrator can implement graceful degradation—for example, skipping a high-cost "detailed summary" if the budget is nearly exhausted.


Visualizing the Architecture

The following images illustrate the transition from a basic loop to a structured harness:

1. The Conceptual Shift

2. The Execution Flow

3. Memory and Context Management

4. Verification Layers

5. The Final Integrated Harness