--- title: "RAG and Evaluation for Financial Agents" url: "https://vanducng.dev/2026/01/21/RAG-and-Evaluation-for-Financial-Agents/" date: "2026-01-21" series: "finance-ai" categories: ["AI"] tags: ["llm","python","rag","finance"] --- # RAG and Evaluation for Financial Agents Financial agents need two critical capabilities that set them apart from simple chatbots: the ability to ground responses in authoritative documents through intelligent retrieval, and continuous evaluation to ensure accuracy in regulated environments. Agentic RAG transforms retrieval from a passive lookup into an active reasoning loop, while long-term memory enables personalization across sessions. Together with robust evaluation frameworks, these capabilities create production-ready financial assistants. ## From Basic RAG to Agentic RAG Traditional Retrieval-Augmented Generation follows a simple pattern: accept a query, retrieve documents, generate an answer. This works for straightforward questions but falls short in complex financial scenarios where initial retrieval might miss critical context. ```mermaid flowchart TB subgraph Agentic["Agentic RAG"] direction LR Q2[Query] --> R2[Retrieve] R2 --> E[Evaluate] E --> D{Sufficient?} D -->|No| RF[Reformulate] RF --> R2 D -->|Yes| G2[Generate] G2 --> A2[Answer] end subgraph Traditional["Traditional RAG"] direction LR Q1[Query] --> R1[Retrieve Once] R1 --> G1[Generate] G1 --> A1[Answer] end classDef pinkClass fill:#E74C3C,stroke:#333,stroke-width:2px,color:#fff classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff class Traditional pinkClass class Agentic greenClass ``` Agentic RAG introduces a feedback loop where the agent: - Assesses whether retrieved documents are relevant - Identifies missing information - Reformulates queries when needed - Retries retrieval with improved strategies before answering ### The Retrieve-Reason-Retry Pattern Consider a compliance question about investment restrictions. A basic RAG might surface general policy documents and produce a vague answer. An agentic RAG agent would: 1. **Retrieve** initial documents about investment policies 2. **Reason** about what's missing (specific asset class restrictions, regulatory citations) 3. **Retry** with refined queries targeting compliance frameworks ```python @dataclass class RAGResponse: """Response from an agentic RAG agent""" query: str answer: str sources: List[str] retrieved_chunks: int needed_retry: bool confidence: str # "high", "medium", "low" ``` ### Financial Policy Assistant Here's how an agentic RAG agent handles employee benefits queries: ```python class PolicyRAGAgent: """Autonomous RAG agent with self-improvement capabilities""" def __init__(self, collection): self.collection = collection self.llm = OpenAI() def process_query(self, question: str) -> RAGResponse: """Full RAG workflow with reflection and retry""" # Step 1: Initial retrieval docs = self.retrieve_documents(question) # Step 2: Generate initial answer answer = self.generate_answer(question, docs) # Step 3: Self-evaluate - does answer need more context? if self.should_retry(question, answer, docs): # Step 4: Improve the query and retry better_query = self.improve_query(question, answer) docs = self.retrieve_documents(better_query) answer = self.generate_answer(question, docs) needed_retry = True else: needed_retry = False # Step 5: Assess confidence confidence = self.assess_confidence(answer, docs) return RAGResponse( query=question, answer=answer, sources=[d.metadata['source'] for d in docs], retrieved_chunks=len(docs), needed_retry=needed_retry, confidence=confidence ) def should_retry(self, question: str, answer: str, docs: List) -> bool: """Use LLM to evaluate if answer is sufficient""" prompt = f""" Question: {question} Answer: {answer} Documents used: {len(docs)} Is this answer complete and well-grounded? Consider: - Does it directly address the question? - Are there gaps in the information? - Does it rely on speculation vs. retrieved facts? Respond with just YES or NO. """ response = self.llm.complete(prompt) return "NO" in response.upper() def improve_query(self, original: str, initial_answer: str) -> str: """Generate a better search query based on gaps""" prompt = f""" Original question: {original} Initial answer (incomplete): {initial_answer} Generate a more specific search query that would find the missing information. Focus on what the answer lacks. """ return self.llm.complete(prompt).strip() ``` ### Handling Out-of-Scope Questions A well-designed agentic RAG gracefully handles questions outside its knowledge base: ```python def answer_with_boundaries(self, question: str) -> RAGResponse: """Answer with awareness of knowledge boundaries""" response = self.process_query(question) # If confidence is low after retry, acknowledge limitations if response.needed_retry and response.confidence == "low": response.answer = ( "I don't have enough information in the policy documents " "to answer that question. This may require consulting " "with the relevant department directly." ) return response ``` ## Long-Term Memory for Financial Agents While short-term memory maintains coherence within a session, long-term memory enables agents to recall facts and preferences across sessions - essential for personalized financial advice. ```mermaid flowchart TD subgraph Memory["Long-Term Memory Types"] S[Semantic Memory Facts & Preferences] E[Episodic Memory Past Interactions] P[Procedural Memory Behavioral Adaptations] end U[User Interaction] --> S U --> E U --> P S --> R[Recall for Future Sessions] E --> R P --> R classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff class S blueClass class E orangeClass class P greenClass ``` ### Three Types of Long-Term Memory **Semantic Memory** stores factual information learned from users: - Client names and account numbers - Risk tolerance preferences - Investment goals and constraints **Episodic Memory** captures past interactions: - Previous advisory conversations - Successful problem resolutions - Transaction history context **Procedural Memory** encodes behavioral adaptations: - Preferred communication style (formal vs. casual) - Report formatting preferences - Escalation thresholds ### Memory-Enabled Treasury Assistant ```python from sqlalchemy import Column, String, DateTime, Float, Integer, Text, JSON from sqlalchemy.orm import declarative_base from datetime import datetime, timedelta Base = declarative_base() class MemoryEntry(Base): """Database model for agent memory""" __tablename__ = 'treasury_memory' id = Column(String(36), primary_key=True) # Core memory data topic = Column(String(100), nullable=False, index=True) fact_text = Column(Text, nullable=False) source = Column(String(50), nullable=False) # "cfo", "policy", "interaction" # Temporal metadata created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, onupdate=datetime.utcnow) ttl_days = Column(Integer, nullable=True) # Time-to-live # Prioritization weight = Column(Float, default=1.0) pinned = Column(Boolean, default=False) access_count = Column(Integer, default=1) # Organization tags = Column(JSON, default=list) def is_expired(self) -> bool: """Check if memory has expired based on TTL""" if self.ttl_days is None: return False expiry = self.created_at + timedelta(days=self.ttl_days) return datetime.utcnow() > expiry @dataclass class MemoryPolicy: """Configuration for memory management""" # TTL by topic (days) - None means permanent ttl_policies: Dict[str, Optional[int]] = field(default_factory=lambda: { 'cash_policy': None, # Permanent 'investment_rules': None, # Permanent 'counterparty_limits': None, # Permanent 'market_opportunities': 30, # Expire after 30 days 'temporary_approvals': 90, # Expire after 90 days }) # Weighting formula recency_boost: float = 0.1 frequency_boost: float = 0.05 pinned_boost: float = 2.0 max_digest_size: int = 12 ``` ### Intelligent Memory Manager The memory manager uses LLM to canonicalize facts for consistency: ```python class TreasuryMemoryManager: """Intelligent memory manager with LLM integration""" def __init__(self, policy: MemoryPolicy = None): self.policy = policy or MemoryPolicy() self.session = SessionLocal() def canonicalize_fact(self, fact_text: str, topic: str) -> str: """Normalize fact text for consistent storage""" prompt = f""" Canonicalize this {topic} policy into a clear statement: Original: {fact_text} Rules: - Use standard financial terminology - Format amounts as $X,XXX,XXX or $XM - Use standard percentage format - Keep compliance language precise - Maintain core meaning Canonicalized fact: """ response = self.llm.complete(prompt, temperature=0.1) return response.strip() def upsert_memory(self, topic: str, fact: str, source: str = "interaction") -> MemoryEntry: """Add or update a memory entry""" # Canonicalize for consistency canonical_fact = self.canonicalize_fact(fact, topic) # Check for existing similar memory existing = self.find_similar(canonical_fact, topic) if existing: # Update existing memory existing.fact_text = canonical_fact existing.access_count += 1 existing.updated_at = datetime.utcnow() return existing # Create new memory entry = MemoryEntry( id=str(uuid.uuid4()), topic=topic, fact_text=canonical_fact, source=source, ttl_days=self.policy.ttl_policies.get(topic) ) self.session.add(entry) self.session.commit() return entry def get_policy_digest(self, max_items: int = None) -> List[str]: """Get prioritized list of active policies""" max_items = max_items or self.policy.max_digest_size # Query non-expired memories memories = self.session.query(MemoryEntry).all() active = [m for m in memories if not m.is_expired()] # Score and sort by priority scored = [] for m in active: score = self.calculate_priority(m) scored.append((score, m)) scored.sort(reverse=True, key=lambda x: x[0]) # Return top items with pinned indicator result = [] for _, m in scored[:max_items]: prefix = "📌 " if m.pinned else "" result.append(f"{prefix}{m.fact_text}") return result def calculate_priority(self, memory: MemoryEntry) -> float: """Calculate dynamic priority score""" base = memory.weight # Recency boost days_old = (datetime.utcnow() - memory.created_at).days recency = self.policy.recency_boost * max(0, 30 - days_old) # Frequency boost frequency = self.policy.frequency_boost * memory.access_count # Pinned boost pinned = self.policy.pinned_boost if memory.pinned else 0 return base + recency + frequency + pinned ``` ### Memory Scoping Memories can be scoped at different levels: ```python class ScopedMemoryManager: """Memory manager with access scoping""" def retrieve_memories(self, user_id: str, team_id: str = None, include_global: bool = True) -> List[MemoryEntry]: """Retrieve memories at appropriate scope levels""" memories = [] # User-specific memories (highest priority) memories.extend( self.get_by_scope("user", user_id) ) # Team-shared memories if team_id: memories.extend( self.get_by_scope("team", team_id) ) # Global organizational memories if include_global: memories.extend( self.get_by_scope("global", "org") ) return self.deduplicate(memories) ``` ## Agent Evaluation Building agents is only half the challenge - ensuring they work correctly, safely, and efficiently requires systematic evaluation. This is especially critical in financial services where errors have regulatory and monetary consequences. ### Four Dimensions of Evaluation ```mermaid flowchart TD E[Agent Evaluation] --> TC[Task Completion] E --> QC[Quality Control] E --> TI[Tool Interaction] E --> SM[System Metrics] TC --> TC1[Goal achieved?] TC --> TC2[Steps required?] TC --> TC3[Human intervention?] QC --> QC1[Correct format?] QC --> QC2[Instructions followed?] QC --> QC3[Context used?] TI --> TI1[Right tools?] TI --> TI2[Valid arguments?] TI --> TI3[Appropriate sequence?] SM --> SM1[Latency] SM --> SM2[Token usage] SM --> SM3[Error rate] classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff classDef pinkClass fill:#E74C3C,stroke:#333,stroke-width:2px,color:#fff class TC blueClass class QC orangeClass class TI greenClass class SM pinkClass ``` ### Three Evaluation Strategies **1. Final Response Evaluation** Judges only the end result - ideal for end-to-end testing but provides no insight into how the result was produced. ```python def evaluate_final_response(agent_answer: str, correct_answer: str, question: str) -> Dict: """Black-box evaluation of final output""" prompt = f""" Evaluate the agent's answer against the correct answer. Question: {question} Correct Answer: {correct_answer} Agent Answer: {agent_answer} Score accuracy 0-100 based on: - 90-100: Fully correct, all key facts present - 70-89: Mostly correct, minor omissions - 50-69: Partially correct, significant gaps - Below 50: Incorrect or misleading Return JSON: {{"score": , "reasoning": ""}} """ return llm.complete(prompt, response_format="json") ``` **2. Single-Step Evaluation** Focuses on individual decisions - fast and ideal for debugging specific issues. ```python def evaluate_tool_choice(context: str, available_tools: List[str], chosen_tool: str, expected_tool: str) -> Dict: """Evaluate a single tool selection decision""" return { "correct": chosen_tool == expected_tool, "chosen": chosen_tool, "expected": expected_tool, "context_summary": context[:200] } ``` **3. Trajectory Evaluation** Reviews the full execution path - richest insight but requires more setup. ```python def evaluate_trajectory(execution_trace: List[Dict], expected_trajectory: List[Dict]) -> Dict: """Evaluate complete agent execution path""" results = { "tool_accuracy": 0, "argument_accuracy": 0, "sequence_score": 0, "efficiency_score": 0 } # Compare tool calls for i, (actual, expected) in enumerate( zip(execution_trace, expected_trajectory) ): if actual['tool'] == expected['tool']: results['tool_accuracy'] += 1 if actual.get('args') == expected.get('args'): results['argument_accuracy'] += 1 # Normalize scores total = len(expected_trajectory) results['tool_accuracy'] /= total results['argument_accuracy'] /= total # Efficiency: fewer steps is better results['efficiency_score'] = min(1.0, total / len(execution_trace)) return results ``` ### Financial Agent Evaluation Framework For insurance claims processing, we evaluate three weighted metrics: ```python class InsuranceAgentEvaluator: """Evaluation framework for insurance claims assistant""" def evaluate_complete_response(self, agent_response: Dict, gold_item: Dict) -> Dict: """Comprehensive evaluation with weighted scoring""" # Metric 1: Factual Accuracy (40% weight) accuracy = self.evaluate_factual_accuracy( agent_response['answer'], gold_item['correct_answer'], gold_item['question'] ) # Metric 2: Citation Compliance (30% weight) citation = self.evaluate_citation_compliance( agent_response['answer'], gold_item['requires_citation'], agent_response.get('sources', []) ) # Metric 3: Retrieval Relevance (30% weight) retrieval = self.evaluate_retrieval_relevance( gold_item['question'], agent_response['retrieved_docs'], gold_item['expected_doc_ids'] ) # Weighted composite score composite = ( accuracy['score'] * 0.40 + citation['score'] * 0.30 + retrieval['score'] * 0.30 ) return { 'composite_score': composite, 'factual_accuracy': accuracy, 'citation_compliance': citation, 'retrieval_relevance': retrieval } def evaluate_citation_compliance(self, answer: str, should_cite: bool, sources: List[str]) -> Dict: """Check for proper source attribution""" citation_patterns = [ r'\[.*?\]', # [Source Name] r'according to', # According to policy... r'source:', # Source: Document r'per the', # Per the guidelines r'as stated in' # As stated in... ] citations_found = [] for pattern in citation_patterns: matches = re.findall(pattern, answer, re.IGNORECASE) citations_found.extend(matches) has_citations = len(citations_found) > 0 if should_cite: score = 100 if has_citations else 0 else: score = 100 # No citation required return { 'score': score, 'citations_found': citations_found, 'sources_used': sources } def evaluate_retrieval_relevance(self, question: str, retrieved_docs: List, expected_doc_ids: List[str]) -> Dict: """Measure retrieval quality with precision/recall""" retrieved_ids = [doc.metadata.get('doc_id') for doc in retrieved_docs] expected_set = set(expected_doc_ids) retrieved_set = set(retrieved_ids) if retrieved_set: precision = len(expected_set & retrieved_set) / len(retrieved_set) else: precision = 0 if expected_set: recall = len(expected_set & retrieved_set) / len(expected_set) else: recall = 1.0 # F1 score if precision + recall > 0: f1 = 2 * (precision * recall) / (precision + recall) else: f1 = 0 return { 'score': f1 * 100, 'precision': precision, 'recall': recall, 'retrieved_count': len(retrieved_ids), 'expected_count': len(expected_doc_ids) } ``` ### Running Evaluation Suites ```python def run_evaluation_suite(agent, golden_dataset: List[Dict]) -> Dict: """Execute full evaluation across test dataset""" evaluator = InsuranceAgentEvaluator() results = [] for item in golden_dataset: # Get agent response response = agent.answer_question(item['question']) # Evaluate eval_result = evaluator.evaluate_complete_response(response, item) eval_result['question_id'] = item['id'] eval_result['category'] = item['category'] results.append(eval_result) # Aggregate metrics df = pd.DataFrame(results) return { 'avg_composite': df['composite_score'].mean(), 'avg_accuracy': df['factual_accuracy'].apply(lambda x: x['score']).mean(), 'avg_citation': df['citation_compliance'].apply(lambda x: x['score']).mean(), 'avg_retrieval': df['retrieval_relevance'].apply(lambda x: x['score']).mean(), 'results_by_category': df.groupby('category')['composite_score'].mean().to_dict(), 'detailed_results': results } ``` ### Generating Improvement Recommendations Evaluation drives improvement by identifying specific weaknesses: ```python def generate_recommendations(metrics: Dict, results_df) -> List[str]: """Generate actionable recommendations from evaluation results""" recommendations = [] # Low retrieval relevance if metrics['avg_retrieval'] < 70: recommendations.append( "RETRIEVAL: Improve document chunking strategy. Current chunks " "may be too large or missing key metadata. Consider semantic " "chunking based on policy sections." ) # Low citation compliance if metrics['avg_citation'] < 80: recommendations.append( "CITATIONS: Add explicit citation instructions to system prompt. " "Require agent to reference specific policy documents for all " "regulatory guidance." ) # Low accuracy on specific categories for category, score in metrics['results_by_category'].items(): if score < 75: recommendations.append( f"CATEGORY-{category.upper()}: Performance is below threshold. " f"Consider adding more {category} examples to training data " f"or improving {category}-specific retrieval." ) return recommendations ``` ## Takeaways 1. **Agentic RAG transforms retrieval** from passive lookup to active reasoning with reflect-reformulate-retry loops 2. **Self-evaluation enables quality control** - agents assess their own answers and retry when confidence is low 3. **Long-term memory creates personalization** through three types: semantic (facts), episodic (interactions), and procedural (behaviors) 4. **Memory management requires policies** - TTL settings, priority scoring, and scope levels prevent bloat and maintain relevance 5. **Evaluation spans four dimensions**: task completion, quality control, tool interaction, and system metrics 6. **Three evaluation strategies** serve different needs: final response for end-to-end, single-step for debugging, trajectory for comprehensive analysis 7. **Weighted composite scores** provide holistic assessment while highlighting specific areas for improvement --- *This is the eleventh post in my Applied Agentic AI for Finance series. Next: [Multi-Agent Architecture for Trading](https://vanducng.dev/2026/01/24/Multi-Agent-Architecture-for-Trading/) where we'll explore coordinating multiple specialized agents for complex financial operations.*