Build AI agents with n8n and Peliqan

included_schemas = {"chargebee", "pipedrive", "odoo"} # Change this

output_lines = []
for db in pq.list_databases():
    for table in db["tables"]:
        schema_name = next((s["name"] for s in db["schemas"] if s["id"] == table["schema_id"]), None)
        if schema_name not in included_schemas:
            continue
            
        table_meta = pq.get_table(table_id=table["id"])
        columns = table_meta.get("all_fields", [])
        column_names = [col["name"] for col in columns if not col["name"].startswith("_sdc")]

        if column_names:
            output_lines.append(f"\n\nColumns in table {table['name_in_query']}:\n\n" + "\n".join(column_names))

output = "".join(output_lines)
st.text(output)

from datetime import datetime, timedelta

def handler(request):
    cache = pq.get_state()
    if cache and datetime.now() - datetime.strptime(cache["timestamp"], "%Y-%m-%d %H:%M:%S.%f") <= timedelta(hours=24):
        return cache["datamodel"]
        
    included_dbs = [pq.DW_NAME]
    included_schemas = ["chargebee", "pipedrive", "odoo"] # Change this
    
    output = ""
    databases = pq.list_databases()
    for db in databases:
        if db["name"] in included_dbs:
        
            for table in db['tables']:
                
                for schema in db['schemas']:
                    if schema['id'] == table['schema_id']:
                        table['schema_name'] = schema['name']
        
                if table['schema_name'] in included_schemas:
                    table['database_name'] = db['name']
                    table['server_name'] = db['server']['name']
                    table['fully_qualified_name'] = f"{table['database_name']}.{table['name_in_query']}"
                    
                    output += f"\n\nColumns in table %s:\n\n" % table["name_in_query"]
            
                    table_meta_data = pq.get_table(table_id=table["id"])
                    columns = table_meta_data['all_fields']
                    column_names = []
                    for column in columns:
                        column_name = column["name"]
                        column_names.append(column_name)
                    
                    column_names_str = "\n".join(column_names)
                    output += column_names_str

    cache = {
        "datamodel": output,
        "timestamp": datetime.now() 
    }
    pq.set_state(cache)
    
    return output

# This pipeline will vectorize source data (create embeddings) for RAG search in an AI Agent.
# You can build an AI Agent in a Streamit app in Peliqan (see templates) 
# Or e.g. in n8n: add "Postgres PGVector Store" as a tool to the AI Agent, and connect it to your Peliqan data warehouse DB, using table rag.documents

import openai
import json
dbconn = pq.dbconnect(pq.DW_NAME)

openai.api_key = pq.get_secret("OpenAI API key") # Add a secret store under Connections with your OpenAI API key
EMBED_MODEL = "text-embedding-3-large"  # or text-embedding-3-small (dimension: 1536)
EMBED_MODEL_DIMENSION = 3072

def get_embedding(text: str) -> list[float]:
    resp = openai.embeddings.create(model=EMBED_MODEL, input=text)
    return resp.data[0].embedding

def create_table():
    query_create_extension = "CREATE EXTENSION IF NOT EXISTS vector"
    dbconn.execute(pq.DW_NAME, query = query_create_extension)
    
    query_create_table = f"""
        CREATE TABLE IF NOT EXISTS rag.documents (
            id SERIAL PRIMARY KEY,
            text TEXT,
            metadata JSONB, 
            embedding vector({EMBED_MODEL_DIMENSION})
        )
    """
    dbconn.execute(pq.DW_NAME, query = query_create_table)

def vectorize():
    create_table()
    
    # replace this, get records from source table in Peliqan DWH (e.g. Notion pages)
    records = [
        {"id": 1, "text": "Hello world"},
        {"id": 2, "text": "OpenAI makes embeddings easy"},
        {"id": 3, "text": "Postgres with pgvector is powerful"}
    ]
    
    for record in records:
        source_id = record["id"]
        st.write(f"Vectorizing source id {source_id}")
        embedding = get_embedding(record["text"])
        metadata = { 
            "source_id": source_id
        }

        text_str = record["text"].replace('\'','\'\'')
        metadata_str = json.dumps(metadata).replace("'", "''")   # escape single quotes
        embedding_str = "[" + ",".join(f"{x:.6f}" for x in embedding) + "]"  # compact vector

        query_remove_existing_vector = f"DELETE FROM rag.documents WHERE metadata ->> 'source_id' = '{source_id}';"
        result = dbconn.execute(pq.DW_NAME, query = query_remove_existing_vector)
        if result.get("status") != "success":
            st.write(result)
        
        query_insert = f"INSERT INTO rag.documents (text, metadata, embedding) VALUES ('%s', '%s', '%s');" % (text_str, metadata_str, embedding_str)
        result = dbconn.execute(pq.DW_NAME, query = query_insert)
        if result.get("status") != "success":
            st.write(result)

vectorize()