Building an Advanced Agentic Harness

That Basic Harness loop is correct, but naive. A lone pilot in a well-built jet might win a dogfight, but nobody runs an air campaign that way. Real operations add mission planners who decide what sorties to fly before anyone takes off, squadrons that fly independent sorties in parallel, fuel budgets and bingo calls that force a return to base before the tanks run dry, flight recorders that make every mission reconstructible after the fact, and after-action reviews that decide whether the mission actually succeeded. None of these replace the pilot. They wrap the pilot in structure so that the whole system stays fast, safe, debuggable, and measurable. Claude Code, Devin, Cursor, Hermes, and other production agents do exactly the same thing to the basic loop. In this post we upgrade every piece of our basic harness toward that production shape, without hiding any of the mechanics behind a framework. The guiding question for the whole exercise is a simple one: How do you turn a single LLM call into a reliable system that can plan, act, recover, and prove it did the right thing? Our answer is composition. We build small, testable primitives: typed tools, a plan DAG, tiered memory, a verification hierarchy, budgets, and a tracer, and wire them together with a deliberately thin orchestrator. Each primitive exists because naive agents fail in a specific, predictable way. LLMs invent invalid tool arguments, so we add typed tools with Pydantic validation. Everything runs sequentially, so we add a dependency graph and parallel execution. The context window fills with junk, so we add multi-tier memory under a retrieval budget. Bad outputs propagate silently, so we add a verification hierarchy. One prompt tries to do everything, so we split it into Planner , Worker , and Critic roles. Costs run away, so we add multi-dimensional budgeting with graceful degradation. Proving the harness usually works, with an eval suite, retrieval benchmarks, and specialized worker pools will get a full fledge treatment in a future post. The running example Throughout the post we build a city comparison agent: given a list of cities, it produces a report comparing them on population, timezone, and a short narrative summary of each. The task looks almost insultingly simple, but it was chosen carefully. Each city-attribute lookup is independent of every other one, which means a three-city request naturally decomposes into nine tool calls that could all run at the same time. The final report, on the other hand, depends on all of the lookups finishing first, so we’re. well beyond a flat list of steps. We can programmatically check that every requested city actually appears in the report to verify the results. And the tools have wildly different costs: population and timezone lookups are in-memory dictionary reads, while the per-city summaries and the final aggregation each call the LLM, which gives us realistic budget pressure to manage. For the sake of reproducibility, lookup tools read from a small mocked dictionary, CITY_FACTS , so the notebook is fully reproducible without network access. The LLM-backed pieces can run against either a real Anthropic model or a deterministic mock, which brings us to the first primitive. A pluggable brain Every component we are about to build eventually calls an LLM: the planner, the summarizer, the aggregator, the critic. If that call is hard-wired to one SDK, the entire harness becomes untestable and vendor-locked. So before anything else, we define a base class that provides an abstraction over the details of the various LLM calling APIs 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; works for any SDK. return await asyncio.to_thread(self.complete, system, user, role) We also implement a MockProvider for testing and debugging purposes that returns deterministic, role-aware responses: a canonical plan when asked to plan, a templated one-line summary when asked to summarize, a rule-based pass/fail verdict when asked to judge. This allows us to separate “is my orchestration wrong?” from “is the model planning badly?” during development, and it is the reason every experiment in this post is reproducible on any machine. Typed tools In the basic harness we validated tool arguments by hand, an approach collapses quickly: every new tool duplicates validation logic, the LLM never sees a formal schema and just guesses argument shapes, and the resulting errors are ad hoc strings the model can’t self-correct from. The upgrade is to declare each tool’s arguments as a Pydantic model and let one definition drive everything: @dataclass class TypedTool: name: str description: str args_model: type[BaseModel] # Pydantic model defining the arg schema fn: Callable[..., Any] cost_hint: float = 0.0 # relative cost for budget accounting def schema(self) -> dict: """Shape expected by Anthropic/OpenAI tool-use 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()) This approach gets us runtime validation, a JSON Schema in exactly the shape that the Anthropic and OpenAI tool-use APIs expect, documentation (each Field(…, description =…) becomes part of the catalog the planner reads), and a hook for cost accounting via cost_hint. Failing before execution allows us to avoid expensive tool calls with potential side effects. A bad plan should fail fast , at the validation layer, and not deep inside a database query. This approach is similar to what full fledge frameworks like LangChain tools, Anthropic tool use, and OpenAI function calling all converge on. Our registry holds four tools with three cost tiers: get_population and get_timezone() are essentially free dictionary lookups (cost_hint =0.1), summarize_city() makes one LLM call per city (cost_hint =1.0), and aggregate_report() makes the token-heavy synthesis call that produces the final markdown (cost_hint =2.0). Note that the last two are tools that call the LLM internally. LLMs are just like any other tool. The worker sees a uniform tool interface, but some tools are wrappers around sub-prompts, which means you can cache, rate-limit, or swap the inner model independently of the harness. The plan is a Graph The basic harness executed one action per turn. That works when steps are strictly sequential, but our task has nine independent lookups feeding a single aggregation: A while-loop runs these one at a time. A Directed Acyclic Graph expresses the dependencies explicitly and lets an executor run everything that is ready right now, concurrently. So instead of asking the LLM for one action at a time, we ask the Planner for the whole graph up front. The LLM declares the structure before we execute anything. Since the planner is an LLM, it can hallucinate structure too: dependencies on node IDs that don’t exist, or circular dependencies that can never complete. So the very first thing we do with a plan is to validate it before possibly wasting tokens trying to execute a broken plan. def ready_nodes(self) -> list[PlanNode]: """Nodes whose deps are all DONE and are themselves PENDING.""" out = [] for n in self.nodes.values(): if n.status != NodeStatus.PENDING: continue if all(self.nodes[d].status == NodeStatus.DONE for d in n.deps): out.append(n) return out ready_nodes() is the heart of the scheduler: at any moment, it returns the set of nodes whose dependencies are all satisfied. For our three-city goal, the planner emits ten nodes: nine fetches with empty dependency lists, all eligible to run in parallel, and one aggregate_report capstone that depends on all nine. Executing the graph in parallel The executor is a level-synchronous DAG walker: compute the ready set, launch every ready node concurrently with asyncio.gather , mark each one done or failed, and repeat until nothing is left or no forward progress is possible. MAX_CONCURRENT = 5 # cap concurrent tool/LLM calls async def execute_dag(dag, tools, on_step=None): semaphore = asyncio.Semaphore(MAX_CONCURRENT) async def run_node(node): node.status = NodeStatus.RUNNING async with semaphore: try: # Sync tools run in a thread pool so other nodes can proceed node.result = await asyncio.to_thread(tools[node.tool].run, node.args) node.status = NodeStatus.DONE except Exception as exc: node.error = f”{type(exc).__name__}: {exc}” node.status = NodeStatus.FAILED while not dag.is_done(): ready = dag.ready_nodes() if not ready: break # remaining nodes depend on FAILED ancestors await asyncio.gather(*(run_node(n) for n in ready)) Two small decisions carry most of the weight here. First, asyncio.to_thread runs our synchronous tool functions in a thread pool, which means we never have to rewrite tools as async def or couple the harness to async-native SDKs. Second, the semaphore caps concurrency, because without it a fifty-node plan would spawn fifty simultaneous LLM calls and promptly hit rate limits or a cost spike. This is deliberately not a full dynamic scheduler with work-stealing and priority queues. For agent workloads like ours, where each node is an API call lasting hundreds of milliseconds to seconds, level-synchronous parallelism captures most of the win. Sequentially, wall time is roughly the sum of the fetch latencies; in parallel, it is roughly the maximum of them plus the aggregation step. Remembering the right things