RAG Application

The following section describes how Arch can help you build faster, smarter and more accurate Retrieval-Augmented Generation (RAG) applications.

Intent-drift Detection

Developers struggle to handle follow-up or clarifying questions. Specifically, when users ask for changes or additions to previous responses their AI applications often generate entirely new responses instead of adjusting previous ones. Arch offers intent-drift tracking as a feature so that developers can know when the user has shifted away from a previous intent so that they can dramatically improve retrieval accuracy, lower overall token cost and improve the speed of their responses back to users.

Arch uses its built-in lightweight NLI and embedding models to know if the user has steered away from an active intent. Arch’s intent-drift detection mechanism is based on its’ prompt_targets primtive. Arch tries to match an incoming prompt to one of the prompt_targets configured in the gateway. Once it detects that the user has moved away from an active active intent, Arch adds the x-arch-intent-drift headers to the request before sending it your application servers.

Intent Detection Example

@app.route('/process_rag', methods=['POST'])
def process_rag():
    # Extract JSON data from the request
    data = request.get_json()

    user_id = data.get('user_id')
    if not user_id:
        return jsonify({'error': 'User ID is required'}), 400

    client_messages = data.get('messages')
    if not client_messages or not isinstance(client_messages, list):
        return jsonify({'error': 'Messages array is required'}), 400

    # Extract the intent change marker from Arch's headers if present for the current prompt
    intent_changed_header = request.headers.get('x-arch-intent-marker', '').lower()
    if intent_changed_header in ['', 'false']:
        intent_changed = False
    elif intent_changed_header == 'true':
        intent_changed = True
    else:
        # Invalid value provided
        return jsonify({'error': 'Invalid value for x-arch-prompt-intent-change header'}), 400

    # Update user conversation based on intent change
    memory = update_user_conversation(user_id, client_messages, intent_changed)

    # Retrieve messages since last intent change for LLM
    messages_for_llm = get_messages_since_last_intent(memory.chat_memory.messages)

    # Forward messages to upstream LLM
    llm_response = forward_to_llm(messages_for_llm)

Note

Arch is (mostly) stateless so that it can scale in an embarrassingly parrallel fashion. So, while Arch offers intent-drift detetction, you still have to maintain converational state with intent drift as meta-data. The following code snippets show how easily you can build and enrich conversational history with Langchain (in python), so that you can use the most relevant prompts for your retrieval and for prompting upstream LLMs.

Step 1: Define ConversationBufferMemory

from flask import Flask, request, jsonify
from datetime import datetime
import uuid
from langchain.memory import ConversationBufferMemory
from langchain.schema import AIMessage, HumanMessage
from langchain import OpenAI

app = Flask(__name__)

# Global dictionary to keep track of user memories
user_memories = {}

def get_user_conversation(user_id):
    """
    Retrieve the user's conversation memory using LangChain.
    If the user does not exist, initialize their conversation memory.
    """
    if user_id not in user_memories:
        user_memories[user_id] = ConversationBufferMemory(return_messages=True)
    return user_memories[user_id]

Step 2: Update ConversationBufferMemory w/ intent

def update_user_conversation(user_id, client_messages, intent_changed):
    """
    Update the user's conversation memory with new messages using LangChain.
    Each message is augmented with a UUID, timestamp, and intent change marker.
    Only new messages are added to avoid duplication.
    """
    memory = get_user_conversation(user_id)
    stored_messages = memory.chat_memory.messages

    # Determine the number of stored messages
    num_stored_messages = len(stored_messages)
    new_messages = client_messages[num_stored_messages:]

    # Process each new message
    for index, message in enumerate(new_messages):
        role = message.get('role')
        content = message.get('content')
        metadata = {
            'uuid': str(uuid.uuid4()),
            'timestamp': datetime.utcnow().isoformat(),
            'intent_changed': False  # Default value
        }

        # Mark the intent change on the last message if detected
        if intent_changed and index == len(new_messages) - 1:
            metadata['intent_changed'] = True

        # Create a new message with metadata
        if role == 'user':
            memory.chat_memory.add_message(
                HumanMessage(content=content, additional_kwargs={'metadata': metadata})
            )
        elif role == 'assistant':
            memory.chat_memory.add_message(
                AIMessage(content=content, additional_kwargs={'metadata': metadata})
            )
        else:
            # Handle other roles if necessary
            pass

    return memory

Step 3: Get Messages based on latest drift

def get_messages_since_last_intent(messages):
    """
    Retrieve messages from the last intent change onwards using LangChain.
    """
    messages_since_intent = []
    for message in reversed(messages):
        # Insert message at the beginning to maintain correct order
        messages_since_intent.insert(0, message)
        metadata = message.additional_kwargs.get('metadata', {})
        # Break if intent_changed is True
        if metadata.get('intent_changed', False) == True:
            break
    return messages_since_intent

You can used the last set of messages that match to an intent to prompt an LLM, use it with an vector-DB for improved retrieval, etc. With Arch and a few lines of code, you can improve the retrieval accuracy, lower overall token cost and dramatically improve the speed of their responses back to users.

Parameter Extraction for RAG

To build RAG (Retrieval-Augmented Generation) applications, you can configure prompt targets with parameters, enabling Arch to retrieve critical information in a structured way for processing. This approach improves the retrieval quality and speed of your application. By extracting parameters from the conversation, you can pull the appropriate chunks from a vector database or SQL-like data store to enhance accuracy. With Arch, you can streamline data retrieval and processing to build more efficient and precise RAG applications.

Step 1: Define prompt targets with parameter definitions

Prompt Targets

prompt_targets:
  - name: get_device_statistics
    description: >
      This prompt target ensures that when users request device-related statistics, the system accurately retrieves and presents the relevant data
      based on the specified devices and time range. Examples of user prompts, include:

      - "Show me the performance stats for device 12345 over the past week."
      - "What are the error rates for my devices in the last 24 hours?"
      - "I need statistics on device 789 over the last 10 days."

    path: /agent/device_summary
    parameters:
      - name: "device_ids"
        type: list  # Options: integer | float | list | dictionary | set
        description: "A list of device identifiers (IDs) for which the statistics are requested."
        required: true
      - name: "time_range"
        type: integer  # Options: integer | float | list | dictionary | set
        description: "The number of days in the past over which to retrieve device statistics. Defaults to 7 days if not specified."
        required: false
        default: 7

Step 2: Process request parameters in Flask

Once the prompt targets are configured as above, handling those parameters is

Parameter handling with Flask

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/agent/device_summary', methods=['POST'])
def get_device_summary():
    """
    Endpoint to retrieve device statistics based on device IDs and an optional time range.
    """
    data = request.get_json()

    # Validate 'device_ids' parameter
    device_ids = data.get('device_ids')
    if not device_ids or not isinstance(device_ids, list):
        return jsonify({'error': "'device_ids' parameter is required and must be a list"}), 400

    # Validate 'time_range' parameter (optional, defaults to 7)
    time_range = data.get('time_range', 7)
    if not isinstance(time_range, int):
        return jsonify({'error': "'time_range' must be an integer"}), 400

    # Simulate retrieving statistics for the given device IDs and time range
    # In a real application, you would query your database or external service here
    statistics = []
    for device_id in device_ids:
        # Placeholder for actual data retrieval
        stats = {
            'device_id': device_id,
            'time_range': f'Last {time_range} days',
            'data': f'Statistics data for device {device_id} over the last {time_range} days.'
        }
        statistics.append(stats)

    response = {
        'statistics': statistics
    }

    return jsonify(response), 200

if __name__ == '__main__':
    app.run(debug=True)