State and Memory for Trading Agents

LLMs are stateless by nature - each interaction is isolated, with no memory of prior prompts. But financial agents often need context to manage complex, multi-step tasks like loan approvals, insurance claims, or trading workflows. This requires two complementary mechanisms: state for tracking execution progress, and memory for maintaining conversational context across interactions.

The Stateless Problem

Consider a loan approval agent that needs to:

1. Verify applicant documents
2. Get AI risk assessment
3. Make final approval decision

In a stateless system, each step is isolated. The risk assessment step doesn’t know what documents were verified. The decision step can’t access the risk assessment. Information is lost between stages, making the system brittle and unreliable.

flowchart TB
    subgraph Stateless["Stateless: Information Lost"]
        direction LR
        S1[Step 1
Verify Docs] -.-> |Lost| S2[Step 2
Risk Check]
        S2 -.-> |Lost| S3[Step 3
Decision]
    end

    subgraph Stateful["Stateful: Context Preserved"]
        direction LR
        C[Shared Context] --> A1[Step 1]
        C --> A2[Step 2]
        C --> A3[Step 3]
        A1 --> C
        A2 --> C
        A3 --> C
    end

    classDef pinkClass fill:#E74C3C,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class Stateless pinkClass
    class Stateful greenClass

Agent State: The Working Memory

Agent state includes everything the agent knows during one execution:

• The original user input
• System instructions
• Message history
• Tool calls (pending or completed)
• Intermediate results from prior steps

This state is ephemeral - it only exists while the task is running. Think of it like working memory: tracking progress and context throughout execution, then cleared when the task ends.

State Machines for Agent Execution

Agents can be modeled as state machines that transition through defined steps:

flowchart TD
    START[Receive Query] --> PREP[Prepare Messages]
    PREP --> LLM[Call LLM]
    LLM --> CHECK{Tools Needed?}
    CHECK -->|Yes| EXEC[Execute Tools]
    EXEC --> UPDATE[Update State]
    UPDATE --> LLM
    CHECK -->|No| END[Complete Task]

    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff

    class CHECK orangeClass

Each step receives the current state, processes it, and returns an updated version. Transitions between steps depend on state contents.

Applied Example: Insurance Claim Processing

Let’s build a stateful system for processing insurance claims through four stages: intake, assessment, verification, and decision.

Defining States and Context

from enum import Enum
from dataclasses import dataclass, field
from typing import List

class ClaimState(Enum):
    INTAKE = "intake"
    ASSESSMENT = "assessment"
    VERIFICATION = "verification"
    DECISION = "decision"
    COMPLETED = "completed"

@dataclass
class ClaimContext:
    """Claim state maintained across processing steps"""
    claim_id: str
    customer: str
    claim_type: str
    amount: float
    description: str
    policy_active: bool

    # Results from each step
    intake_valid: bool = False
    ai_assessment: str = ""
    fraud_score: float = 0.0
    final_decision: str = ""

    # Workflow tracking
    current_state: ClaimState = ClaimState.INTAKE
    state_history: List[str] = field(default_factory=list)

    def transition(self, new_state: ClaimState):
        """Record state transition for audit trail"""
        self.state_history.append(
            f"{self.current_state.value} → {new_state.value}"
        )
        self.current_state = new_state

The ClaimContext acts as a centralized container for all claim information. The transition method creates an audit trail of state changes.

The Claims Processor

class ClaimProcessor:
    """Stateful insurance claim processing"""

    def __init__(self, claim: dict):
        self.context = ClaimContext(
            claim_id=claim["claim_id"],
            customer=claim["customer"],
            claim_type=claim["claim_type"],
            amount=claim["amount"],
            description=claim["description"],
            policy_active=claim["policy_active"]
        )

    def intake(self) -> bool:
        """Step 1: Validate claim information"""
        print(f"1️⃣ INTAKE: Validating claim {self.context.claim_id}...")

        # Validation logic
        self.context.intake_valid = (
            self.context.amount > 0 and
            self.context.policy_active
        )

        if not self.context.intake_valid:
            return False

        self.context.transition(ClaimState.ASSESSMENT)
        return True

    def assess(self):
        """Step 2: AI risk assessment"""
        print("2️⃣ ASSESSMENT: AI analyzing claim...")

        prompt = f"""Assess the risk of this insurance claim:

Customer: {self.context.customer}
Claim Type: {self.context.claim_type}
Amount: ${self.context.amount:,.0f}
Description: {self.context.description}

Provide risk level (LOW/MEDIUM/HIGH) and key red flags."""

        response = llm.complete(prompt)
        self.context.ai_assessment = response
        self.context.fraud_score = 0.3 if "HIGH" in response else 0.1

        print(f"   Assessment: {response[:100]}...")
        self.context.transition(ClaimState.VERIFICATION)

    def verify(self):
        """Step 3: Policy rule verification"""
        print("3️⃣ VERIFICATION: Checking policy rules...")

        # Use data from previous steps
        is_new_customer = "new customer" in self.context.description.lower()
        is_high_amount = self.context.amount > 25000
        is_high_fraud = self.context.fraud_score > 0.25

        print(f"   • New customer: {is_new_customer}")
        print(f"   • High amount: {is_high_amount}")
        print(f"   • High fraud risk: {is_high_fraud}")

        self.context.transition(ClaimState.DECISION)

    def decide(self):
        """Step 4: Final decision using all context"""
        print("4️⃣ DECISION: Making approval decision...")

        # Combine insights from all previous steps
        if self.context.fraud_score > 0.25:
            self.context.final_decision = "DENY"
        elif self.context.amount > 25000 and "new" in self.context.description.lower():
            self.context.final_decision = "REQUEST MORE DOCUMENTATION"
        else:
            self.context.final_decision = "APPROVE"

        print(f"   Decision: {self.context.final_decision}")
        self.context.transition(ClaimState.COMPLETED)

    def process(self):
        """Run complete workflow"""
        print(f"\n{'='*50}")
        print(f"INSURANCE CLAIM: {self.context.customer}")
        print(f"{'='*50}")

        if self.intake():
            self.assess()
            self.verify()
            self.decide()
        else:
            print("❌ Claim rejected at intake stage")

        # Show audit trail
        print(f"\n📋 Workflow State History:")
        for step in self.context.state_history:
            print(f"   • {step}")

Running the Processor

claims = [
    {
        "claim_id": "CLM-001",
        "customer": "John Smith",
        "claim_type": "Auto Accident",
        "amount": 5000,
        "description": "Minor fender bender. Two-year customer, clean record.",
        "policy_active": True
    },
    {
        "claim_id": "CLM-002",
        "customer": "Jane Doe",
        "claim_type": "Water Damage",
        "amount": 35000,
        "description": "Pipe burst in basement. New customer (1 month).",
        "policy_active": True
    }
]

for claim in claims:
    processor = ClaimProcessor(claim)
    processor.process()

Output shows how each claim flows through states with preserved context:

==================================================
INSURANCE CLAIM: John Smith
==================================================
1️⃣ INTAKE: Validating claim CLM-001...
2️⃣ ASSESSMENT: AI analyzing claim...
   Assessment: Risk Assessment: LOW...
3️⃣ VERIFICATION: Checking policy rules...
   • New customer: False
   • High amount: False
   • High fraud risk: False
4️⃣ DECISION: Making approval decision...
   Decision: APPROVE

📋 Workflow State History:
   • intake → assessment
   • assessment → verification
   • verification → decision
   • decision → completed

State vs Memory: Understanding the Difference

While both provide context, state and memory serve different purposes:

Aspect	State	Short-Term Memory	Long-Term Memory
Scope	Single execution	Single session	Across sessions
Duration	Task lifetime	Conversation lifetime	Persistent
Tracked By	Execution ID	Session ID	User ID
Contains	Tool calls, transitions	Conversation turns	User preferences
Use Case	Workflow steps	Dialogue continuity	Personalization

flowchart TD
    subgraph State["State (Execution)"]
        E1[Step 1] --> E2[Step 2]
        E2 --> E3[Step 3]
    end

    subgraph STM["Short-Term Memory (Session)"]
        T1[Turn 1] --> T2[Turn 2]
        T2 --> T3[Turn 3]
    end

    subgraph LTM["Long-Term Memory (User)"]
        S1[Session 1] --> S2[Session 2]
        S2 --> S3[Session 3]
    end

    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 State blueClass
    class STM orangeClass
    class LTM greenClass

Short-Term Memory: Simulating Continuity

LLMs don’t have real memory - what looks like memory is constructed by including previous interactions in each prompt. This creates the illusion of continuity.

Memory Strategies

1. Full Conversation History
Send all previous messages with each new input:

messages = [
    {"role": "user", "content": "I'm 35 years old..."},
    {"role": "assistant", "content": "Great! Let me help..."},
    {"role": "user", "content": "I have $50K saved..."},
    {"role": "assistant", "content": "With your savings..."},
    {"role": "user", "content": "What about bonds?"}  # Current turn
]

Pros: Full context preserved
Cons: Token-heavy, may exceed context window

2. Sliding Window
Keep only the most recent N turns:

def add_turn(self, user_input: str, response: str):
    """Add turn with sliding window"""
    self.history.append({
        "user": user_input,
        "assistant": response
    })

    # Keep only last 6 turns
    if len(self.history) > self.max_history:
        self.history = self.history[-self.max_history:]

Pros: Bounded token usage
Cons: Loses earlier context

3. Summarization
Condense earlier messages into compact summaries:

def update_summary(self, new_input: str):
    """Update running summary of key facts"""

    prompt = f"""Current summary:
{self.session_summary}

New input: "{new_input}"

Update summary to include new information.
Keep 3-5 key facts. Return bullet points only."""

    self.session_summary = llm.complete(prompt)

Pros: Efficient, retains essentials
Cons: May lose nuance

Applied Example: Financial Advisor with Memory

Let’s build a financial advisor that remembers client details throughout a conversation.

Multi-Component Memory System

from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional
from datetime import datetime

@dataclass
class ClientProfile:
    """Structured client information"""
    client_id: Optional[str] = None
    age: Optional[int] = None
    annual_income: Optional[str] = None
    risk_tolerance: Optional[str] = None  # conservative/moderate/aggressive
    investment_goals: List[str] = field(default_factory=list)
    current_holdings: Optional[str] = None
    time_horizon: Optional[str] = None

    def get_captured_info(self) -> Dict:
        """Return non-empty profile fields"""
        return {
            field: value
            for field, value in asdict(self).items()
            if value and (not isinstance(value, list) or len(value) > 0)
        }

@dataclass
class ConversationTurn:
    """Single conversation exchange"""
    client_input: str
    advisor_response: str
    timestamp: str


class FinancialAdvisorAssistant:
    """Financial advisor with short-term memory management"""

    def __init__(self, max_history: int = 6):
        # Three memory components
        self.profile = ClientProfile()           # Structured data
        self.conversation_history: List[ConversationTurn] = []  # Sliding window
        self.session_summary: List[str] = []     # Running summary
        self.max_history = max_history

    def add_turn(self, client_input: str, advisor_response: str):
        """Add conversation turn with sliding window"""
        turn = ConversationTurn(
            client_input=client_input,
            advisor_response=advisor_response,
            timestamp=datetime.now().isoformat()
        )
        self.conversation_history.append(turn)

        # Sliding window
        if len(self.conversation_history) > self.max_history:
            self.conversation_history = self.conversation_history[-self.max_history:]

    def update_session_summary(self, client_input: str):
        """Update running summary of client's situation"""

        current = "\n".join(f"- {fact}" for fact in self.session_summary)

        prompt = f"""Current session summary:
{current if current else 'None yet'}

Latest client input: "{client_input}"

Update summary to include new information. Keep 3-5 key facts about:
- Client's financial goals and timeline
- Risk preferences and concerns
- Important personal circumstances

Return only bullet points, one per line."""

        response = llm.complete(prompt, temperature=0.3)

        self.session_summary = [
            line.strip()
            for line in response.strip().split('\n')
            if line.strip()
        ][-5:]  # Keep max 5 facts

    def extract_profile(self, client_input: str):
        """Extract structured profile from client input"""

        prompt = f"""Extract client financial profile from: "{client_input}"

Current profile:
{json.dumps(self.profile.get_captured_info(), indent=2)}

Extract and return ONLY new information as valid JSON:
{{
    "age": number or null,
    "risk_tolerance": "conservative|moderate|aggressive" or null,
    "annual_income": string or null,
    "investment_goals": list of strings or null,
    "time_horizon": string or null
}}

Return valid JSON only."""

        try:
            response = llm.complete(prompt, response_format={"type": "json_object"})
            updates = json.loads(response)

            # Update profile with extracted values
            for field, value in updates.items():
                if value is not None and hasattr(self.profile, field):
                    setattr(self.profile, field, value)
        except Exception as e:
            print(f"Profile extraction error: {e}")

    def generate_response(self, client_input: str) -> str:
        """Generate contextual response using all memory"""

        # Build memory context
        memory_context = ""
        if self.session_summary:
            memory_context = "Key context from this session:\n"
            memory_context += "\n".join(f"- {fact}" for fact in self.session_summary)

        # Build conversation context
        conversation_context = ""
        if self.conversation_history:
            conversation_context = "Recent conversation:\n"
            for turn in self.conversation_history[-3:]:
                conversation_context += f"Client: {turn.client_input}\n"
                conversation_context += f"Advisor: {turn.advisor_response}\n"

        captured = self.profile.get_captured_info()

        prompt = f"""You are a professional financial advisor assistant.

{memory_context}

{conversation_context}

Client profile captured:
{json.dumps(captured, indent=2) if captured else 'None yet'}

Client just said: "{client_input}"

Guidelines:
1. Provide helpful, personalized financial guidance
2. Reference previous context naturally
3. NEVER ask for information you already have
4. Be concise and professional

Respond naturally as a financial advisor:"""

        return llm.complete(prompt, temperature=0.7)

    def process_input(self, client_input: str) -> str:
        """Process input through full memory pipeline"""

        # Update memory components
        self.update_session_summary(client_input)
        self.extract_profile(client_input)

        # Generate response
        response = self.generate_response(client_input)

        # Store turn
        self.add_turn(client_input, response)

        return response

Demo Conversation

def run_advisor_demo():
    assistant = FinancialAdvisorAssistant(max_history=4)

    conversation = [
        "Hi, I'm 35 years old and want to start investing for retirement. I have about $50K saved up.",
        "I'm pretty nervous about the stock market. I'd say I'm conservative with risk.",
        "My main goal is retirement in 30 years, but I also want to buy a house in 5 years.",
        "What do you recommend for someone like me?",
    ]

    print("💼 FINANCIAL ADVISOR CONVERSATION")

    for i, client_input in enumerate(conversation, 1):
        print(f"\n👤 CLIENT (Turn {i}): {client_input}")

        response = assistant.process_input(client_input)
        print(f"💰 ADVISOR: {response}")

        # Show memory state every 2 turns
        if i % 2 == 0:
            print("\n🧠 MEMORY STATE:")
            print(f"   Summary: {assistant.session_summary}")
            print(f"   Profile: {list(assistant.profile.get_captured_info().keys())}")
            print(f"   History: {len(assistant.conversation_history)} turns")

run_advisor_demo()

The advisor remembers everything: age (35), savings ($50K), risk tolerance (conservative), goals (retirement + house), and timeline (30 years, 5 years). No repeated questions, just personalized advice building on accumulated context.

Memory Architecture Patterns

Pattern 1: Layered Memory

flowchart TD
    INPUT[Client Input] --> PROFILE[Profile Extraction]
    INPUT --> SUMMARY[Summary Update]
    INPUT --> HISTORY[History Storage]

    PROFILE --> CONTEXT[Context Assembly]
    SUMMARY --> CONTEXT
    HISTORY --> CONTEXT

    CONTEXT --> LLM[LLM Response]
    LLM --> OUTPUT[Advisor Response]
    LLM --> HISTORY

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff

    class CONTEXT blueClass

Pattern 2: Memory for Trading Sessions

@dataclass
class TradingSessionMemory:
    """Memory for multi-turn trading assistance"""

    # Session context
    session_id: str
    started_at: datetime

    # Portfolio state
    current_positions: Dict[str, float] = field(default_factory=dict)
    pending_orders: List[Dict] = field(default_factory=list)

    # Conversation memory
    discussed_symbols: List[str] = field(default_factory=list)
    risk_warnings_given: List[str] = field(default_factory=list)

    # User preferences (extracted from conversation)
    preferred_sectors: List[str] = field(default_factory=list)
    risk_budget: Optional[float] = None

    def add_discussed_symbol(self, symbol: str):
        if symbol not in self.discussed_symbols:
            self.discussed_symbols.append(symbol)

    def get_context_summary(self) -> str:
        return f"""Session Context:
- Active positions: {list(self.current_positions.keys())}
- Discussed symbols: {self.discussed_symbols}
- Pending orders: {len(self.pending_orders)}
- Risk warnings given: {self.risk_warnings_given}"""

Financial-Specific Considerations

Audit Trail Requirements

Financial workflows need complete audit trails:

@dataclass
class AuditableContext:
    """Context with full audit trail for compliance"""

    workflow_id: str
    created_at: datetime

    # State history with timestamps
    transitions: List[Dict] = field(default_factory=list)

    def transition(self, from_state: str, to_state: str, reason: str):
        self.transitions.append({
            "timestamp": datetime.now().isoformat(),
            "from": from_state,
            "to": to_state,
            "reason": reason
        })

    def get_audit_log(self) -> str:
        """Generate compliance-ready audit log"""
        log = f"Workflow: {self.workflow_id}\n"
        log += f"Created: {self.created_at}\n\n"

        for t in self.transitions:
            log += f"[{t['timestamp']}] {t['from']} → {t['to']}\n"
            log += f"  Reason: {t['reason']}\n"

        return log

Memory Limits for Sensitive Data

class SecureMemory:
    """Memory with data retention policies"""

    def __init__(self, retention_seconds: int = 3600):
        self.retention_seconds = retention_seconds
        self.data: Dict[str, Tuple[Any, datetime]] = {}

    def store(self, key: str, value: Any):
        self.data[key] = (value, datetime.now())

    def retrieve(self, key: str) -> Optional[Any]:
        if key not in self.data:
            return None

        value, stored_at = self.data[key]

        # Check retention
        age = (datetime.now() - stored_at).total_seconds()
        if age > self.retention_seconds:
            del self.data[key]
            return None

        return value

    def clear_sensitive(self):
        """Clear all sensitive data"""
        sensitive_keys = [k for k in self.data.keys()
                        if "ssn" in k.lower() or "account" in k.lower()]
        for key in sensitive_keys:
            del self.data[key]

Takeaways

1. LLMs are stateless - state and memory must be explicitly managed to maintain context across interactions

2. State tracks execution progress - use state machines with context objects for multi-step workflows like loan approval or claims processing

3. Short-term memory simulates continuity - through full history, sliding windows, or summarization strategies

4. Multi-component memory works best - combine structured profiles, running summaries, and conversation history

5. Financial workflows need audit trails - every state transition should be logged with timestamps and reasons

6. Memory enables personalization - agents can provide context-aware, non-repetitive responses by remembering what clients have shared

---

This is the ninth post in my Applied Agentic AI for Finance series. Next: Connecting Agents to Financial Data Sources where we’ll explore integrating external APIs, web search, and databases.