The enterprise AI landscape is undergoing a fundamental shift. We’re moving from primarily request-response applications toward systems that can dynamically reason about goals, select actions, use tools, maintain state and operate within defined control boundaries.
Traditional LLMs excel at single-turn tasks. Enterprise agents must handle autonomy — perceiving their environment, reasoning about complex problems, dynamically selecting tools, executing actions, and learning from outcomes — all while maintaining strict security, compliance, and cost controls.
This comprehensive reference guide addresses production requirements for building trustworthy agentic AI systems in enterprise environments, with particular emphasis on financial services, healthcare, and other regulated domains.
Part 1: Foundations and Taxonomy
1.1 What is an AI Agent?
An AI agent is a system capable of:
Perceiving its environment through sensors, APIs, and data sources
Reasoning about problems using logical inference and learned patterns
Planning multi-step solutions to achieve defined goals
Acting by selecting and executing appropriate tools and APIs
Maintaining state within a single execution or across sessions
Operating under constraints and policies that limit autonomy
Key characteristic: Goal-directed behavior with dynamic decision-making over actions/tools within defined boundaries.
Agents do not necessarily require:
Persistent memory across sessions (though this is useful)
Continuous online learning (though this can enhance performance)
Unbounded autonomy (in fact, the opposite is required)
1.2 Corrected Taxonomy: LLM, Workflow, and Agent
These categories often overlap. Define them by primary characteristics, not mutually exclusive features:
Memory, planning, reflection, and learning are architectural capabilities, not mandatory requirements for agents.
Important distinctions:
An LLM application CAN:
Call tools and manage state
Retrieve data dynamically
Maintain conversation history
Use external memory
A workflow CAN:
Contain LLM-based decisions
Use dynamic branching based on model output
Implement adaptive retrieval
Incorporate probabilistic components
An agent CAN:
Be stateless across sessions (maintains state only during execution)
Operate without long-term learning (uses retrieved memories instead)
Be deterministic in decision logic (if constraints are tight)
Perception transforms environmental data into actionable insights. RAG reduces hallucination risk by grounding model responses in retrieved enterprise data, but retrieval quality, source quality, context selection, and model behavior can still produce unsupported or incorrect answers.
class PerceptionEngine: def __init__(self, data_sources: List[DataSource]): self.data_sources = data_sources self.rag_engine = RAGEngine() async def perceive(self, query: str, context: ExecutionContext) -> Perception: # Parallel retrieval from multiple sources results = await asyncio.gather( self.rag_engine.retrieve(query, context), self.fetch_structured_data(query, context), self.fetch_api_data(query, context) ) # Deduplicate and rank by relevance unified = self.deduplicate(results) ranked = self.rank_by_relevance(unified, context) # Apply authorization and data governance filtered = self.apply_data_policies(ranked, context) return Perception( data=filtered, source_confidence=self.assess_source_quality(filtered), gaps=self.identify_missing_information(filtered), timestamp=datetime.utcnow() )
Perception challenges:
Latency: Sub-second retrieval for interactive response
Consistency: Managing eventual consistency across sources
Freshness: Balancing cache efficiency with data recency
Authorization: Row/column-level filtering per user
Relevance: Selecting pertinent data from massive datasets
Poisoning: Detecting and filtering malicious injected data
2.4 Planning: From Goal to Task Sequence
class PlanningEngine: def __init__(self, llm: LLMClient, tool_registry: ToolRegistry): self.llm = llm self.tools = tool_registry async def plan(self, goal: str, context: ExecutionContext) -> Plan: # Retrieve similar successful plans from memory similar_plans = await self.memory.retrieve_similar_plans(goal, limit=3) # Generate plan using LLM prompt = self.build_planning_prompt(goal, similar_plans, self.tools) response = await self.llm.generate(prompt, max_tokens=2000) # Parse plan structure plan = self.parse_plan(response) # Validate against policies and constraints validated = await self.validate_plan_safety(plan, context) return validated async def validate_plan_safety(self, plan: Plan, context: ExecutionContext) -> Plan: """Ensure plan adheres to security, compliance and capability constraints""" for step in plan.steps: # Verify tool exists tool = self.tools.get(step.tool_name) if not tool: raise PlanValidationError(f"Unknown tool: {step.tool_name}") # Check authorization early (Phase 6) if not context.can_use_tool(tool.name): raise AuthorizationError(f"Not authorized to use {step.tool_name}") # Validate arguments match tool signature self.validate_arguments(step.args, tool.signature) return plan
2.5 Reasoning: Evidence-Based Decision-Making
Important: Model-generated confidence scores are not automatically calibrated probabilities.
class ReasoningEngine: def __init__(self, llm: LLMClient, knowledge_base: VectorDB): self.llm = llm self.knowledge_base = knowledge_base async def reason(self, question: str, context: ContextWindow) -> Reasoning: # Generate structured decision trace (NOT raw chain-of-thought) decision_trace = await self.generate_decision_trace(question, context) # Verify facts against knowledge base verified_facts = await self.verify_facts(decision_trace) # Calculate evidence-based risk score risk_score = self.calculate_risk_score( evidence_quality=verified_facts.quality, retrieval_confidence=verified_facts.source_confidence, policy_constraints=context.constraints, business_impact=decision_trace.impact_assessment ) if risk_score > self.escalation_threshold: return Reasoning( conclusion="UNCERTAIN", confidence=risk_score, requires_human_review=True, reasoning_summary=decision_trace.summary, supporting_evidence=verified_facts.items ) return Reasoning( conclusion=decision_trace.recommendation, confidence=risk_score, reasoning_summary=decision_trace.summary, supporting_evidence=verified_facts.items )
What to log: goal, selected action, tool arguments, policy decisions, retrieved sources, validation results, outcome, escalation decisions — NOT raw LLM-generated reasoning chains.
NIST notes that generated reasoning/citations can themselves be confabulated, so raw chain-of-thought does not ensure trustworthiness.
2.6 Memory: Sophisticated Knowledge Systems
Agents may use multi-layered memory depending on use case:
class MemorySystem: def __init__(self): # Short-term: Current execution context self.working_memory = ContextWindow(max_tokens=4000) # Long-term semantic: Vector embeddings of facts/patterns self.semantic_memory = VectorDatabase() # Episodic: Specific execution traces and outcomes self.episodic_memory = TimeseriesDB() async def store_experience(self, execution: ExecutionTrace): """Store experience for future reference""" # Extract key learnings learnings = self.extract_learnings(execution) # Store execution trace with provenance await self.episodic_memory.store( id=execution.execution_id, data=execution, provenance=execution.audit_trail, ttl=self.calculate_ttl(execution) ) # Store semantic embedding embedding = await self.embed(execution.summary) await self.semantic_memory.store( embedding=embedding, metadata={ "execution_id": execution.execution_id, "success": execution.succeeded, "created_at": execution.created_at, "user_id": execution.user_id } )
Memory governance requirements:
Time-to-live (TTL) policies
User/tenant isolation
Deletion workflows
Poisoning detection
Access control
Compliance retention policies
Encryption at rest and in transit
Part 3: Production Architecture
3.1 Corrected Enterprise Security Architecture
The architecture must enforce zero-trust agent execution:
USER │ ▼ ┌──────────────────────────┐ │ Identity & IAM │ │ • SSO (SAML/OAuth/OIDC) │ │ • MFA enforcement │ │ • Token issuance │ │ • Session validation │ └────────────┬─────────────┘ │ ▼ ┌──────────────────────────┐ │ AI Gateway │ │ • Request validation │ │ • Prompt injection check │ │ • DLP/PII scanning │ │ • Rate limiting │ │ • Billing/quota │ └────────────┬─────────────┘ │ ▼ ┌──────────────────────────┐ │ Policy Decision Point │ │ • RBAC evaluation │ │ • ABAC rules │ │ • Data classification │ │ • Risk assessment │ └────────────┬─────────────┘ │ (If policy denies → REJECT) │ ▼ ┌──────────────────────────────────┐ │ Agent Runtime (Isolated) │ │ Perceive → Reason → Plan │ │ (State in isolation) │ └────────────┬─────────────────────┘ │ ┌────────────┼──────────────┐ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌──────────┐ │ RAG/DB │ │ MCP │ │ External │ │(policy)│ │ Servers│ │ APIs │ └────────┘ └────────┘ └──────────┘ │ │ │ └────────────┼──────────────┘ │ ▼ ┌──────────────────────────┐ │ Tool Authorization │ ← CRITICAL │ (BEFORE execution) │ │ • User permission check │ │ • Policy authorization │ │ • Risk classification │ │ • High-risk approval? │ └────────────┬─────────────┘ │ (If denied → REJECT) │ ▼ ┌──────────────────────────┐ │ Execution Sandbox │ │ • OS/filesystem isolation│ │ • Network egress control │ │ • Resource limits │ │ • Timeout enforcement │ └────────────┬─────────────┘ │ ▼ ┌──────────────────────────┐ │ Result Validation │ │ • Output filtering │ │ • DLP/PII redaction │ │ • Citation validation │ │ • Accuracy checks │ └────────────┬─────────────┘ │ ▼ OUTPUT │ ▼ ┌──────────────────────────┐ │ Observability & Audit │ │ • Structured logging │ │ • Distributed tracing │ │ • SIEM/threat detection │ │ • Cost tracking │ │ • Compliance audit │ └──────────────────────────┘
Key principle: Authorization occurs BEFORE tool execution, not after.
RAG reduces hallucination and confabulation risk by grounding model responses in retrieved enterprise data, but retrieval quality, source quality, context selection, and model behavior can still produce unsupported or incorrect answers.
class RAGEngine: def __init__(self, vector_db: VectorDB, embedding_model: EmbeddingModel, rerank_model: RerankModel, source_validator: SourceValidator): self.vector_db = vector_db self.embeddings = embedding_model self.reranker = rerank_model self.source_validator = source_validator async def retrieve_and_augment(self, query: str, context: ExecutionContext, top_k: int = 5) -> AugmentedContext: """Retrieve relevant documents with quality assessment""" # 1. Embed query query_embedding = await self.embeddings.embed(query) # 2. Dense retrieval with row-level filtering dense_results = await self.vector_db.search( query_embedding, top_k=top_k*2, filters=self.build_authorization_filters(context) ) # 3. Sparse/keyword retrieval sparse_results = await self.vector_db.keyword_search( query, top_k=top_k*2, filters=self.build_authorization_filters(context) ) # 4. Merge and deduplicate merged = self.merge_results(dense_results, sparse_results) # 5. Rerank for relevance reranked = await self.reranker.rerank(query, merged, top_k) # 6. Validate sources and check for poisoning validated = [] for doc in reranked: source_check = await self.source_validator.validate(doc) if source_check.is_trustworthy: validated.append((doc, source_check.confidence)) else: self.audit_log.warn(f"Untrusted source detected: {doc.id}") # 7. Format context for model formatted = self.format_context(validated) # 8. Assess overall retrieval quality retrieval_quality = self.assess_retrieval_quality(validated) return AugmentedContext( documents=formatted, source_confidence=retrieval_quality.confidence, has_gaps=len(validated) < top_k, quality_assessment=retrieval_quality, citations=[d.source_id for d, _ in validated] )
RAG security controls:
Document provenance tracking
Source trust levels
Document integrity validation
Ingestion-time validation
Row-level access control
Column-level masking
Prompt injection scanning on retrieved content
Embedding poisoning detection
Citation validation
Stale-document invalidation
4.2 MCP: Standardization Without Security Boundaries
Critical clarification: The Model Context Protocol (MCP) is an interoperability standard for tool/resource discovery. MCP is not itself a security boundary.
MCP standardization does NOT eliminate tool security risks. MCP servers still require:
Authentication (API keys, mTLS)
Authorization (user/role-based)
Network isolation (allowlisting, VPC boundaries)
Input validation at the tool level
Output filtering at the tool level
Rate limiting per caller
Audit logging of calls
Version control and change management
Monitoring and alerting
class MCPServerGateway: """Bridge between agent and MCP servers with security enforcement""" def __init__(self, server_registry: MCPServerRegistry): self.registry = server_registry async def invoke_mcp_tool(self, server_name: str, tool_name: str, args: Dict, context: ExecutionContext) -> Any: """Invoke MCP tool with security controls""" # 1. Verify server is allowlisted server = self.registry.get(server_name) if not server or not server.is_allowlisted: raise SecurityError(f"MCP server not allowed: {server_name}") # 2. Authenticate to MCP server auth_result = await self.authenticate_mcp_server(server) if not auth_result.success: raise AuthenticationError(f"Failed to authenticate to MCP server: {server_name}") # 3. Validate tool exists and user is authorized tool = server.list_tools().get(tool_name) if not tool: raise ToolNotFound(f"Tool not found on MCP server: {tool_name}") if not context.can_use_tool(f"{server_name}:{tool_name}"): raise AuthorizationError(f"User not authorized for {server_name}:{tool_name}") # 4. Sanitize arguments sanitized_args = self.sanitize_arguments(args, tool.input_schema) # 5. Apply rate limiting await self.rate_limiter.check_limit( f"{context.user_id}:{server_name}", tokens=1 ) # 6. Invoke tool through MCP protocol try: result = await asyncio.wait_for( server.call_tool(tool_name, sanitized_args), timeout=30 ) # 7. Validate output output_validation = await self.validate_output(result, tool) if not output_validation.valid: raise OutputValidationError(output_validation.reason) # 8. Apply output filtering filtered_output = await self.filter_output(result, context) # 9. Log execution await self.audit_log.log_mcp_tool_call( server=server_name, tool=tool_name, user=context.user_id, status="SUCCESS" ) return filtered_output except Exception as e: await self.audit_log.log_mcp_tool_call( server=server_name, tool=tool_name, user=context.user_id, status="ERROR", error=str(e) ) raise
Part 5: Agentic Threat Model and Security
5.1 Agentic AI Threat Model
OWASP explicitly identifies agent-specific risks. Use this threat model to structure security testing:
5.2 Red Team Testing Framework
class RedTeamingFramework: """Systematic security testing of agents""" async def run_red_team_exercises(self, agent: Agent) -> RedTeamReport: """Execute comprehensive security tests""" report = RedTeamReport() tests = [ # Jailbreak resistance self.test_direct_jailbreaks(agent), self.test_role_switching(agent), self.test_instruction_injection(agent), self.test_prompt_leaking(agent), # Authorization enforcement self.test_unauthorized_tool_access(agent), self.test_privilege_escalation(agent), self.test_user_isolation(agent), # Data security self.test_pii_leakage(agent), self.test_data_exfiltration(agent), self.test_sql_injection(agent), # Operational resilience self.test_timeout_handling(agent), self.test_error_cases(agent), self.test_resource_exhaustion(agent), # Adversarial robustness self.test_adversarial_prompts(agent), self.test_confusing_context(agent), self.test_contradictory_information(agent) ] for test in tests: result = await test report.add_result(result) if result.severity == "CRITICAL": report.add_remediation(result.recommended_fix) return report
Part 6: AI Governance and Lifecycle
6.1 Agent Risk Classification
Governance depends on risk tier:
Tier 1 — Informational
No external actions
Read-only
Example: Document summarizer
Controls:
Basic authentication
Audit logging
Usage monitoring
Tier 2 — Assistive
Creates recommendations but human executes
No autonomous actions
Example: Customer issue classifier
Controls:
Role-based access
Output audit
Decision logging
Tier 3 — Controlled Action
Agent performs low-risk actions under strict policy
Medium impact
Example: Ticket creation
Controls:
Policy-based authorization
Risk assessment
Audit trail
Failure notification
Tier 4 — High-Impact Autonomous
Agent can affect financial, customer, production or regulated systems
High potential impact
Rarely fully autonomous
Example: Payment authorization
Controls:
Explicit approval workflows
Risk-based limits
Comprehensive audit
Model risk management
Human oversight
6.2 AI Governance Lifecycle
Use-Case Intake ↓Risk Classification (Tier 1-4) ↓Architecture Review • Threat modeling • Security design review • Compliance requirements ↓Data Assessment • Classification • Sensitivity • Privacy impact • Retention ↓Model Assessment • Capability validation • Bias testing • Robustness evaluation • Explainability review ↓Security Assessment • Penetration testing • Red team exercises • Compliance audit • Control validation ↓Evaluation Setup • Baseline metrics • Test data • Evaluation criteria • Success threshold ↓Business Approval • Risk acceptance • Stakeholder sign-off • Budget allocation ↓Pilot Deployment • Limited traffic • Continuous monitoring • Ready rollback ↓Production Deployment • Gradual rollout • Canary testing • Failure monitoring ↓Continuous Monitoring • Performance tracking • Security monitoring • Cost analysis • User feedback ↓Periodic Re-certification • Quarterly/annual review • Updated threat model • Compliance refresh • Model performance audit ↓Retirement/Decommissioning • Data deletion per policies • Audit trail archival • Lessons learned
6.3 Model Risk Management (Financial Services)
Critical addition for regulated environments:
Model Inventory:
Complete registry of all agents/models
Owner and approval chain
Risk tier and impact classification
Deployment environment
Version and change history
Model Validation:
Pre-deployment testing
Performance benchmarking
Bias and fairness assessment
Explainability validation
Regulatory compliance check
Model Performance Monitoring:
Accuracy tracking
Drift detection
Bias monitoring
Degradation alerts
Challenger model testing
Change Management:
Version control for prompts/policies
Testing on changes
Approval workflow
Rollback procedures
Audit trails
Documentation:
Model card
Risk assessment
Validation report
Explainability documentation
Incident log
Part 7: AI Supply Chain Security
Critical for enterprise security posture:
┌─────────────────────────────────────┐│ Foundation Model Provider ││ (OpenAI, Anthropic, etc.) ││ • Attestation ││ • Audit reports ││ • Security SLA │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ Model Provenance Verification ││ • Checksums ││ • Signed artifacts ││ • Security audit trail │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ Embedding/Reranker Models ││ • Security assessment ││ • Version pinning ││ • Update procedures │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ Agent Framework & Libraries ││ • Dependency scanning ││ • Vulnerability assessment ││ • License compliance ││ • Update management │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ MCP Servers & Plugins ││ • Code review ││ • Security testing ││ • Allowlist management ││ • Version control │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ Container Images & Artifacts ││ • Image scanning ││ • SBOM generation ││ • Signing ││ • Registry security │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ Python/Language Dependencies ││ • Dependency audit ││ • Vulnerability tracking ││ • Supply chain verification ││ • Update procedures │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ External Tools & APIs ││ • Security assessment ││ • Contract review ││ • Rate limiting agreements ││ • Data residency confirmation │└────────────┬────────────────────────┘ │┌────────────▼────────────────────────┐│ Data Sources & Vectors ││ • Source verification ││ • Data governance ││ • Licensing compliance ││ • Regular audits │└─────────────────────────────────────┘
Controls:
SBOM (Software Bill of Materials) for all components
Dependency scanning for vulnerabilities
Image signing with code attestation
Policy enforcement on allowed models/tools
Version pinning to prevent unexpected changes
Regular audits of supply chain
Incident response procedures for compromised components
Important correction: Retry only idempotent operations or use idempotency keys for state-changing operations. Avoid independent retries at multiple layers unless retry budget is explicitly coordinated.
class ResilientExecutor: async def execute_with_resilience(self, operation: Callable, args: Dict, idempotency_key: Optional[str] = None, is_idempotent: bool = False) -> Any: """Execute with coordinated retry strategy""" # Check circuit breaker if self.circuit_breaker.is_open(): raise CircuitBreakerOpenError("Service temporarily unavailable") last_error = None for attempt in range(1, self.retry_config.max_attempts + 1): try: result = await asyncio.wait_for( operation(**args), timeout=self.retry_config.timeout ) self.circuit_breaker.record_success() return result except (TimeoutError, ConnectionError) as e: last_error = e # For state-changing operations, use idempotency key if not is_idempotent and not idempotency_key: self.circuit_breaker.record_failure() raise if attempt < self.retry_config.max_attempts: wait_seconds = self.retry_config.initial_backoff * (2 ** (attempt - 1)) await asyncio.sleep(wait_seconds) except Exception as e: self.circuit_breaker.record_failure() raise self.circuit_breaker.record_failure() raise ExecutionError(f"Failed after {self.retry_config.max_attempts} attempts")
8.5 Cost Management
class CostManager: async def track_execution_cost(self, execution: ExecutionTrace) -> Cost: """Calculate and track execution cost with budget enforcement""" cost = Cost() # Tokens input_cost = execution.input_tokens * self.model_pricing[execution.model].input_cost / 1000 output_cost = execution.output_tokens * self.model_pricing[execution.model].output_cost / 1000 cost.add_input_cost(input_cost) cost.add_output_cost(output_cost) # Tools for tool in execution.tools_used: if tool.has_cost: tool_cost = await self.get_tool_cost(tool.name) cost.add_tool_cost(tool_cost) # Budget enforcement user_budget = self.budgets.get(execution.user_id) if user_budget and cost.total > user_budget.remaining: raise BudgetExceededError( f"Execution would exceed budget: ${cost.total} > ${user_budget.remaining}" ) # Deduct from budget if user_budget: user_budget.deduct(cost.total) return cost async def optimize_cost(self, plan: Plan) -> OptimizedPlan: """Optimize plan for cost efficiency""" optimized_steps = [] for step in plan.steps: # Check for cached results (reuse eliminates cost) cached = await self.check_cache(step) if cached: step.use_cached_result(cached) optimized_steps.append(step) continue # Try cheaper alternatives alternatives = self.find_alternatives(step) best = min(alternatives, key=lambda x: x.estimated_cost) optimized_steps.append(best) return OptimizedPlan(optimized_steps)
Part 9: Financial Services Implementation
9.1 Corrected Terminology and Requirements
Financial services is a highly regulated environment with stringent requirements for security, privacy, resilience, auditability, and model risk management.
Regulatory frameworks and standards:
Assurance frameworks:
SOC 2 (Type II for continuous audits)
Security/payment standards:
PCI DSS (payment data)
NIST Cybersecurity Framework
OWASP Top 10
Regulatory obligations:
Bank Secrecy Act (AML/KYC)
Gramm-Leach-Bliley Act (GLBA)
Dodd-Frank Act (if applicable)
Fair Credit Reporting Act (FCRA)
Equal Credit Opportunity Act (ECOA)
Regional privacy laws (GDPR if EU, CCPA if CA, etc.)
9.2 Model Risk Management Framework
For applicable financial-services use cases, organizations should consider:
Model Inventory
@dataclassclass ModelInventory: model_id: str name: str owner: User approval_chain: List[User] risk_tier: RiskTier deployed_environment: str version: str created_date: datetime last_modified_date: datetime sunset_date: Optional[datetime] documentation: ModelCard
Model Validation
class ModelValidator: async def validate_model(self, model: Agent) -> ValidationResult: """Pre-deployment validation""" # Accuracy testing accuracy = await self.test_accuracy(model) if accuracy.score < self.min_accuracy_threshold: return ValidationResult(passed=False, reason="Accuracy below threshold") # Bias assessment bias_result = await self.assess_bias(model) if bias_result.has_discriminatory_bias: return ValidationResult(passed=False, reason="Bias detected") # Explainability explainability = await self.assess_explainability(model) if not explainability.adequate_for_regulator: return ValidationResult( passed=False, reason="Cannot sufficiently explain model decisions" ) # Regulatory compliance compliance = await self.check_regulatory_compliance(model) if not compliance.compliant: return ValidationResult( passed=False, reason=f"Compliance failure: {compliance.reason}" ) return ValidationResult(passed=True)
php
async def assess_bias(self, model: Agent) -> BiasResult: """Test for discriminatory bias""" # Run model on protected characteristics protected_groups = ["race", "gender", "age", "disability"] for group in protected_groups: # Compare outcomes across groups group_outcomes = await self.evaluate_across_group(model, group) # Check for disparate impact (80% rule) if self.has_disparate_impact(group_outcomes): return BiasResult( has_discriminatory_bias=True, affected_group=group ) return BiasResult(has_discriminatory_bias=False)
Model Performance Monitoring
class ModelPerformanceMonitor: async def monitor_production(self, model_id: str): """Continuous production monitoring""" while True: # Accuracy drift detection current_accuracy = await self.measure_accuracy(model_id) baseline = await self.get_baseline_accuracy(model_id) if self.is_significant_drift(current_accuracy, baseline): await self.alert_model_owner( model_id, reason="Significant accuracy drift detected" ) # Input distribution shift distribution_shift = await self.detect_distribution_shift(model_id) if distribution_shift.significant: await self.alert_model_owner( model_id, reason="Significant input distribution shift" ) # AML/KYC compliance monitoring if model_id in self.aml_models: aml_accuracy = await self.measure_aml_accuracy(model_id) if aml_accuracy < self.required_aml_accuracy: await self.escalate_critical_alert(model_id) # Challenger model evaluation if await self.has_challenger_model(model_id): comparison = await self.compare_with_challenger(model_id) if comparison.challenger_superior: await self.initiate_model_replacement(model_id) # Sleep and repeat await asyncio.sleep(3600) # Check hourly
9.3 Financial Services Agent with Full Compliance
class FinancialServicesAgent: def __init__(self, config: FinancialAgentConfig): self.config = config self.compliance = ComplianceEngine() self.aml = AMLScreener() self.audit = AuditLog() self.model_risk = ModelRiskEngine() async def execute(self, request: AgentRequest, user_context: UserContext) -> Response: """Execute with full financial compliance""" execution_id = generate_uuid() # 1. Model risk assessment model_risk = await self.model_risk.assess_execution_risk( model_id=self.config.model_id, input_size=len(request.data), output_impact=request.potential_impact ) if model_risk.unacceptable: await self.audit.log_execution_blocked( execution_id, user_context, model_risk.reason ) raise ModelRiskError(model_risk.reason) # 2. Regulatory checks regulatory = await self.compliance.check_regulatory( user_context, request.action ) if not regulatory.allowed: await self.audit.log_regulatory_violation( execution_id, user_context, regulatory.reason ) raise ComplianceViolation(regulatory.reason) # 3. AML/KYC screening aml_check = await self.aml.screen_user(user_context.user_id) if not aml_check.passed: await self.audit.log_aml_alert( execution_id, user_context.user_id, aml_check.reason ) raise AMLViolation(aml_check.reason) # 4. Transaction-specific checks (if applicable) if hasattr(request, 'transaction'): tx_check = await self.validate_transaction( request.transaction, user_context ) if not tx_check.valid: return Response(success=False, error=tx_check.reason) # 5. Execute with full audit trail try: result = await self.agent_runtime.execute( request=request, user_context=user_context, execution_id=execution_id ) # 6. Post-execution reporting await self.compliance.report_execution( execution_id, user_context, request, result ) # 7. Regulatory reporting if applicable if result.requires_regulatory_reporting: await self.compliance.submit_regulatory_report( execution_id, result ) return Response(success=True, data=result.output) except Exception as e: await self.audit.log_execution_error( execution_id, user_context.user_id, str(e) ) raise
Part 10: Reference Implementation
10.1 Important Caveat: Illustrative Reference Implementation
This Python implementation is intentionally simplified for pedagogical purposes.
It omits:
Provider-specific API implementations
Persistence layers and database details
Distributed locking and consensus
Secret management and key rotation
Complete network security
Comprehensive error typing
Production logging infrastructure
Feature flags and configuration management
Metrics export formatters
10.2 Core Agent Structure
from typing import Dict, List, Optional, Anyfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom enum import Enumimport asyncioimport logging
class ExecutionStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" TIMEOUT = "timeout"
@dataclassclass ExecutionTrace: execution_id: str agent_id: str user_id: str goal: str status: ExecutionStatus iterations: List[Dict] = field(default_factory=list) result: Optional[str] = None error: Optional[str] = None started_at: datetime = field(default_factory=datetime.utcnow) completed_at: Optional[datetime] = None @property def duration_seconds(self) -> float: end = self.completed_at or datetime.utcnow() return (end - self.started_at).total_seconds()
python
class ProductionAgent: def __init__(self, config: Dict): self.config = config self.logger = logging.getLogger(f"agent.{config['name']}") async def execute(self, goal: str, user_context: Dict, max_iterations: int = 10) -> ExecutionTrace: """Main agent execution loop""" trace = ExecutionTrace( execution_id=self.generate_uuid(), agent_id=self.config['name'], user_id=user_context['user_id'], goal=goal, status=ExecutionStatus.PENDING, iterations=[] ) try: trace.status = ExecutionStatus.RUNNING self.logger.info(f"Starting {trace.execution_id}: {goal}") # Phase 1: Perception perception = await self.perceive(goal, user_context) # Phase 2: Planning plan = await self.plan(goal, perception, user_context) # Phase 3: Execution loop for iteration in range(max_iterations): # Reasoning reasoning = await self.reason(goal, plan, iteration, perception) # Select action action = self.select_action(plan, reasoning, iteration) if action['type'] == 'complete': trace.result = action.get('result') trace.status = ExecutionStatus.SUCCESS break elif action['type'] == 'tool_call': # Authorization happens BEFORE execution if not user_context.get('can_use_tool')(action['tool']): trace.status = ExecutionStatus.FAILED trace.error = f"Not authorized for {action['tool']}" break # Execute tool tool_result = await self.execute_tool( action['tool'], action['args'], user_context ) perception.update(tool_result) elif action['type'] == 'escalate': trace.result = f"Escalated: {action['reason']}" break # Record iteration trace.iterations.append({ 'iteration': iteration, 'action': action['type'], 'timestamp': datetime.utcnow() }) if trace.status == ExecutionStatus.RUNNING: trace.status = ExecutionStatus.TIMEOUT trace.error = "Max iterations exceeded" return trace except Exception as e: trace.status = ExecutionStatus.FAILED trace.error = str(e) self.logger.error(f"Execution {trace.execution_id} failed: {e}") raise finally: trace.completed_at = datetime.utcnow() await self.persist_trace(trace) async def perceive(self, goal: str, user_context: Dict) -> Dict: """Gather contextual data""" return {"goal": goal, "context": {}} async def plan(self, goal: str, perception: Dict, user_context: Dict) -> Dict: """Generate execution plan""" return {"steps": [], "contingencies": []} async def reason(self, goal: str, plan: Dict, iteration: int, perception: Dict) -> Dict: """Analyze situation""" return { "summary": f"Iteration {iteration}", "confidence": 0.8, "options": [] } def select_action(self, plan: Dict, reasoning: Dict, iteration: int) -> Dict: """Choose next action""" return {"type": "complete", "result": "Goal achieved"} async def execute_tool(self, tool_name: str, args: Dict, user_context: Dict) -> Any: """Execute tool""" return {"success": True, "output": ""} async def persist_trace(self, trace: ExecutionTrace): """Store execution trace""" pass def generate_uuid(self) -> str: """Generate unique ID""" import uuid return str(uuid.uuid4())
python
async def main(): config = { "name": "example_agent", "domain": "example", "mission": "Example mission" } agent = ProductionAgent(config) user_context = { "user_id": "user_123", "can_use_tool": lambda tool: True } result = await agent.execute( goal="Complete example task", user_context=user_context ) print(f"Execution {result.execution_id}: {result.status.value}") print(f"Duration: {result.duration_seconds}s")
Part 11: What NOT to Automate (Financial Services)
Do NOT give unrestricted autonomy for:
Final credit decisions → AI recommends → Deterministic validation → Human/policy approval → Execution
Suspicious activity determinations → AML flags → Review → Human analyst judgment → Regulatory filing
Regulatory filings without validation → Model generates → Compliance review → Legal review → Authorized signature
Irreversible financial transactions → AI recommends → Policy check → Human approval (with timeouts) → Execution
Account closure → AI recommends → Customer contact → Human verification → Execution
Enterprise agentic AI represents a shift from primarily request-response applications toward systems that can dynamically reason about goals, select actions, use tools, maintain state and operate within defined control boundaries.
This shift brings tremendous capability but substantial complexity and risk. Success requires:
Security-first architecture: Authorization before execution, zero-trust assumptions