Get Started

Get an API Key

Contact patrick@rippletide.com or yann@rippletide.com to get an API key and create your agent.

Step 1: Create an Agent

First, create an agent with a name and an intial prompt describing the goal of the Agent:

import os
import uuid
import requests
from enum import Enum

RIPPLETIDE_API_KEY = os.environ["RIPPLETIDE_API_KEY"]
BASE_URL = "https://agent.rippletide.com/api/sdk"
headers = {"x-api-key": RIPPLETIDE_API_KEY, "Content-Type": "application/json"}

def create_agent():
    url = f"{BASE_URL}/agent"
    data = {"name": "example-agent", "prompt": "Interact with the user to understand which product he might be interested in and suggest him the best offer. Once done, complete his order."}

    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()

# Create the agent
agent = create_agent()
agent_id = agent["id"]

Step 2: Define Q&A Knowledge and Tags

Put all the questions that the Agent needs to know to perform the use-case. Describe it in the form of questions and answers.

Example: Q - What products can be ordered? A - You can order our last Samsung TV RED43

Then, we will tag the answers. A tag comes with a description of the glossary term and is used to annotate the Q&A.

from enum import Enum

class Tags(Enum):
    HIGH_TECH_PRODUCTS = ("high_tech_products", "all products in high-tech category")
    DELIVERY_DATE = ("delivery_date", "the date the customer will receive his product")
    PURCHASE_HISTORY = ("purchase_history", "the list of all product bought")
    DISCOUNT = ("discount", "all types of discount that can apply on products")

small_q_and_a = [
    {
        "question": "What is the price?",
        "answer": "The Samsung TV RED43 costs $990",
        "tags": [Tags.HIGH_TECH_PRODUCTS.value[0], Tags.DISCOUNT.value[0]],
    },
]
# Fetch already existing tags
response = requests.get(
    f"{BASE_URL}/tag",
    headers=headers,
)
if response.status_code == 200 and response.json():
    tag_ids = {t['name']: t['id'] for t in response.json()["tags"]}

for tag in Tags:
    tag_name, tag_description = tag.value

    if tag_name in tag_ids.keys():
        continue

    tag_data = {"name": tag_name, "description": tag_description}

    # create tag
    response = requests.post(
        f"{BASE_URL}/tag",
        headers=headers,
        json=tag_data
    )
    response.raise_for_status()
    tag_ids[tag_name] = response.json()["id"]

# Create Q&As and link them to tags
for faq in small_q_and_a:
    faq_data = {
        "question": faq["question"],
        "answer": faq["answer"],
        "agent_id": agent_id
    }

    response = requests.post(
        f"{BASE_URL}/q-and-a",
        headers=headers,
        json=faq_data
    )
    response.raise_for_status()
    faq_id = response.json()["id"]

    # Link FAQ to tags
    for tag_name in faq["tags"]:
        if tag_name in tag_ids:
            faq_tag_data = {"q_and_a_id": faq_id, "tag_id": tag_ids[tag_name]}
            requests.post(
                f"{BASE_URL}/q-and-a-tag",
                headers=headers,
                json=faq_tag_data
            )

Step 4: Define Possible Actions, States and Requirements

Define the actions that the Agent can perform. An action can be to call an external tool.

A state is describing the current situation of the Agent. For example: the user described his needs or a proposal has been sent. You can define predicates (requirements) that need to be satisfied before doing an action.

some_actions = [
    {
        "name": "add_product_to_cart",
        "description": "Add a product to the cart",
        "what_to_do": "Add the selected product to the user's cart"
    },
    {
        "name": "checkout",
        "description": "Check out the cart",
        "what_to_do": "Process the checkout for the products in the cart"
    },
    {
        "name": "open_support_ticket",
        "description": "Open a support ticket when the agent cannot answer",
        "what_to_do": "Collect user contact information and open a support ticket"
    },
    {
        "name": "book_meeting",
        "description": "Book a meeting with a sales person",
        "what_to_do": "Schedule a meeting based on sales person availability"
    }
]

# Set up actions
for action in some_actions:
    action_data = {"agent_id": agent_id, **action}
    response = requests.post(
        f"{BASE_URL}/action",
        headers=headers,
        json=action_data
    )
    response.raise_for_status()

predicate_state = {
    "transition_kind": "branch",
    "question_to_evaluate": "The user described his needs",
    "possible_values": ["recommend_product", "describe_discount_mechanism"],
    "re_evaluate": True,
    "value_to_node": {
        "recommend_product": {
            "transition_kind": "go_to_next",
            "question_to_evaluate": "A product has been chosen",
            "next_node": {
                "transition_kind": "branch",
                "question_to_evaluate": "What would you like to do next?",
                "possible_values": ["ask_delivery_date", "checkout", "add_discount", "delete_product"],
                "value_to_node": {
                    "ask_delivery_date": {
                        "transition_kind": "end",
                        "question_to_evaluate": "When would you like your product delivered?"
                    },
                    "checkout": {
                        "transition_kind": "end",
                        "question_to_evaluate": "Proceeding to checkout."
                    },
                    "add_discount": {
                        "transition_kind": "end",
                        "question_to_evaluate": "Applying available discounts."
                    },
                    "delete_product": {
                        "transition_kind": "end",
                        "question_to_evaluate": "Product removed from cart."
                    }
                }
            }
        },
        "describe_discount_mechanism": {
            "transition_kind": "end",
            "question_to_evaluate": "Let me explain how discounts work."
        }
    }
}

# Set up state predicate
response = requests.put(
    f"{BASE_URL}/state-predicate/{agent_id}",
    headers=headers,
    json={"state_predicate": predicate_state}
)
response.raise_for_status()

Complete Example

import os
import uuid
import requests
from enum import Enum

RIPPLETIDE_API_KEY = os.environ["RIPPLETIDE_API_KEY"]  # This API key is given to developer by Rippletide

BASE_URL = "https://agent.rippletide.com/api/sdk"
headers = {"x-api-key": RIPPLETIDE_API_KEY, "Content-Type": "application/json"}


class Tags(Enum):
    HIGH_TECH_PRODUCTS = ("high_tech_products", "all products in high-tech category")
    DELIVERY_DATE = ("delivery_date", "the date the customer will receive his product")
    PURCHASE_HISTORY = ("purchase_history", "the list of all product bought")
    DISCOUNT = ("discount", "all types of discount that can apply on products")


small_q_and_a = [
    {
        "question": "What is the price?",
        "answer": "The Samsung TV RED43 costs $990",
        "tags": [Tags.HIGH_TECH_PRODUCTS.value[0], Tags.DISCOUNT.value[0]],
    },
]

some_guardrails = [
    {"type": "action", "instruction": "Always stay professional"},
    {"type": "action", "instruction": "Never talk about pricing"},
]

some_actions = [
    {
        "name": "add_product_to_cart",
        "description": "Add a product to the cart",
        "what_to_do": "Add the selected product to the user's cart"
    },
    {
        "name": "checkout",
        "description": "Check out the cart",
        "what_to_do": "Process the checkout for the products in the cart"
    },
    {
        "name": "open_support_ticket",
        "description": "Open a support ticket when the agent cannot answer",
        "what_to_do": "Collect user contact information and open a support ticket"
    },
    {
        "name": "book_meeting",
        "description": "Book a meeting with a sales person",
        "what_to_do": "Schedule a meeting based on sales person availability"
    }
]
predicate_state = {
    "transition_kind": "branch",
    "question_to_evaluate": "The user described his needs",
    "possible_values": ["recommend_product", "describe_discount_mechanism"],
    "re_evaluate": True,
    "value_to_node": {
        "recommend_product": {
            "transition_kind": "go_to_next",
            "question_to_evaluate": "A product has been chosen",
            "next_node": {
                "transition_kind": "branch",
                "question_to_evaluate": "What would you like to do next?",
                "possible_values": ["ask_delivery_date", "checkout", "add_discount", "delete_product"],
                "value_to_node": {
                    "ask_delivery_date": {
                        "transition_kind": "end",
                        "question_to_evaluate": "When would you like your product delivered?"
                    },
                    "checkout": {
                        "transition_kind": "end",
                        "question_to_evaluate": "Proceeding to checkout."
                    },
                    "add_discount": {
                        "transition_kind": "end",
                        "question_to_evaluate": "Applying available discounts."
                    },
                    "delete_product": {
                        "transition_kind": "end",
                        "question_to_evaluate": "Product removed from cart."
                    }
                }
            }
        },
        "describe_discount_mechanism": {
            "transition_kind": "end",
            "question_to_evaluate": "Let me explain how discounts work."
        }
    }
}



def create_agent():
    url = f"{BASE_URL}/agent"
    data = {"name": "example-agent", "prompt": "Interact with the user to understand which product he might be interested in and suggest him the best offer. Once done, complete his order."}

    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return response.json()


def setup_agent_knowledge(agent_id):
    # Create tags
    tag_ids = {}

    # Fetch already existing tags
    response = requests.get(
        f"{BASE_URL}/tag",
        headers=headers,
    )
    if response.status_code == 200 and response.json():
        tag_ids = {t['name']: t['id'] for t in response.json()["tags"]}

    for tag in Tags:
        tag_name, tag_description = tag.value

        if tag_name in tag_ids.keys():
            continue

        tag_data = {"name": tag_name, "description": tag_description}

        # create tag
        response = requests.post(
            f"{BASE_URL}/tag",
            headers=headers,
            json=tag_data
        )
        response.raise_for_status()
        tag_ids[tag_name] = response.json()["id"]

    # Create QAndAs
    for q_and_a in small_q_and_a:
        q_and_a_data = {
            "question": q_and_a["question"],
            "answer": q_and_a["answer"],
            "agent_id": agent_id
        }

        response = requests.post(
            f"{BASE_URL}/q-and-a",
            headers=headers,
            json=q_and_a_data
        )
        response.raise_for_status()
        q_and_a_id = response.json()["id"]

        # Link QAndA to tags
        for tag_name in q_and_a["tags"]:
            if tag_name in tag_ids:
                q_and_a_tag_data = {"q_and_a_id": q_and_a_id, "tag_id": tag_ids[tag_name]}
                requests.post(
                    f"{BASE_URL}/q-and-a-tag",
                    headers=headers,
                    json=q_and_a_tag_data
                )

    # Set up guardrails
    for guardrail in some_guardrails:
        guardrail_data = {"agent_id": agent_id, **guardrail}
        response = requests.post(
            f"{BASE_URL}/guardrail",
            headers=headers,
            json=guardrail_data
        )
        response.raise_for_status()

    # Set up state predicate
    response = requests.put(
        f"{BASE_URL}/state-predicate/{agent_id}",
        headers=headers,
        json={"state_predicate": predicate_state}
    )
    response.raise_for_status()

    # Set up actions
    for action in some_actions:
        action_data = {"agent_id": agent_id, **action}
        response = requests.post(
            f"{BASE_URL}/action",
            headers=headers,
            json=action_data
        )
        response.raise_for_status()

def chat(agent_id, message, conversation_id: str):
    """Send a message to the agent and get a response."""
    url = f"{BASE_URL}/chat/{agent_id}"

    try:
        response = requests.post(
            url,
            headers=headers,
            json={
                "user_message": message,
                "conversation_uuid": conversation_id
            }
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error sending message: {e}")
        return None


def main():
    print("Creating a new agent...")
    agent = create_agent()

    agent_id = agent["id"]
    print(f"Created agent with ID: {agent_id=}")

    print("adding knowledge to the agent")
    setup_agent_knowledge(agent_id)
    print("\nAgent is ready! Type your message or 'exit' to quit.")

    conversation_id = str(uuid.uuid4())

    # Example message exchange
    while True:
        user_input = input("\nYou: ")
        if user_input.lower() in ["exit", "quit"]:
            break

        response = chat(agent_id, user_input, conversation_id)
        if response:
            print(f"\nAssistant: {response['answer']}")
        else:
            print("Failed to get a response from the API")


if __name__ == "__main__":
    main()

Next Steps

Now that you have created your agent with all the necessary components, you can:

1. Test the agent by sending messages through the chat function
2. Customize the Q&A, tags, actions, and state transitions for your specific use case
3. Integrate the agent into your application using the provided API endpoints