The IOS Agent Layer — Building Reliable Multi-Step Analysis Pipelines for Finance
A production agent layer for document analysis is not a GPT wrapper with function calling. It’s a state machine with failure handling, a typed tool registry, deterministic routing logic, and a human..
The IOS architecture blueprint described the analysis engine this way: “Route to extraction / prompt chain / agent by complexity.” The agent layer was acknowledged as early. This post is what early actually means — and what it takes to make it production-grade.
A production agent layer for document analysis is not a GPT wrapper with function calling. It’s a state machine with failure handling, a typed tool registry, deterministic routing logic, and a human handoff protocol that fires before the system produces something it shouldn’t. Getting that right is the current focus of IOS engineering.
The Intuition: Why Document Analysis Agents Are Different
Most agent frameworks are designed for general task completion: search the web, write code, answer questions, book a meeting. The agent can take any sequence of actions, can branch in any direction, and the cost of a wrong turn is usually just extra API calls and wasted time.
Document analysis in finance is different in three ways that matter for agent design.
First, the action space is bounded. Analyzing a regulatory filing means: retrieve relevant passages, extract specific metrics, compare against framework requirements, flag discrepancies, and produce a structured output with citations. Every analysis task reduces to a small set of typed operations. You don’t want a general agent that could call any function in any order — you want a constrained agent that cannot accidentally do the wrong thing.
Second, the outputs are consequential. If an agent analyzing a capital adequacy assessment hallucinates a Tier 1 capital ratio, or misattributes a figure to the wrong year, or misreads a threshold from the wrong framework, those errors can propagate into actual risk assessments. Downstream use cases include things analysts act on. That changes the error tolerance from “acceptable” to “must not happen.”
Third, confidence matters more than capability. A general agent should handle hard cases. A finance analysis agent should route hard cases to a human before attempting them badly. The right answer is not “try harder” — it’s “flag for review and be honest about uncertainty.”
These three constraints shape the entire architecture: bounded tools, output validation with confidence gating, and mandatory human handoff for anything below threshold.
IOS Agent State Machine and Reliability Metrics
The Architecture: A Typed State Machine
The IOS agent layer is implemented as a typed state machine with five states: RECEIVE_TASK, PLAN, EXECUTE_STEP, VALIDATE_OUTPUT, and either DELIVER or HUMAN_REVIEW.
The key design decision is that the agent cannot transition between states arbitrarily. Each transition is typed: RECEIVE_TASK only transitions to PLAN; PLAN only transitions to EXECUTE_STEP; EXECUTE_STEP transitions to itself (next step) or VALIDATE_OUTPUT (all steps done); VALIDATE_OUTPUT transitions to either DELIVER or HUMAN_REVIEW based on confidence.
There is no path from EXECUTE_STEP back to PLAN that bypasses VALIDATE_OUTPUT. There is no way for the agent to deliver output without a validation gate. These constraints are structural, not behavioral — they cannot be overridden by a poorly-written prompt.
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable, Any
from enum import Enum
import time
class AgentState(Enum):
RECEIVE_TASK = "receive_task"
PLAN = "plan"
EXECUTE_STEP = "execute_step"
VALIDATE_OUTPUT = "validate_output"
DELIVER = "deliver"
HUMAN_REVIEW = "human_review"
@dataclass
class AnalysisRequest:
request_id: str
doc_ids: List[str]
query_type: str # "extract_metric" | "compare_frameworks" | "compliance_check" | ...
query_text: str
regulatory_framework: Optional[str] = None
analytical_steps_defined: bool = False
constraints: Dict[str, Any] = field(default_factory=dict)
@dataclass
class AnalysisStep:
step_id: str
tool_name: str
tool_args: Dict[str, Any]
expected_output_type: str
retry_limit: int = 2
@dataclass
class StepResult:
step_id: str
tool_name: str
output: Any
confidence: float
citations: List[Dict]
error: Optional[str] = None
retries_used: int = 0
@dataclass
class AgentRunState:
request: AnalysisRequest
state: AgentState
plan: List[AnalysisStep] = field(default_factory=list)
results: List[StepResult] = field(default_factory=list)
current_step_idx: int = 0
overall_confidence: float = 0.0
validation_issues: List[str] = field(default_factory=list)
started_at: float = field(default_factory=time.time)
The Tool Registry: Typed, Bounded, Testable
Every tool the agent can call is registered explicitly. No dynamic tool generation, no passing arbitrary code, no LLM-generated function names. The tool registry is a typed dictionary that maps tool names to callable handlers with defined input and output schemas.
This matters for two reasons. First, it makes the agent testable in isolation: you can mock any tool and verify the agent’s behavior under specific failure conditions without running actual LLM inference. Second, it makes the failure surface explicit: the agent can only fail in ways that correspond to known tool failures, not in arbitrary ways from an unconstrained action space.
from typing import TypedDict
class ToolRegistry:
def __init__(self, retrieval_service, extraction_service,
framework_db, llm_client, output_validator):
self._tools: Dict[str, Callable] = {
"retrieve_context": self._make_retrieve_tool(retrieval_service),
"extract_metric": self._make_extract_tool(extraction_service, llm_client),
"compare_frameworks": self._make_compare_tool(framework_db, llm_client),
"flag_disclosure": self._make_flag_tool(llm_client),
"generate_summary": self._make_summary_tool(llm_client),
"validate_citation": self._make_citation_tool(retrieval_service),
}
def invoke(self, tool_name: str, args: Dict) -> StepResult:
if tool_name not in self._tools:
raise ValueError(f"Unknown tool: {tool_name}. Registry: {list(self._tools.keys())}")
return self._tools[tool_name](**args)
def _make_extract_tool(self, extraction_service, llm_client):
def extract_metric(
chunks: List[Dict],
metric: str,
doc_id: str,
fallback_to_qualitative: bool = True,
) -> StepResult:
# Attempt structured extraction
result = extraction_service.extract(chunks, metric)
if result.value is not None and result.confidence >= 0.7:
return StepResult(
step_id="extract",
tool_name="extract_metric",
output={"metric": metric, "value": result.value, "unit": result.unit},
confidence=result.confidence,
citations=result.source_citations,
)
# Fallback: qualitative flag
if fallback_to_qualitative:
qualitative = llm_client.complete(
f"Does the following text discuss {metric}? Reply with yes/no and a brief quote if yes.\\n\\n"
+ "\\n".join([c["text"] for c in chunks[:3]])
)
return StepResult(
step_id="extract",
tool_name="extract_metric",
output={"metric": metric, "value": None, "qualitative": qualitative},
confidence=0.55, # below our 0.85 gate → will route to human
citations=[],
error="numeric_extraction_failed_qualitative_fallback",
)
raise ValueError(f"Metric extraction failed for {metric} with no fallback")
return extract_metric
The Planner: Routing by Complexity
One of the most important decisions in the agent architecture is how task complexity is assessed before planning. Getting this wrong means either over-engineering simple extractions into multi-step agent loops (slow, expensive) or under-engineering complex analyses into single-pass extractions (wrong).
The IOS planner uses a rule-based classifier. Not an LLM-based planner — a rule-based one. This is deliberate. LLM-based planning is flexible but unpredictable: the same request might produce different plans across runs, and debugging a planning failure requires inspecting LLM reasoning rather than code logic.
For a bounded action space in a well-defined domain, rules are better.
class AnalysisPlanner:
SINGLE_METRIC_TYPES = {"extract_metric", "classify_document", "check_threshold"}
COMPLIANCE_TYPES = {"compliance_assessment", "regulatory_gap_analysis"}
def plan(self, request: AnalysisRequest) -> List[AnalysisStep]:
complexity = self._assess_complexity(request)
if complexity == "extraction":
return self._plan_extraction(request)
elif complexity == "chain":
return self._plan_chain(request)
elif complexity == "compliance":
return self._plan_compliance(request)
else:
return self._plan_full_agent(request)
def _assess_complexity(self, request: AnalysisRequest) -> str:
if (request.query_type in self.SINGLE_METRIC_TYPES
and len(request.doc_ids) == 1):
return "extraction"
elif (request.analytical_steps_defined
and len(request.doc_ids) == 1):
return "chain"
elif request.query_type in self.COMPLIANCE_TYPES:
return "compliance"
else:
return "full_agent"
def _plan_extraction(self, request: AnalysisRequest) -> List[AnalysisStep]:
return [
AnalysisStep("s1", "retrieve_context",
{"query": request.query_text, "doc_ids": request.doc_ids, "k": 6},
"chunks"),
AnalysisStep("s2", "extract_metric",
{"metric": request.constraints.get("metric", request.query_text),
"doc_id": request.doc_ids[0]},
"metric_result"),
AnalysisStep("s3", "validate_citation",
{"claim": "metric_extraction"},
"citation_check"),
]
def _plan_compliance(self, request: AnalysisRequest) -> List[AnalysisStep]:
return [
AnalysisStep("s1", "retrieve_context",
{"query": request.query_text, "doc_ids": request.doc_ids, "k": 10},
"chunks"),
AnalysisStep("s2", "extract_metric",
{"metric": request.constraints.get("metric"), "doc_id": request.doc_ids[0]},
"metric_result"),
AnalysisStep("s3", "compare_frameworks",
{"framework": request.regulatory_framework},
"framework_comparison"),
AnalysisStep("s4", "flag_disclosure",
{"topic": request.constraints.get("topic")},
"disclosure_flags"),
AnalysisStep("s5", "generate_summary",
{"include_citations": True, "format": "structured"},
"analysis_result"),
]
Agent Tool Traces, Recovery Logic, Latency, and Routing
Error Recovery and Graceful Degradation
Tool failures are expected. The extraction tool fails when a metric isn’t present as a number. The retrieval tool returns low-confidence chunks when the query doesn’t match well. The LLM occasionally produces malformed structured output.
The agent must handle each of these without crashing, without silently returning wrong output, and without blindly retrying. The pattern we use is layered recovery:
class DocumentAnalysisAgent:
CONFIDENCE_THRESHOLD = 0.85
def execute_step_with_recovery(
self, step: AnalysisStep, state: AgentRunState
) -> StepResult:
last_error = None
for attempt in range(step.retry_limit + 1):
try:
result = self.tool_registry.invoke(step.tool_name, step.tool_args)
result.retries_used = attempt
return result
except Exception as e:
last_error = e
# Try degraded fallback on last retry
if attempt == step.retry_limit:
fallback_result = self._attempt_graceful_degradation(step, state, str(e))
if fallback_result:
return fallback_result
# All retries exhausted, no fallback—return low-confidence error result
return StepResult(
step_id=step.step_id,
tool_name=step.tool_name,
output=None,
confidence=0.0,
citations=[],
error=str(last_error),
retries_used=step.retry_limit,
)
def _attempt_graceful_degradation(
self, step: AnalysisStep, state: AgentRunState, error: str
) -> Optional[StepResult]:
"""
Return a partial result rather than nothing.
Partial results always have confidence < CONFIDENCE_THRESHOLD,
so they automatically route to human review.
"""
if step.tool_name == "extract_metric":
# Fall back to qualitative flag if numeric extraction fails
return self.tool_registry.invoke(
"flag_disclosure",
{"topic": step.tool_args.get("metric"), "chunks": state.results[-1].output}
)
return None
The key insight in that code: graceful degradation always produces results with confidence below the 0.85 threshold. This means degraded results automatically route to human review — you don’t have to remember to add a special flag. The confidence gate handles it structurally.
The Validation Gate and Human Handoff
Every completed plan goes through a validation gate before delivery. The gate checks three things: citation coverage (every numerical claim has a document reference), numerical constraint validity (extracted values are in plausible ranges for the metric type), and confidence (the minimum confidence across all step results).
from dataclasses import dataclass
@dataclass
class ValidationResult:
valid: bool
confidence: float
issues: List[str]
route_to_human: bool
reason: Optional[str] = None
class OutputValidator:
ALWAYS_REVIEW = {"compliance_assessment", "risk_flag", "regulatory_gap_analysis"}
def validate(self, state: AgentRunState) -> ValidationResult:
issues = []
min_confidence = min(
(r.confidence for r in state.results if r.confidence > 0), default=0.0
)
# Citation check
for result in state.results:
if result.output and "value" in str(result.output) and not result.citations:
issues.append(f"Missing citation for step {result.step_id}")
# Confidence gate
route = (
min_confidence < self.CONFIDENCE_THRESHOLD
or len(issues) > 0
or state.request.query_type in self.ALWAYS_REVIEW
)
return ValidationResult(
valid=len(issues) == 0,
confidence=min_confidence,
issues=issues,
route_to_human=route,
reason=(
f"low_confidence:{min_confidence:.2f}" if min_confidence < self.CONFIDENCE_THRESHOLD
else "requires_human_review" if state.request.query_type in self.ALWAYS_REVIEW
else "citation_issues" if issues else None
)
)
Notice ALWAYS_REVIEW: compliance assessments and risk flags always route to a human, regardless of confidence. Not because the system can’t produce an answer — but because the stakes are high enough that automated delivery isn’t appropriate even when the confidence is high. This is a domain policy decision, not a technical one. It belongs in code, not in a prompt.
Where IOS Currently Stands
The architecture described above is the target. The current state of the IOS agent layer is partial implementation:
The tool registry exists and has
retrieve_context,extract_metric, andgenerate_summaryimplemented.The planner handles
extractionandchaincomplexity; thecomplianceandfull_agentpaths are stubs.The validation gate exists and the confidence threshold is enforced; citation checking is not yet implemented.
The human review queue exists but is currently manual — flagged items go to a spreadsheet rather than a proper review interface.
The retry/recovery logic was implemented this week as a result of writing this post. The act of writing the architecture forces you to be precise about the parts that aren’t yet built.
Current reliability metrics (on a test set of 80 analysis requests): task completion 71%, citation accuracy 62%, hallucination rate 22%. Production targets: 92%, 89%, 4%.
The gap is real. The path to closing it is: finish the compliance planner, implement citation checking in the validator, and build the proper human review interface. Those are the next three engineering tasks.
Production Considerations
Idempotency. Each step should be idempotent — running it twice produces the same result. This enables safe retries without corrupting state.
Step result caching. If the same (query, doc_id) retrieval request is made twice in the same session, cache the result. Agent loops with self-correction can burn through API budget fast.
Timeouts per tool. Each tool call should have an explicit timeout. A retrieval call that takes 30 seconds has probably hit a database issue. Fail fast, log the timeout, fall through to degradation.
Audit log. Every agent run should produce a complete audit log: request received, plan generated, each step’s input/output/confidence, validation result, and delivery decision. For finance and risk applications, this log is essential for post-hoc review. Store it durably alongside the output.
What Comes Next
With better retrieval (BGE-finance-v1) and a more reliable agent layer, the two biggest failure modes in IOS are closing. But there’s a third problem we haven’t addressed: we still don’t know if the agent is getting better over time in ways that matter to analysts.
Unit evals measure correctness against known answers. Task evals measure end-to-end completion. But neither tells you whether analysts prefer the outputs — whether the structured assessments are useful, whether the summary language matches what a compliance officer actually needs, whether the citation format is trustworthy in practice.
That’s the preference evaluation problem. It’s the next unfinished piece. And it’s the hardest one.
Action for today: Write down the three most common failure modes your agent system produces. For each one, identify whether the failure is in the tool (wrong output), the planner (wrong tool selected), the validator (failure not caught), or the handoff (failure caught but not routed correctly). The location of the failure determines the fix.
I am publishing lessons and my thoughts as I’m constructing IntelligenceOS — a domain AI platform, for Finance Ops and Risk — built on a family of fine-tuned models — in public, one project at a time. You’ll get the real account: what I built, what broke, what I learned, and how it compounds into a production system.
Interesting state-machine framing. For finance analysis, a clever loop with tools attached is not enough. The system needs typed tools, explicit failure states, deterministic routing where possible, and a human handoff before analysis turns into a decision.