---
title: "Prompt Chaining and Routing in Trading Systems"
url: "https://vanducng.dev/2026/01/03/Prompt-Chaining-and-Routing-in-Trading-Systems/"
date: "2026-01-03"
series: "finance-ai"
categories: ["AI"]
tags: ["llm","python","finance"]
---

# Prompt Chaining and Routing in Trading Systems

Trading systems process diverse request types - market orders, limit orders, options, compliance checks - each requiring specialized handling. Two fundamental patterns enable this: **prompt chaining** for sequential multi-step processing, and **routing** for intelligent task distribution. Mastering these patterns is essential for building robust financial workflows.

## Prompt Chaining: The Assembly Line

Prompt chaining decomposes complex tasks into a sequence of simpler steps, where each step's output becomes the next step's input. Like an assembly line where specialized stations perform focused tasks, prompt chains break down problems that would overwhelm a single LLM call.

```mermaid
flowchart LR
    I[Trade Request] --> S1[Parse Order]
    S1 --> V1{Valid?}
    V1 -->|Yes| S2[Risk Check]
    V1 -->|No| E1[Error Handler]
    S2 --> V2{Approved?}
    V2 -->|Yes| S3[Execute Trade]
    V2 -->|No| E2[Rejection Handler]
    S3 --> S4[Confirm & Log]
    S4 --> O[Confirmation]

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

    class V1,V2 orangeClass
```

### Why Chain Works Better Than Single Prompts

Consider processing a trading order with a single prompt:

```python
# Single prompt approach - problematic
single_prompt = """
Parse this trade request, validate the order details, check risk limits,
determine if margin requirements are met, execute the trade if approved,
generate confirmation, and log the transaction.

Request: {trade_request}
"""
```

Problems with this approach:
- **Cognitive overload**: Too many tasks reduce accuracy on each
- **Debugging difficulty**: Can't tell where failures occur
- **No intermediate validation**: Errors cascade undetected
- **No recovery points**: Must restart entire process on failure

With chaining:

```python
# Chained approach - each step focused
parse_prompt = "Extract order type, symbol, quantity, and price from: {request}"
validate_prompt = "Validate this order against market rules: {parsed_order}"
risk_prompt = "Check position limits and margin for: {validated_order}"
execute_prompt = "Generate execution instruction for: {approved_order}"
```

Each step can be validated, retried, or handled separately.

### The Error Propagation Problem

The critical challenge with chaining: errors compound. A small mistake in step 1 becomes a disaster by step 5.

```mermaid
flowchart LR
    S1[Step 1
Small Error] --> S2[Step 2
Amplified]
    S2 --> S3[Step 3
Worse]
    S3 --> S4[Step 4
Disaster]

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

    class S1 orangeClass
    class S4 pinkClass
```

**Example**: A parsing error extracts quantity as 10,000 instead of 1,000. Without validation:
- Risk check approves (incorrect quantity within limits)
- Execution proceeds with 10x intended size
- Customer loses significant money

**Solution**: Validation gates between every step.

## Validation Strategies

Four primary approaches to validating intermediate outputs:

### 1. Programmatic Checks

Direct code verification of specific conditions:

```python
from pydantic import BaseModel, Field, validator

class TradeOrder(BaseModel):
    symbol: str = Field(min_length=1, max_length=10)
    quantity: int = Field(gt=0, le=1000000)
    order_type: Literal["market", "limit", "stop"]
    price: Optional[float] = Field(gt=0)

    @validator('price')
    def limit_orders_need_price(cls, v, values):
        if values.get('order_type') == 'limit' and v is None:
            raise ValueError('Limit orders require a price')
        return v

def validate_parsed_order(output: str) -> TradeOrder:
    """Programmatic validation of parsed output"""
    try:
        data = json.loads(output)
        return TradeOrder(**data)
    except (json.JSONDecodeError, ValidationError) as e:
        raise ValidationError(f"Invalid order format: {e}")
```

**Best for**: Format validation, range checks, required fields

### 2. LLM-Based Validation

Use another LLM call to assess quality:

```python
validation_prompt = """
Review this risk assessment for a trade order.

Assessment: {risk_assessment}
Order details: {order}

Check for:
1. All risk factors considered (market risk, credit risk, liquidity risk)
2. Calculations appear reasonable
3. Recommendation aligns with risk level
4. No obvious contradictions

Rate quality 1-10 and explain any issues found.
"""
```

**Best for**: Subjective quality, completeness, logical consistency

### 3. Rule-Based Validation

Explicit rules for expected patterns:

```python
class TradingRules:
    MAX_ORDER_VALUE = 1_000_000
    ALLOWED_EXCHANGES = ["NYSE", "NASDAQ", "CBOE"]
    TRADING_HOURS = (9, 30, 16, 0)  # 9:30 AM - 4:00 PM

    @classmethod
    def validate(cls, order: TradeOrder) -> list[str]:
        violations = []

        # Value limit
        order_value = order.quantity * (order.price or 0)
        if order_value > cls.MAX_ORDER_VALUE:
            violations.append(f"Order value ${order_value:,.0f} exceeds limit")

        # Exchange check
        if order.exchange not in cls.ALLOWED_EXCHANGES:
            violations.append(f"Exchange {order.exchange} not allowed")

        # Trading hours
        now = datetime.now()
        if not cls._is_trading_hours(now):
            violations.append("Outside trading hours")

        return violations
```

**Best for**: Business rules, compliance requirements, operational constraints

### 4. Confidence Scoring

Use LLM-provided confidence metrics:

```python
parse_prompt = """
Parse this trade request and provide confidence score.

Request: {request}

Response format:
{
    "parsed": {...},
    "confidence": 0.0-1.0,
    "uncertain_fields": ["field names where confidence is low"]
}
"""

def validate_with_confidence(response: dict, threshold: float = 0.8):
    if response["confidence"] < threshold:
        uncertain = response.get("uncertain_fields", [])
        raise LowConfidenceError(
            f"Confidence {response['confidence']:.2f} below threshold. "
            f"Uncertain fields: {uncertain}"
        )
    return response["parsed"]
```

**Best for**: Ambiguous inputs, gauging uncertainty, deciding when to escalate

## Handling Validation Failures

When validation fails, several recovery strategies are available:

### Simple Retry

For transient issues (LLM hiccups, timeouts):

```python
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))
def call_step_with_retry(prompt: str, input_data: dict):
    return llm.complete(prompt.format(**input_data))
```

### Re-prompt with Feedback

When validation provides specific failure reasons:

```python
def step_with_correction(prompt: str, input_data: dict, max_attempts: int = 3):
    for attempt in range(max_attempts):
        output = llm.complete(prompt.format(**input_data))

        try:
            validated = validate(output)
            return validated
        except ValidationError as e:
            if attempt == max_attempts - 1:
                raise

            # Add error feedback to next attempt
            input_data["previous_attempt"] = output
            input_data["error_feedback"] = str(e)
            prompt = prompt + """

            Your previous response had issues:
            Previous attempt: {previous_attempt}
            Error: {error_feedback}

            Please correct and try again.
            """
```

### Fallback Mechanisms

When retries fail:

```python
def step_with_fallback(input_data: dict):
    try:
        return primary_agent.run(input_data)
    except Exception:
        try:
            # Try simpler backup approach
            return backup_agent.run(input_data)
        except Exception:
            # Return safe default or escalate
            return {
                "status": "MANUAL_REVIEW_REQUIRED",
                "reason": "Automated processing failed",
                "original_input": input_data
            }
```

### Self-Critique and Refinement

LLM reviews its own output:

```python
critique_prompt = """
Review this trade analysis for quality:
{analysis}

Criteria:
- All required fields present
- Numbers are reasonable for the market
- Conclusions follow from data

Identify any issues.
"""

refine_prompt = """
Original analysis: {analysis}
Critique: {critique}

Produce an improved version addressing the critique.
"""

def self_refining_step(input_data: dict):
    analysis = analyst_agent.run(input_data)
    critique = critic_agent.run(analysis)

    if "no issues" in critique.lower():
        return analysis

    refined = refiner_agent.run(analysis=analysis, critique=critique)
    return refined
```

## The Routing Pattern

While chaining handles sequential processing, **routing** handles task distribution - directing different inputs to appropriate specialized handlers.

```mermaid
flowchart TD
    I[Incoming Request] --> C[Classifier]
    C --> R{Router}
    R -->|Equity| EQ[Equity Handler]
    R -->|Options| OPT[Options Handler]
    R -->|Forex| FX[Forex Handler]
    R -->|Crypto| CR[Crypto Handler]
    EQ --> O[Output]
    OPT --> O
    FX --> O
    CR --> O

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

    class C blueClass
    class R orangeClass
```

### Two Stages: Classification + Dispatch

**Stage 1: Classification**

Determine the input's nature, category, or intent:

```python
classification_prompt = """
Classify this trading request into one category:

Request: {request}

Categories:
- EQUITY_ORDER: Stock buy/sell orders
- OPTIONS_ORDER: Options contracts
- FOREX_ORDER: Currency exchange
- CRYPTO_ORDER: Cryptocurrency trades
- ACCOUNT_QUERY: Balance, position inquiries
- COMPLIANCE_CHECK: Regulatory verification

Respond with category name only.
"""
```

**Stage 2: Task Dispatch**

Route to appropriate handler based on classification:

```python
class TradingRouter:
    def __init__(self):
        self.handlers = {
            "EQUITY_ORDER": EquityOrderHandler(),
            "OPTIONS_ORDER": OptionsOrderHandler(),
            "FOREX_ORDER": ForexOrderHandler(),
            "CRYPTO_ORDER": CryptoOrderHandler(),
            "ACCOUNT_QUERY": AccountQueryHandler(),
            "COMPLIANCE_CHECK": ComplianceHandler(),
        }
        self.classifier = ClassifierAgent()

    def route(self, request: str) -> dict:
        # Classification
        category = self.classifier.classify(request)

        # Dispatch
        handler = self.handlers.get(category)
        if handler is None:
            return self.unknown_handler(request, category)

        return handler.process(request)
```

### Classification Methods

**Rule-Based Classification**

Fast and predictable for clear patterns:

```python
def rule_based_classify(request: str) -> str:
    request_lower = request.lower()

    # Keyword matching
    if any(word in request_lower for word in ["option", "call", "put", "strike"]):
        return "OPTIONS_ORDER"
    if any(word in request_lower for word in ["btc", "eth", "crypto", "bitcoin"]):
        return "CRYPTO_ORDER"
    if any(word in request_lower for word in ["eur", "usd", "forex", "fx"]):
        return "FOREX_ORDER"
    if any(word in request_lower for word in ["buy", "sell", "stock", "shares"]):
        return "EQUITY_ORDER"
    if any(word in request_lower for word in ["balance", "position", "holdings"]):
        return "ACCOUNT_QUERY"

    return "UNKNOWN"
```

**LLM-Based Classification**

Better for nuanced or ambiguous requests:

```python
llm_classification_prompt = """
Analyze this trading request and classify it.

Request: "{request}"

Consider:
- What action is being requested?
- What asset type is involved?
- Is this a trade or an inquiry?

Return JSON:
{
    "category": "CATEGORY_NAME",
    "confidence": 0.0-1.0,
    "reasoning": "brief explanation"
}
"""
```

**Hybrid Classification**

Best of both worlds:

```python
def hybrid_classify(request: str) -> str:
    # Try rule-based first (fast, cheap)
    category = rule_based_classify(request)

    if category != "UNKNOWN":
        return category

    # Fall back to LLM for unclear cases
    result = llm_classify(request)

    if result["confidence"] < 0.7:
        return "MANUAL_REVIEW"

    return result["category"]
```

## Applied Example: Trade Order Pipeline

Let's build a complete pipeline combining chaining and routing:

```mermaid
flowchart TD
    REQ[Trade Request] --> PARSE[1. Parse Request]
    PARSE --> CLASSIFY[2. Classify Order Type]
    CLASSIFY --> ROUTE{Route by Type}

    ROUTE -->|Equity| EQ_CHAIN[Equity Chain]
    ROUTE -->|Options| OPT_CHAIN[Options Chain]
    ROUTE -->|Complex| HUMAN[Human Review]

    subgraph EQ_CHAIN[Equity Processing Chain]
        E1[Validate Order] --> E2[Check Risk]
        E2 --> E3[Execute]
        E3 --> E4[Confirm]
    end

    subgraph OPT_CHAIN[Options Processing Chain]
        O1[Validate Greeks] --> O2[Check Margin]
        O2 --> O3[Execute]
        O3 --> O4[Confirm]
    end

    EQ_CHAIN --> OUT[Output]
    OPT_CHAIN --> OUT

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

    class ROUTE orangeClass
```

### Implementation

```python
class TradePipeline:
    def __init__(self):
        self.parser = OrderParser()
        self.classifier = OrderClassifier()
        self.equity_chain = EquityOrderChain()
        self.options_chain = OptionsOrderChain()

    def process(self, request: str) -> dict:
        # Step 1: Parse with validation
        parsed = self.parser.parse(request)
        if not self.parser.validate(parsed):
            return {"status": "PARSE_ERROR", "details": parsed.errors}

        # Step 2: Classify
        order_type = self.classifier.classify(parsed)

        # Step 3: Route to appropriate chain
        if order_type == "EQUITY":
            return self.equity_chain.process(parsed)
        elif order_type == "OPTIONS":
            return self.options_chain.process(parsed)
        else:
            return {"status": "MANUAL_REVIEW", "order": parsed}


class EquityOrderChain:
    def process(self, order: dict) -> dict:
        # Chain step 1: Validate
        validated = self.validate_step(order)
        if not validated["is_valid"]:
            return {"status": "VALIDATION_FAILED", **validated}

        # Chain step 2: Risk check
        risk_result = self.risk_step(validated)
        if risk_result["risk_level"] == "HIGH":
            return {"status": "RISK_REJECTED", **risk_result}

        # Chain step 3: Execute
        execution = self.execute_step(risk_result)
        if execution["status"] != "FILLED":
            return {"status": "EXECUTION_FAILED", **execution}

        # Chain step 4: Confirm
        confirmation = self.confirm_step(execution)

        return {
            "status": "SUCCESS",
            "confirmation": confirmation,
            "audit_trail": {
                "validation": validated,
                "risk": risk_result,
                "execution": execution
            }
        }
```

## Context Management in Chains

As chains grow, context management becomes critical. LLMs have finite context windows, and overloading them degrades performance.

### Selective Context Passing

Only pass what the next step needs:

```python
def build_context_for_step(step_name: str, full_context: dict) -> dict:
    """Extract only relevant context for each step"""

    context_requirements = {
        "validate": ["parsed_order", "market_rules"],
        "risk_check": ["validated_order", "position_limits", "account_balance"],
        "execute": ["approved_order", "exchange_connection"],
        "confirm": ["execution_result", "customer_email"]
    }

    required = context_requirements.get(step_name, [])
    return {k: full_context[k] for k in required if k in full_context}
```

### Contextual Reiteration

Restate critical information in each prompt to prevent "forgetting":

```python
execution_prompt = """
Execute this trade order.

CRITICAL CONTEXT (do not modify):
- Account: {account_id}
- Risk approval: {risk_approval_id}
- Maximum slippage: {max_slippage}%

Order to execute:
{approved_order}

Generate execution instruction.
"""
```

### Context Summarization

For long chains, summarize earlier steps:

```python
def summarize_previous_steps(chain_history: list) -> str:
    summary_prompt = """
    Summarize these processing steps in 2-3 sentences,
    preserving key decisions and values:

    {steps}
    """
    return llm.complete(summary_prompt.format(steps=chain_history))
```

## Takeaways

1. **Prompt chaining** breaks complex tasks into focused steps, enabling validation and error recovery at each stage

2. **Validation gates** between steps prevent error propagation - use programmatic checks, LLM validation, rules, or confidence scoring as appropriate

3. **Recovery strategies** include retry, re-prompt with feedback, fallback mechanisms, and self-critique refinement

4. **Routing** combines classification and dispatch to direct diverse inputs to specialized handlers

5. **Hybrid classification** uses fast rule-based methods first, falling back to LLM for ambiguous cases

6. **Context management** requires selective passing, critical reiteration, and summarization to work within LLM constraints

---

*This is the fifth post in my Applied Agentic AI for Finance series. Next: [Parallel Processing and Quality Control in Finance](https://vanducng.dev/2026/01/06/Parallel-Processing-and-Quality-Control-in-Finance/) where we'll explore concurrent execution and evaluator-optimizer patterns.*