AI Assistant Architecture

The XRP Transaction Risk AI platform uses three specialized OpenAI assistants, each configured with specific instructions and access to shared vector storage containing crawled business information.

Three-Assistant System

Summary Assistant

Generates concise regulatory overviews

Report Assistant

Identifies compliance red flags

Resource Assistant

Discovers relevant regulatory documents

Configuration and Initialization

All assistants are initialized with OpenAI API credentials and vector storage access:

# OpenAI configuration
client = OpenAI(api_key=st.secrets["OPENAI_API_KEY"])
vector_storage_id = st.secrets["VECTOR_STORAGE_ID"]
report_assistant_id = st.secrets["ASSISTANT_ID"]
summary_assistant_id = st.secrets["SUMMARY_ASSISTANT"]
resource_assistant_id = st.secrets["RESOURCE_ASSISTANT"]

Source location: ripple_challange.py:73-77

Each assistant ID is stored in Streamlit secrets for security and easy rotation. These IDs are obtained from the OpenAI Assistants API after creating and configuring each assistant.

Assistant Orchestration

The assistants are organized in a dictionary structure for systematic execution:

assistants_ = {
    'resource': resource_assistant_id,
    'report': report_assistant_id,
    'summary': summary_assistant_id,
}
prompts_ = {
    'resource': resource_prompt,
    'report': report_prompt,
    'summary': summary_prompt,
}

Why This Architecture?

Benefits of Specialized Assistants:

Focused Expertise: Each assistant is trained for a specific task
Parallel Processing: Can be run concurrently for faster results
Independent Scaling: Adjust resources per assistant based on demand
Version Control: Update one assistant without affecting others
Cost Optimization: Use different models/parameters per task
Maintainability: Easier to debug and improve specific functions

Core Execution Function

All three assistants use the same execution pattern with streaming responses:

def run_assistant(prompt, assistant_id):
    thread = client.beta.threads.create(
        messages=[
            {
                "role": "user",
                "content": [{"type": "text", "text": prompt}],
            }
        ]
    )
    run = client.beta.threads.runs.create(
        thread_id=thread.id, assistant_id=assistant_id, stream=True
    )

    result_text = ""
    for event in run:
        if isinstance(event, ThreadMessageCompleted):
            result_text = event.data.content[0].text.value
        if isinstance(event, ThreadRunFailed):
            print(event)
            break
    return result_text

Source location: ripple_challange.py:153-173

Execution Flow Breakdown

Create Thread

Initialize a new conversation thread with the user prompt:

thread = client.beta.threads.create(
    messages=[{
        "role": "user",
        "content": [{"type": "text", "text": prompt}]
    }]
)

Each thread is isolated and maintains its own conversation context.

Start Streaming Run

Execute the assistant with streaming enabled:

run = client.beta.threads.runs.create(
    thread_id=thread.id, 
    assistant_id=assistant_id, 
    stream=True
)

Streaming provides real-time results and better user experience compared to waiting for the complete response.

Process Events

Listen for completion or failure events:

result_text = ""
for event in run:
    if isinstance(event, ThreadMessageCompleted):
        result_text = event.data.content[0].text.value
    if isinstance(event, ThreadRunFailed):
        print(event)
        break

Event Types:

ThreadMessageCompleted: Successfully generated response
ThreadRunFailed: Error occurred during execution

Return Results

return result_text

Summary Assistant

Provides high-level regulatory compliance overviews:

Purpose
Configuration
Use Cases

Objective: Generate executive-level summaries of regulatory requirementsPrompt Template:

summary_prompt = f"Provide a brief summary of the financial regulations relevant to the company: {company_name}"

Expected Output:

2-3 paragraph overview
Key regulations mentioned by name
Industry classification
Overall compliance posture

Report Assistant

Performs deep analysis to identify specific compliance risks:

Purpose
Configuration
Detection Areas

Objective: Identify concrete red flags and compliance gapsPrompt Template:

report_prompt = f"Identify any financial compliance red flags in the company data: {company_name} that might affect their business compliance."

Expected Output:

Specific red flags with evidence
Risk severity levels
Citations from crawled content
Recommended actions

Resource Assistant

Discoverers relevant regulatory documents and references:

Purpose
Configuration
Document Types

Objective: Compile list of applicable regulations and documentsPrompt Template:

resource_prompt = f"List the relevant financial regulatory documents for the company: {company_name}"

Expected Output:

Regulation names and numbers
Issuing authorities
Document URLs (when available)
Jurisdictional scope

Execution Loop

All three assistants are executed systematically:

# Start AI assistants
results = {}
for key in assistants_.keys():
    results[key] = run_assistant(
        prompt=prompts_[key],
        assistant_id=assistants_[key]
    )

Source location: ripple_challange.py:176-181

Optimization Opportunity

Current Implementation: Sequential executionPerformance Enhancement:

import concurrent.futures

def run_assistants_parallel():
    results = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        future_to_key = {
            executor.submit(run_assistant, prompts_[key], assistants_[key]): key
            for key in assistants_.keys()
        }
        for future in concurrent.futures.as_completed(future_to_key):
            key = future_to_key[future]
            results[key] = future.result()
    return results

This could reduce total execution time by ~3x.

Vector Storage Integration

All assistants share access to the same vector storage containing crawled website content:

# From crawl_util.py
vector_store_file = self.client.beta.vector_stores.files.create(
    vector_store_id=self.vector_storage_id, file_id=file_.id
)

How It Works
Query Process
Caching Strategy

Vector Storage Flow:

Website content is crawled and saved as text file
File is uploaded to OpenAI with purpose=“assistants”
File is added to shared vector store
All three assistants query this vector store
Semantic search returns relevant passages

# Check Redis cache before crawling
if file_id := self.r.get(url):
    self.r.zadd("vs_files", {file_id: int(time.time())})
    return

Benefits:

Avoid re-crawling recently analyzed sites
Reduce OpenAI API costs
Faster response times
Track file usage patterns

Event Handling

The streaming implementation handles multiple event types:

Stream Event Types

from openai.types.beta.assistant_stream_event import (
    ThreadMessageCompleted, 
    ThreadRunFailed
)

for event in run:
    if isinstance(event, ThreadMessageCompleted):
        result_text = event.data.content[0].text.value
    if isinstance(event, ThreadRunFailed):
        print(event)
        break

Event Categories:

ThreadCreated: Thread initialized
ThreadRunCreated: Run started
ThreadRunQueued: Waiting in queue
ThreadRunInProgress: Currently processing
ThreadMessageCreated: Response generation started
ThreadMessageInProgress: Response being generated
ThreadMessageCompleted: ✅ Success - response ready
ThreadRunFailed: ❌ Error occurred
ThreadRunCompleted: Run finished successfully

Error Handling Best Practices

Current Implementation
Enhanced Error Handling
Retry Logic

if isinstance(event, ThreadRunFailed):
    print(event)
    break
return result_text  # May be empty string on failure

def run_assistant_robust(prompt, assistant_id):
    try:
        thread = client.beta.threads.create(
            messages=[{"role": "user", "content": [{"type": "text", "text": prompt}]}]
        )
        run = client.beta.threads.runs.create(
            thread_id=thread.id, assistant_id=assistant_id, stream=True
        )
        
        result_text = ""
        for event in run:
            if isinstance(event, ThreadMessageCompleted):
                result_text = event.data.content[0].text.value
            if isinstance(event, ThreadRunFailed):
                error_msg = event.data.last_error.message if event.data.last_error else "Unknown error"
                raise Exception(f"Assistant run failed: {error_msg}")
        
        if not result_text:
            raise Exception("No response generated")
        
        return result_text
        
    except Exception as e:
        st.error(f"Assistant error: {str(e)}")
        return f"Error: Unable to generate {assistant_id} analysis"

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def run_assistant_with_retry(prompt, assistant_id):
    return run_assistant(prompt, assistant_id)

Assistant Configuration Best Practices

When Creating Assistants in OpenAI Console:

Name clearly: “XRP Risk - Summary Assistant”
Detailed instructions: Provide context about the domain
Enable file search: Required for vector storage access
Link vector store: Attach the shared storage ID
Set temperature: Lower for factual, higher for creative
Save assistant ID: Store in secrets management

Performance Metrics

Average Execution

5-15 seconds per assistant

Token Usage

500-2000 tokens per response

Vector Queries

3-10 chunks retrieved per prompt

Next Steps

Risk Assessment

See how assistants fit into the overall workflow

XRP Transactions

Learn about transaction execution after analysis

Get Started

Core Features

Guides

AI Assistant Architecture

AI Assistant Architecture

Three-Assistant System

Summary Assistant

Report Assistant

Resource Assistant

Configuration and Initialization

Assistant Orchestration

Core Execution Function

Execution Flow Breakdown

Summary Assistant

Report Assistant

Resource Assistant

Execution Loop

Vector Storage Integration

Event Handling

Error Handling Best Practices

Assistant Configuration Best Practices

Performance Metrics

Average Execution

Token Usage

Vector Queries

Next Steps

Risk Assessment

XRP Transactions

Build docs developers (and LLMs) love

Get Started

Core Features

Guides

​AI Assistant Architecture

​Three-Assistant System

Summary Assistant

Report Assistant

Resource Assistant

​Configuration and Initialization

​Assistant Orchestration

​Core Execution Function

​Execution Flow Breakdown

​Summary Assistant

​Report Assistant

​Resource Assistant

​Execution Loop

​Vector Storage Integration

​Event Handling

​Error Handling Best Practices

​Assistant Configuration Best Practices

​Performance Metrics

Average Execution

Token Usage

Vector Queries

​Next Steps

Risk Assessment

XRP Transactions

Build docs developers (and LLMs) love

AI Assistant Architecture

Three-Assistant System

Configuration and Initialization

Assistant Orchestration

Core Execution Function

Execution Flow Breakdown

Summary Assistant

Report Assistant

Resource Assistant

Execution Loop

Vector Storage Integration

Event Handling

Error Handling Best Practices

Assistant Configuration Best Practices

Performance Metrics

Next Steps