relevanceapi.py

"""
These are the Relevance API endpoints.
Before running this script, ensure you have the following environment variables
set as ENVIRONMENT variables or in your .env file:

RELEVANCEAI_BASE_URL="https://api-{region}.stack.tryrelevance.com"
RELEVANCEAI_API_KEY="Your_Api_Key"
RELEVANCEAI_REGION_ID="Your_Region_Id"

You can install the required packages with the following pip command:
pip install requests python-dotenv

"""

import os
import requests
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Environment variables
RELEVANCEAI_REGION_ID = os.getenv("RELEVANCEAI_REGION_ID")
RELEVANCEAI_BASE_URL = os.getenv("RELEVANCEAI_BASE_URL")
RELEVANCEAI_API_KEY = os.getenv("RELEVANCEAI_API_KEY")
RELEVANCEAI_API_TIMEOUT = int(os.getenv( "RELEVANCEAI_API_TIMEOUT", "30" ))

# replace {RELEVANCEAI_REGION_ID} with the actual region ID
RELEVANCEAI_BASE_URL = RELEVANCEAI_BASE_URL.replace("{region}", RELEVANCEAI_REGION_ID)


def get_projects(project_id: str):
    """
    Get info on a specific project. But only seems to work on default project?
    """
    endpoint = f"{RELEVANCEAI_BASE_URL}/latest/projects/list"
    authorization_token = f"{project_id}:{RELEVANCEAI_API_KEY}"
    headers = {"Authorization": authorization_token}
    response = requests.get(endpoint, headers=headers, json=None, timeout=RELEVANCEAI_API_TIMEOUT)
    response.raise_for_status()
    return response.json()



def get_agents(project_id: str):
    """
    List all available agents in the project.
    """
    endpoint = f"{RELEVANCEAI_BASE_URL}/latest/agents/list"
    authorization_token = f"{project_id}:{RELEVANCEAI_API_KEY}"
    headers = {"Authorization": authorization_token}
    response = requests.post(endpoint, headers=headers, json=None, timeout=RELEVANCEAI_API_TIMEOUT)
    response.raise_for_status()
    return response.json()


def get_agent_conversations(project_id: str, params):
    """
    List all conversations for a given project / agent. Must pass agent_id in params filter.
    params: {
        "filters": [
            {"filter_type": "exact_match", "field": "type", "condition_value": "conversation"},
            {"filter_type": "exact_match", "field": "conversation.agent_id", "condition_value": AGENT_ID}
        ],
        "include_hidden": True
    }
    """
    endpoint = f"{RELEVANCEAI_BASE_URL}/latest/knowledge/sets/list"
    authorization_token = f"{project_id}:{RELEVANCEAI_API_KEY}"
    headers = {"Authorization": authorization_token}
    response = requests.post(endpoint, headers=headers, json=params, timeout=RELEVANCEAI_API_TIMEOUT)
    response.raise_for_status()
    return response.json()






def get_agent_conversation(project_id: str, conversation_id, params):
    """
    List all conversations for the project/agent. Must pass agent_id in params filter.
    params: {
            "filters": [
                {"filter_type": "exact_match", "field": "type", "condition_value": "conversation"},
                {"filter_type": "exact_match", "field": "conversation.agent_id", "condition_value": AGENT_ID}
            ],
            "include_hidden": True
        }
    """
    endpoint = f"{RELEVANCEAI_BASE_URL}/latest/knowledge/list"
    authorization_token = f"{project_id}:{RELEVANCEAI_API_KEY}"
    headers = {"Authorization": authorization_token}
    payload = {
        "knowledge_set": conversation_id,
        "filters": params
    }
    response = requests.post(endpoint, headers=headers, json=payload, timeout=RELEVANCEAI_API_TIMEOUT)
    response.raise_for_status()
    return response.json()






def agent_conversation_create(project_id: str, agent_id, message):
    """
    Trigger a new conversation (task) with an agent.
    """
    endpoint = f"{RELEVANCEAI_BASE_URL}/latest/agents/trigger"
    authorization_token = f"{project_id}:{RELEVANCEAI_API_KEY}"
    headers = {
        "Authorization": authorization_token,
        "Content-Type": "application/json"
    }
    payload = {
        "message": {
            "role": "user",
            "content": message
        },
        "agent_id": agent_id
    }
    response = requests.post(endpoint, headers=headers, json=payload, timeout=RELEVANCEAI_API_TIMEOUT)
    response.raise_for_status()
    return response.json()





def agent_conversation_add_message(project_id: str, agent_id, conversation_id, message):
    """
    Add a message to an existing conversation (task) with an agent.
    """
    endpoint = f"{RELEVANCEAI_BASE_URL}/latest/agents/trigger"
    authorization_token = f"{project_id}:{RELEVANCEAI_API_KEY}"
    headers = {
        "Authorization": authorization_token,
        "Content-Type": "application/json"
    }
    payload = {
        "message": {
            "role": "user",
            "content": message
        },
        "conversation_id": conversation_id,
        "agent_id": agent_id
    }
    response = requests.post(endpoint, headers=headers, json=payload, timeout=RELEVANCEAI_API_TIMEOUT)
    response.raise_for_status()
    return response.json()








def agent_conversation_stream_messages(project_id: str, studio_id, job_id, params):
    """
    Lists conversations using the async_poll endpoint.

    :param studio_id: Studio ID for the agent.
    :param job_id: Job ID for the specific task.
    :param params = {
                "ending_update_only": "true"  # Example parameter, adjust as needed
            }
    """
    endpoint = f"{RELEVANCEAI_BASE_URL}/latest/studios/{studio_id}/async_poll/{job_id}"
    authorization_token = f"{project_id}:{RELEVANCEAI_API_KEY}"
    headers = {
        "Authorization": authorization_token,
        "Content-Type": "application/json"
    }

    # Send the GET request
    response = requests.get(endpoint, headers=headers, params=params, timeout=RELEVANCEAI_API_TIMEOUT)
    response.raise_for_status()

    if response.status_code == 200:
        return response.json()
    else:
        print(f"Request failed with status code: {response.status_code}")
        return response.text