Hello IRIS Fans and Welcome to IRIS iRacing!

Here were going to take 3 laps of your time and demonstrate how I wired up my Racing SIM to IRIS for "As Real Time as It Gets" Metrics reporting.  I missed the window for the contest, which happens quite often, but I still ended up 3rd I think in the demo race in the video below.

Technical Salad

Below are the technical ingredients for this demonstration for a salad you can post on Instragram.

• Python iRacing SDK ( https://github.com/kutu/pyirsdk/tree/master )
• InterSystems IRIS Community / Python Interop
• Prometheus
• Grafana
• Custom Metric Class

Hi, Community!

🐍Great news—you can now use Python to customize your productions and add custom code!

Using Embedded Python in the BPL and DTL Editors

I combined @Rodolfo Pscheidt https://github.com/RodolfoPscheidtJr/ollama-ai-iris with some files from @Guillaume Rongier https://openexchange.intersystems.com/package/iris-rag-demo.

My own project is https://github.com/oliverwilms/ollama-ai-iris

I can run load_data.py and it connects to IRIS (same container).

When I try to run query_data.py https://github.com/oliverwilms/ollama-ai-iris/blob/main/query_data.py , it cannot connect to ollama:

ConnectionError: Failed to connect to Ollama. Please check that Ollama is downloaded, running and accessible.

The InterSystems IRIS is a great data platform and it is met the current features required by the market. In this article, you see the top 10:

Note: this list was updated because many features are added to IRIS in last 3 years (thanks @Kristina Lauer)

I read the article by @Rodolfo Pscheidt Jr:

https://community.intersystems.com/post/ollama-ai-iris

I forked his app and copied selected files from @Guillaume Rongier iris-rag-demo to make it containerized:

ollama-ai-iris/test_conn.py at 3da62770a90e34eff92241abbfc4f2c2a7027898 · oliverwilms/ollama-ai-iris
 

I struggle to understand what each element is or I do not understand why the code does not work:

url = f"iris://teste:teste@localhost:51774/TESTE"
 

Is teste:teste refer to username / password? I tried _SYSTEM:SYS

Is 51774 webserver port? Is TESTE namespace?

I cloned iris-rag-demo from Open Exchange and issued docker-compose up -d. I went to the front end and type in the chat message:

Who was the 46th President of United States of America?

I got a run time error:

RuntimeError: ERROR <Ens>ErrBPTerminated: Terminating BP ChatProcess # due to error: ERROR #5002: ObjectScript error: <PYTHON EXCEPTION> *<class 'RuntimeError'>: <PYTHON EXCEPTION> <class 'ValueError' > ERROR #5002: ObjectScript error: <PYTHON EXCEPTION> *<class 'RuntimeError'

Traceback:

Interested in sharing your thoughts? Sign up to participate here.

I know that people who are completely new to VS Code, Git, Docker, FHIR, and other tools can sometimes struggle with setting up the environment. So I decided to write an article that walks through the entire setup process step by step to make it easier to get started.

I’d really appreciate it if you could leave a comment at the end - let me know if the instructions were clear, if anything was missing, or if there’s anything else you'd find helpful.

The setup includes:

✅ VS Code – Code editor
✅ Git – Version control system
✅ Docker – Runs an instance of IRIS for Health Community
✅ VS Code REST Client Extension – For running FHIR API queries
✅ Python – For writing FHIR-based scripts
✅ Jupyter Notebooks – For AI and FHIR assignments

Before you begin: Ensure you have administrator privileges on your system.

In addition to reading the guide, you can also follow the steps in the videos:

For Windows

Hello Community,

I got the PROTECT error while running functions. But, I could able to call the classmethods and methods in class definition with classMethodObject, classMethodValue etc.. from python. without any errors
python code

irispy.functionString('fnString','IRISPython',14)
irispy.function('fnString','IRISPython',14)

raise RuntimeError(error_message)
RuntimeError: <PROTECT> *Function not allowed

IRISPython.mac
fnString(fn1) public {
  quit"Hello "_fn1
}

​

An experiment on how to use the LangChain framework, IRIS Vector Search, and LLMs to generate IRIS-compatible SQL from user prompts.

This article was based in this notebook. You can run it with a ready to use environment with this application in OpenExchange.

Setup

First, we need to install the necessary libraries:

!pip install --upgrade --quiet langchain langchain-openai langchain-iris pandas

Next, we import the required modules and set up the environment:

import os
import datetime
import hashlib
from copy import deepcopy
from sqlalchemy import create_engine
import getpass
import pandas as pd
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.docstore.document import Document
from langchain_community.document_loaders import DataFrameLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain_core.output_parsers import StrOutputParser
from langchain.globals import set_llm_cache
from langchain.cache import SQLiteCache
from langchain_iris import IRISVector

We will use SQLiteCache to cache the LLM calls:

# Cache for LLM calls
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

Set the IRIS database connection parameters:

# IRIS database connection parameters
os.environ["ISC_LOCAL_SQL_HOSTNAME"] = "localhost"
os.environ["ISC_LOCAL_SQL_PORT"] = "1972"
os.environ["ISC_LOCAL_SQL_NAMESPACE"] = "IRISAPP"
os.environ["ISC_LOCAL_SQL_USER"] = "_system"
os.environ["ISC_LOCAL_SQL_PWD"] = "SYS"

If the OpenAI API key is not already set in the environment, prompt the user to enter it:

if not "OPENAI_API_KEY" in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass()

Create the connection string for the IRIS database:

# IRIS database connection string
args = {
    'hostname': os.getenv("ISC_LOCAL_SQL_HOSTNAME"), 
    'port': os.getenv("ISC_LOCAL_SQL_PORT"), 
    'namespace': os.getenv("ISC_LOCAL_SQL_NAMESPACE"), 
    'username': os.getenv("ISC_LOCAL_SQL_USER"), 
    'password': os.getenv("ISC_LOCAL_SQL_PWD")
}
iris_conn_str = f"iris://{args['username']}:{args['password']}@{args['hostname']}:{args['port']}/{args['namespace']}"

Establish the connection to the IRIS database:

# Connection to IRIS database
engine = create_engine(iris_conn_str)
cnx = engine.connect().connection

Prepare a dictionary to hold context information for the system prompt:

# Dict for context information for system prompt
context = {}
context["top_k"] = 3

Prompt Creation

To transform user inputs into SQL queries compatible with the IRIS database, we need to create an effective prompt for the language model. We start with an initial prompt that provides basic instructions for generating SQL queries. This template is derived from LangChain's default prompts for MSSQL and customized for the IRIS database.

# Basic prompt template with IRIS database SQL instructions
iris_sql_template = """
You are an InterSystems IRIS expert. Given an input question, first create a syntactically correct InterSystems IRIS query to run and return the answer to the input question.
Unless the user specifies in the question a specific number of examples to obtain, query for at most {top_k} results using the TOP clause as per InterSystems IRIS. You can order the results to return the most informative data in the database.
Never query for all columns from a table. You must query only the columns that are needed to answer the question. Wrap each column name in single quotes ('') to denote them as delimited identifiers.
Pay attention to use only the column names you can see in the tables below. Be careful to not query for columns that do not exist. Also, pay attention to which column is in which table.
Pay attention to use CAST(CURRENT_DATE as date) function to get the current date, if the question involves "today".
Use double quotes to delimit columns identifiers.
Return just plain SQL; don't apply any kind of formatting.
"""

This basic prompt configures the language model (LLM) to function as an SQL expert with specific guidance for the IRIS database. Next, we provide an auxiliary prompt with information about the database schema to avoid hallucinations.

# SQL template extension for including tables context information
tables_prompt_template = """
Only use the following tables:
{table_info}
"""

To enhance the accuracy of the LLM's responses, we use a technique called few-shot prompting. This involves presenting some examples to the LLM.

# SQL template extension for including few shots
prompt_sql_few_shots_template = """
Below are a number of examples of questions and their corresponding SQL queries.

{examples_value}
"""

We define the template for few-shot examples:

# Few shots prompt template
example_prompt_template = "User input: {input}\nSQL query: {query}"
example_prompt = PromptTemplate.from_template(example_prompt_template)

We build the user prompt using the few-shot template:

# User prompt template
user_prompt = "\n" + example_prompt.invoke({"input": "{input}", "query": ""}).to_string()

Finally, we compose all prompts to create the final one:

# Complete prompt template
prompt = (
    ChatPromptTemplate.from_messages([("system", iris_sql_template)])
    + ChatPromptTemplate.from_messages([("system", tables_prompt_template)])
    + ChatPromptTemplate.from_messages([("system", prompt_sql_few_shots_template)])
    + ChatPromptTemplate.from_messages([("human", user_prompt)])
)
prompt

This prompt expects the variables examples_value, input, table_info, and top_k.

Here is how the prompt is structured:

ChatPromptTemplate(
    input_variables=['examples_value', 'input', 'table_info', 'top_k'], 
    messages=[
        SystemMessagePromptTemplate(
            prompt=PromptTemplate(
                input_variables=['top_k'], 
                template=iris_sql_template
            )
        ), 
        SystemMessagePromptTemplate(
            prompt=PromptTemplate(
                input_variables=['table_info'], 
                template=tables_prompt_template
            )
        ), 
        SystemMessagePromptTemplate(
            prompt=PromptTemplate(
                input_variables=['examples_value'], 
                template=prompt_sql_few_shots_template
            )
        ), 
        HumanMessagePromptTemplate(
            prompt=PromptTemplate(
                input_variables=['input'], 
                template=user_prompt
            )
        )
    ]
)

To visualize how the prompt will be sent to the LLM, we can use placeholder values for the required variables:

prompt_value = prompt.invoke({
    "top_k": "<top_k>",
    "table_info": "<table_info>",
    "examples_value": "<examples_value>",
    "input": "<input>"
})
print(prompt_value.to_string())

System: 
You are an InterSystems IRIS expert. Given an input question, first create a syntactically correct InterSystems IRIS query to run and return the answer to the input question.
Unless the user specifies in the question a specific number of examples to obtain, query for at most <top_k> results using the TOP clause as per InterSystems IRIS. You can order the results to return the most informative data in the database.
Never query for all columns from a table. You must query only the columns that are needed to answer the question. Wrap each column name in single quotes ('') to denote them as delimited identifiers.
Pay attention to use only the column names you can see in the tables below. Be careful to not query for columns that do not exist. Also, pay attention to which column is in which table.
Pay attention to use CAST(CURRENT_DATE as date) function to get the current date, if the question involves "today".
Use double quotes to delimit columns identifiers.
Return just plain SQL; don't apply any kind of formatting.

System: 
Only use the following tables:
<table_info>

System: 
Below are a number of examples of questions and their corresponding SQL queries.

<examples_value>

Human: 
User input: <input>
SQL query: 

Now, we are ready to send this prompt to the LLM by providing the necessary variables. Let's move on to the next step when you're ready.

Providing Table Information

To create accurate SQL queries, we need to provide the language model (LLM) with detailed information about the database tables. Without this information, the LLM might generate queries that seem plausible but are incorrect due to hallucinations. Therefore, our first step is to create a function that retrieves table definitions from the IRIS database.

Function to Retrieve Table Definitions

The following function queries the INFORMATION_SCHEMA to get the table definitions for a specified schema. If a specific table is provided, it retrieves the definition for that table; otherwise, it retrieves definitions for all tables in the schema.

def get_table_definitions_array(cnx, schema, table=None):
    cursor = cnx.cursor()

    # Base query to get columns information
    query = """
    SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_DEFAULT, PRIMARY_KEY, null EXTRA
    FROM INFORMATION_SCHEMA.COLUMNS
    WHERE TABLE_SCHEMA = %s
    """
    
    # Parameters for the query
    params = [schema]

    # Adding optional filters
    if table:
        query += " AND TABLE_NAME = %s"
        params.append(table)
    
    # Execute the query
    cursor.execute(query, params)

    # Fetch the results
    rows = cursor.fetchall()
    
    # Process the results to generate the table definition(s)
    table_definitions = {}
    for row in rows:
        table_schema, table_name, column_name, column_type, is_nullable, column_default, column_key, extra = row
        if table_name not in table_definitions:
            table_definitions[table_name] = []
        table_definitions[table_name].append({
            "column_name": column_name,
            "column_type": column_type,
            "is_nullable": is_nullable,
            "column_default": column_default,
            "column_key": column_key,
            "extra": extra
        })

    primary_keys = {}
    
    # Build the output string
    result = []
    for table_name, columns in table_definitions.items():
        table_def = f"CREATE TABLE {schema}.{table_name} (\n"
        column_definitions = []
        for column in columns:
            column_def = f"  {column['column_name']} {column['column_type']}"
            if column['is_nullable'] == "NO":
                column_def += " NOT NULL"
            if column['column_default'] is not None:
                column_def += f" DEFAULT {column['column_default']}"
            if column['extra']:
                column_def += f" {column['extra']}"
            column_definitions.append(column_def)
        if table_name in primary_keys:
            pk_def = f"  PRIMARY KEY ({', '.join(primary_keys[table_name])})"
            column_definitions.append(pk_def)
        table_def += ",\n".join(column_definitions)
        table_def += "\n);"
        result.append(table_def)

    return result

Retrieve Table Definitions for a Schema

For this example, we use the Aviation schema, which is available here.

# Retrieve table definitions for the Aviation schema
tables = get_table_definitions_array(cnx, "Aviation")
print(tables)

This function returns the CREATE TABLE statements for all tables in the Aviation schema:

[
    'CREATE TABLE Aviation.Aircraft (\n  Event bigint NOT NULL,\n  ID varchar NOT NULL,\n  AccidentExplosion varchar,\n  AccidentFire varchar,\n  AirFrameHours varchar,\n  AirFrameHoursSince varchar,\n  AirFrameHoursSinceLastInspection varchar,\n  AircraftCategory varchar,\n  AircraftCertMaxGrossWeight integer,\n  AircraftHomeBuilt varchar,\n  AircraftKey integer NOT NULL,\n  AircraftManufacturer varchar,\n  AircraftModel varchar,\n  AircraftRegistrationClass varchar,\n  AircraftSerialNo varchar,\n  AircraftSeries varchar,\n  Damage varchar,\n  DepartureAirportId varchar,\n  DepartureCity varchar,\n  DepartureCountry varchar,\n  DepartureSameAsEvent varchar,\n  DepartureState varchar,\n  DepartureTime integer,\n  DepartureTimeZone varchar,\n  DestinationAirportId varchar,\n  DestinationCity varchar,\n  DestinationCountry varchar,\n  DestinationSameAsLocal varchar,\n  DestinationState varchar,\n  EngineCount integer,\n  EvacuationOccurred varchar,\n  EventId varchar NOT NULL,\n  FlightMedical varchar,\n  FlightMedicalType varchar,\n  FlightPhase integer,\n  FlightPlan varchar,\n  FlightPlanActivated varchar,\n  FlightSiteSeeing varchar,\n  FlightType varchar,\n  GearType varchar,\n  LastInspectionDate timestamp,\n  LastInspectionType varchar,\n  Missing varchar,\n  OperationDomestic varchar,\n  OperationScheduled varchar,\n  OperationType varchar,\n  OperatorCertificate varchar,\n  OperatorCertificateNum varchar,\n  OperatorCode varchar,\n  OperatorCountry varchar,\n  OperatorIndividual varchar,\n  OperatorName varchar,\n  OperatorState varchar,\n  Owner varchar,\n  OwnerCertified varchar,\n  OwnerCountry varchar,\n  OwnerState varchar,\n  RegistrationNumber varchar,\n  ReportedToICAO varchar,\n  SeatsCabinCrew integer,\n  SeatsFlightCrew integer,\n  SeatsPassengers integer,\n  SeatsTotal integer,\n  SecondPilot varchar,\n  childsub bigint NOT NULL DEFAULT $i(^Aviation.EventC("Aircraft"))\n);',
    'CREATE TABLE Aviation.Crew (\n  Aircraft varchar NOT NULL,\n  ID varchar NOT NULL,\n  Age integer,\n  AircraftKey integer NOT NULL,\n  Category varchar,\n  CrewNumber integer NOT NULL,\n  EventId varchar NOT NULL,\n  Injury varchar,\n  MedicalCertification varchar,\n  MedicalCertificationDate timestamp,\n  MedicalCertificationValid varchar,\n  Seat varchar,\n  SeatbeltUsed varchar,\n  Sex varchar,\n  ShoulderHarnessUsed varchar,\n  ToxicologyTestPerformed varchar,\n  childsub bigint NOT NULL DEFAULT $i(^Aviation.AircraftC("Crew"))\n);',
    'CREATE TABLE Aviation.Event (\n  ID bigint NOT NULL DEFAULT $i(^Aviation.EventD),\n  AirportDirection integer,\n  AirportDistance varchar,\n  AirportElevation integer,\n  AirportLocation varchar,\n  AirportName varchar,\n  Altimeter varchar,\n  EventDate timestamp,\n  EventId varchar NOT NULL,\n  EventTime integer,\n  FAADistrictOffice varchar,\n  InjuriesGroundFatal integer,\n  InjuriesGroundMinor integer,\n  InjuriesGroundSerious integer,\n  InjuriesHighest varchar,\n  InjuriesTotal integer,\n  InjuriesTotalFatal integer,\n  InjuriesTotalMinor integer,\n  InjuriesTotalNone integer,\n  InjuriesTotalSerious integer,\n  InvestigatingAgency varchar,\n  LightConditions varchar,\n  LocationCity varchar,\n  LocationCoordsLatitude double,\n  LocationCoordsLongitude double,\n  LocationCountry varchar,\n  LocationSiteZipCode varchar,\n  LocationState varchar,\n  MidAir varchar,\n  NTSBId varchar,\n  NarrativeCause varchar,\n  NarrativeFull varchar,\n  NarrativeSummary varchar,\n  OnGroundCollision varchar,\n  SkyConditionCeiling varchar,\n  SkyConditionCeilingHeight integer,\n  SkyConditionNonCeiling varchar,\n  SkyConditionNonCeilingHeight integer,\n  TimeZone varchar,\n  Type varchar,\n  Visibility varchar,\n  WeatherAirTemperature integer,\n  WeatherPrecipitation varchar,\n  WindDirection integer,\n  WindDirectionIndicator varchar,\n  WindGust integer,\n  WindGustIndicator varchar,\n  WindVelocity integer,\n  WindVelocityIndicator varchar\n);'
]

With these table definitions, we can proceed to the next step, which is to integrate them into our prompt for the LLM. This ensures that the LLM has accurate and comprehensive information about the database schema when generating SQL queries.

Selecting the Most Relevant Tables

When working with databases, especially larger ones, sending the Data Definition Language (DDL) for all tables in a prompt can be impractical. While this approach might work for small databases, real-world databases often contain hundreds or thousands of tables, making it inefficient to process all of them.

Moreover, it’s unlikely that a language model needs to be aware of every table in the database to generate SQL queries effectively. To address this challenge, we can leverage semantic search capabilities to select only the most relevant tables based on the user's query.

Approach

We achieve this by using semantic search with IRIS Vector Search. Note that this method is most effective if your SQL element identifiers (such as tables, fields, and keys) have meaningful names. If your identifiers are arbitrary codes, consider using a data dictionary instead.

Steps

1. Retrieve Table Information

First, extract the table definitions into a pandas DataFrame:

# Retrieve table definitions into a pandas DataFrame
table_def = get_table_definitions_array(cnx=cnx, schema='Aviation')
table_df = pd.DataFrame(data=table_def, columns=["col_def"])
table_df["id"] = table_df.index + 1
table_df

The DataFrame (table_df) will look something like this:

2. Split Definitions into Documents

Next, split the table definitions into Langchain Documents. This step is crucial for handling large chunks of text and extracting text embeddings:

loader = DataFrameLoader(table_df, page_content_column="col_def")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=400, chunk_overlap=20, separator="\n")
tables_docs = text_splitter.split_documents(documents)
tables_docs

The resulting tables_docs list contains split documents with metadata, like so:

[Document(metadata={'id': 1}, page_content='CREATE TABLE Aviation.Aircraft (\n  Event bigint NOT NULL,\n  ID varchar NOT NULL,\n  ...'),
 Document(metadata={'id': 2}, page_content='CREATE TABLE Aviation.Crew (\n  Aircraft varchar NOT NULL,\n  ID varchar NOT NULL,\n  ...'),
 Document(metadata={'id': 3}, page_content='CREATE TABLE Aviation.Event (\n  ID bigint NOT NULL DEFAULT $i(^Aviation.EventD),\n  ...')]

3. Extract Embeddings and Store in IRIS

Now, use the IRISVector class from langchain-iris to extract embedding vectors and store them:

tables_vector_store = IRISVector.from_documents(
    embedding=OpenAIEmbeddings(), 
    documents=tables_docs,
    connection_string=iris_conn_str,
    collection_name="sql_tables",
    pre_delete_collection=True
)

Note: The pre_delete_collection flag is set to True for demonstration purposes to ensure a fresh collection in each test run. In a production environment, this flag should generally be set to False.

4. Find Relevant Documents

With the table embeddings stored, you can now query for relevant tables based on user input:

input_query = "List the first 2 manufacturers"
relevant_tables_docs = tables_vector_store.similarity_search(input_query, k=3)
relevant_tables_docs

For example, querying for manufacturers might return:

[Document(metadata={'id': 1}, page_content='GearType varchar,\n  LastInspectionDate timestamp,\n  ...'),
 Document(metadata={'id': 1}, page_content='AircraftModel varchar,\n  AircraftRegistrationClass varchar,\n  ...'),
 Document(metadata={'id': 3}, page_content='LocationSiteZipCode varchar,\n  LocationState varchar,\n  ...')]

From the metadata, you can see that only table ID 1 (Aviation.Aircraft) is relevant, which aligns with the query.

5. Handling Edge Cases

While this approach is generally effective, it may not always be perfect. For instance, querying for crash sites might also return less relevant tables:

input_query = "List the top 10 most crash sites"
relevant_tables_docs = tables_vector_store.similarity_search(input_query, k=3)
relevant_tables_docs

Results might include:

[Document(metadata={'id': 3}, page_content='LocationSiteZipCode varchar,\n  LocationState varchar,\n  ...'),
 Document(metadata={'id': 3}, page_content='InjuriesGroundSerious integer,\n  InjuriesHighest varchar,\n  ...'),
 Document(metadata={'id': 1}, page_content='CREATE TABLE Aviation.Aircraft (\n  Event bigint NOT NULL,\n  ID varchar NOT NULL,\n  ...')]

Despite retrieving the correct Aviation.Event table twice, the Aviation.Aircraft table may also appear, which could be improved with additional filtering or thresholding. This is beyond the scope of this example and will be left for future implementations.

6. Define a Function to Retrieve Relevant Tables

To automate this process, define a function to filter and return the relevant tables based on user input:

def get_relevant_tables(user_input, tables_vector_store, table_df):
    relevant_tables_docs = tables_vector_store.similarity_search(user_input)
    relevant_tables_docs_indices = [x.metadata["id"] for x in relevant_tables_docs]
    indices = table_df["id"].isin(relevant_tables_docs_indices)
    relevant_tables_array = [x for x in table_df[indices]["col_def"]]
    return relevant_tables_array

This function will help in efficiently retrieving only the relevant tables to send to the LLM, reducing the prompt length and improving overall query performance.

Selecting the Most Relevant Examples (Few-Shot Prompting)

When working with language models (LLMs), providing them with relevant examples helps ensure accurate and contextually appropriate responses. These examples, referred to as "few-shot" examples, guide the LLM in understanding the structure and context of the queries it should handle.

In our case, we need to populate the examples_value variable with a diverse set of SQL queries that cover a broad spectrum of IRIS SQL syntax and the tables available in the database. This helps prevent the LLM from generating incorrect or irrelevant queries.

Defining Example Queries

Below is a list of example queries designed to illustrate various SQL operations:

examples = [
    {"input": "List all aircrafts.", "query": "SELECT * FROM Aviation.Aircraft"},
    {"input": "Find all incidents for the aircraft with ID 'N12345'.", "query": "SELECT * FROM Aviation.Event WHERE EventId IN (SELECT EventId FROM Aviation.Aircraft WHERE ID = 'N12345')"},
    {"input": "List all incidents in the 'Commercial' operation type.", "query": "SELECT * FROM Aviation.Event WHERE EventId IN (SELECT EventId FROM Aviation.Aircraft WHERE OperationType = 'Commercial')"},
    {"input": "Find the total number of incidents.", "query": "SELECT COUNT(*) FROM Aviation.Event"},
    {"input": "List all incidents that occurred in 'Canada'.", "query": "SELECT * FROM Aviation.Event WHERE LocationCountry = 'Canada'"},
    {"input": "How many incidents are associated with the aircraft with AircraftKey 5?", "query": "SELECT COUNT(*) FROM Aviation.Aircraft WHERE AircraftKey = 5"},
    {"input": "Find the total number of distinct aircrafts involved in incidents.", "query": "SELECT COUNT(DISTINCT AircraftKey) FROM Aviation.Aircraft"},
    {"input": "List all incidents that occurred after 5 PM.", "query": "SELECT * FROM Aviation.Event WHERE EventTime > 1700"},
    {"input": "Who are the top 5 operators by the number of incidents?", "query": "SELECT TOP 5 OperatorName, COUNT(*) AS IncidentCount FROM Aviation.Aircraft GROUP BY OperatorName ORDER BY IncidentCount DESC"},
    {"input": "Which incidents occurred in the year 2020?", "query": "SELECT * FROM Aviation.Event WHERE YEAR(EventDate) = '2020'"},
    {"input": "What was the month with most events in the year 2020?", "query": "SELECT TOP 1 MONTH(EventDate) EventMonth, COUNT(*) EventCount FROM Aviation.Event WHERE YEAR(EventDate) = '2020' GROUP BY MONTH(EventDate) ORDER BY EventCount DESC"},
    {"input": "How many crew members were involved in incidents?", "query": "SELECT COUNT(*) FROM Aviation.Crew"},
    {"input": "List all incidents with detailed aircraft information for incidents that occurred in the year 2012.", "query": "SELECT e.EventId, e.EventDate, a.AircraftManufacturer, a.AircraftModel, a.AircraftCategory FROM Aviation.Event e JOIN Aviation.Aircraft a ON e.EventId = a.EventId WHERE Year(e.EventDate) = 2012"},
    {"input": "Find all incidents where there were more than 5 injuries and include the aircraft manufacturer and model.", "query": "SELECT e.EventId, e.InjuriesTotal, a.AircraftManufacturer, a.AircraftModel FROM Aviation.Event e JOIN Aviation.Aircraft a ON e.EventId = a.EventId WHERE e.InjuriesTotal > 5"},
    {"input": "List all crew members involved in incidents with serious injuries, along with the incident date and location.", "query": "SELECT c.CrewNumber AS 'Crew Number', c.Age, c.Sex AS Gender, e.EventDate AS 'Event Date', e.LocationCity AS 'Location City', e.LocationState AS 'Location State' FROM Aviation.Crew c JOIN Aviation.Event e ON c.EventId = e.EventId WHERE c.Injury = 'Serious'"}
]

Selecting Relevant Examples

Given the ever-expanding list of examples, it’s impractical to provide the LLM with all of them. Instead, we use IRIS Vector Search along with the SemanticSimilarityExampleSelector class to identify the most relevant examples based on user prompts.

Define the Example Selector:

example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    IRISVector,
    k=5,
    input_keys=["input"],
    connection_string=iris_conn_str,
    collection_name="sql_samples",
    pre_delete_collection=True
)

Note: The pre_delete_collection flag is used here for demonstration purposes to ensure a fresh collection in each test run. In a production environment, this flag should be set to False to avoid unnecessary deletions.

Query the Selector:

To find the most relevant examples for a given input, use the selector as follows:

input_query = "Find all events in 2010 informing the Event Id and date, location city and state, aircraft manufacturer and model."
relevant_examples = example_selector.select_examples({"input": input_query})

The results might look like this:

[{'input': 'List all incidents with detailed aircraft information for incidents that occurred in the year 2012.', 'query': 'SELECT e.EventId, e.EventDate, a.AircraftManufacturer, a.AircraftModel, a.AircraftCategory FROM Aviation.Event e JOIN Aviation.Aircraft a ON e.EventId = a.EventId WHERE Year(e.EventDate) = 2012'},
 {'input': "Find all incidents for the aircraft with ID 'N12345'.", 'query': "SELECT * FROM Aviation.Event WHERE EventId IN (SELECT EventId FROM Aviation.Aircraft WHERE ID = 'N12345')"},
 {'input': 'Find all incidents where there were more than 5 injuries and include the aircraft manufacturer and model.', 'query': 'SELECT e.EventId, e.InjuriesTotal, a.AircraftManufacturer, a.AircraftModel FROM Aviation.Event e JOIN Aviation.Aircraft a ON e.EventId = a.EventId WHERE e.InjuriesTotal > 5'},
 {'input': 'List all aircrafts.', 'query': 'SELECT * FROM Aviation.Aircraft'},
 {'input': 'Find the total number of distinct aircrafts involved in incidents.', 'query': 'SELECT COUNT(DISTINCT AircraftKey) FROM Aviation.Aircraft'}]

If you specifically need examples related to quantities, you can query the selector accordingly:

input_query = "What is the number of incidents involving Boeing aircraft."
quantity_examples = example_selector.select_examples({"input": input_query})

The output may be:

[{'input': 'How many incidents are associated with the aircraft with AircraftKey 5?', 'query': 'SELECT COUNT(*) FROM Aviation.Aircraft WHERE AircraftKey = 5'},
 {'input': 'Find the total number of distinct aircrafts involved in incidents.', 'query': 'SELECT COUNT(DISTINCT AircraftKey) FROM Aviation.Aircraft'},
 {'input': 'How many crew members were involved in incidents?', 'query': 'SELECT COUNT(*) FROM Aviation.Crew'},
 {'input': 'Find all incidents where there were more than 5 injuries and include the aircraft manufacturer and model.', 'query': 'SELECT e.EventId, e.InjuriesTotal, a.AircraftManufacturer, a.AircraftModel FROM Aviation.Event e JOIN Aviation.Aircraft a ON e.EventId = a.EventId WHERE e.InjuriesTotal > 5'},
 {'input': 'List all incidents with detailed aircraft information for incidents that occurred in the year 2012.', 'query': 'SELECT e.EventId, e.EventDate, a.AircraftManufacturer, a.AircraftModel, a.AircraftCategory FROM Aviation.Event e JOIN Aviation.Aircraft a ON e.EventId = a.EventId WHERE Year(e.EventDate) = 2012'}]

This output includes examples that specifically address counting and quantities.

Future Considerations

While the SemanticSimilarityExampleSelector is powerful, it’s important to note that not all selected examples may be perfect. Future improvements may involve adding filters or thresholds to exclude less relevant results, ensuring that only the most appropriate examples are provided to the LLM.

Accuracy Test

To assess the performance of the prompt and SQL query generation, we need to set up and run a series of tests. The goal is to evaluate how well the LLM generates SQL queries based on user inputs, with and without the use of example-based few shots.

Function to Generate SQL Queries

We start by defining a function that uses the LLM to generate SQL queries based on the provided context, prompt, user input, and other parameters:

def get_sql_from_text(context, prompt, user_input, use_few_shots, tables_vector_store, table_df, example_selector=None, example_prompt=None):
    relevant_tables = get_relevant_tables(user_input, tables_vector_store, table_df)
    context["table_info"] = "\n\n".join(relevant_tables)

    examples = example_selector.select_examples({"input": user_input}) if example_selector else []
    context["examples_value"] = "\n\n".join([
        example_prompt.invoke(x).to_string() for x in examples
    ])
    
    model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    output_parser = StrOutputParser()
    chain_model = prompt | model | output_parser
    
    response = chain_model.invoke({
        "top_k": context["top_k"],
        "table_info": context["table_info"],
        "examples_value": context["examples_value"],
        "input": user_input
    })
    return response

Execute the Prompt

Test the prompt with and without examples:

# Prompt execution **with** few shots
input = "Find all events in 2010 informing the Event Id and date, location city and state, aircraft manufacturer and model."
response_with_few_shots = get_sql_from_text(
    context, 
    prompt, 
    user_input=input, 
    use_few_shots=True, 
    tables_vector_store=tables_vector_store, 
    table_df=table_df,
    example_selector=example_selector, 
    example_prompt=example_prompt,
)
print(response_with_few_shots)

SELECT e.EventId, e.EventDate, e.LocationCity, e.LocationState, a.AircraftManufacturer, a.AircraftModel
FROM Aviation.Event e
JOIN Aviation.Aircraft a ON e.EventId = a.EventId
WHERE Year(e.EventDate) = 2010

# Prompt execution **without** few shots
input = "Find all events in 2010 informing the Event Id and date, location city and state, aircraft manufacturer and model."
response_with_no_few_shots = get_sql_from_text(
    context, 
    prompt, 
    user_input=input, 
    use_few_shots=False, 
    tables_vector_store=tables_vector_store, 
    table_df=table_df,
)
print(response_with_no_few_shots)

SELECT TOP 3 "EventId", "EventDate", "LocationCity", "LocationState", "AircraftManufacturer", "AircraftModel"
FROM Aviation.Event e
JOIN Aviation.Aircraft a ON e.ID = a.Event
WHERE e.EventDate >= '2010-01-01' AND e.EventDate < '2011-01-01'

Utility Functions for Testing

To test the generated SQL queries, we define some utility functions:

def execute_sql_query(cnx, query):
    try:
        cursor = cnx.cursor()
        cursor.execute(query)
        rows = cursor.fetchall()
        return rows
    except:
        print('Error running query:')
        print(query)
        print('-'*80)
    return None

def sql_result_equals(cnx, query, expected):
    rows = execute_sql_query(cnx, query)
    result = [set(row._asdict().values()) for row in rows or []]
    if result != expected and rows is not None:
        print('Result not as expected for query:')
        print(query)
        print('-'*80)
    return result == expected

# SQL test for prompt **with** few shots
print("SQL is OK" if not execute_sql_query(cnx, response_with_few_shots) is None else "SQL is not OK")

    SQL is OK

# SQL test for prompt **without** few shots
print("SQL is OK" if not execute_sql_query(cnx, response_with_no_few_shots) is None else "SQL is not OK")

    error on running query: 
    SELECT TOP 3 "EventId", "EventDate", "LocationCity", "LocationState", "AircraftManufacturer", "AircraftModel"
    FROM Aviation.Event e
    JOIN Aviation.Aircraft a ON e.ID = a.Event
    WHERE e.EventDate >= '2010-01-01' AND e.EventDate < '2011-01-01'
    --------------------------------------------------------------------------------
    SQL is not OK

Define and Execute Tests

Define a set of test cases and run them:

tests = [{
    "input": "What were the top 3 years with the most recorded events?",
    "expected": [{128, 2003}, {122, 2007}, {117, 2005}]
},{
    "input": "How many incidents involving Boeing aircraft.",
    "expected": [{5}]
},{
    "input": "How many incidents that resulted in fatalities.",
    "expected": [{237}]
},{
    "input": "List event Id and date and, crew number, age and gender for incidents that occurred in 2013.",
    "expected": [{1, datetime.datetime(2013, 3, 4, 11, 6), '20130305X71252', 59, 'M'},
                 {1, datetime.datetime(2013, 1, 1, 15, 0), '20130101X94035', 32, 'M'},
                 {2, datetime.datetime(2013, 1, 1, 15, 0), '20130101X94035', 35, 'M'},
                 {1, datetime.datetime(2013, 1, 12, 15, 0), '20130113X42535', 25, 'M'},
                 {2, datetime.datetime(2013, 1, 12, 15, 0), '20130113X42535', 34, 'M'},
                 {1, datetime.datetime(2013, 2, 1, 15, 0), '20130203X53401', 29, 'M'},
                 {1, datetime.datetime(2013, 2, 15, 15, 0), '20130218X70747', 27, 'M'},
                 {1, datetime.datetime(2013, 3, 2, 15, 0), '20130303X21011', 49, 'M'},
                 {1, datetime.datetime(2013, 3, 23, 13, 52), '20130326X85150', 'M', None}]
},{
    "input": "Find the total number of incidents that occurred in the United States.",
    "expected": [{1178}]
},{
    "input": "List all incidents latitude and longitude coordinates with more than 5 injuries that occurred in 2010.",
    "expected": [{-78.76833333333333, 43.25277777777778}]
},{
    "input": "Find all incidents in 2010 informing the Event Id and date, location city and state, aircraft manufacturer and model.",
    "expected": [
        {datetime.datetime(2010, 5, 20, 13, 43), '20100520X60222', 'CIRRUS DESIGN CORP', 'Farmingdale', 'New York', 'SR22'},
        {datetime.datetime(2010, 4, 11, 15, 0), '20100411X73253', 'CZECH AIRCRAFT WORKS SPOL SRO', 'Millbrook', 'New York', 'SPORTCRUISER'},
        {'108', datetime.datetime(2010, 1, 9, 12, 55), '20100111X41106', 'Bayport', 'New York', 'STINSON'},
        {datetime.datetime(2010, 8, 1, 14, 20), '20100801X85218', 'A185F', 'CESSNA', 'New York', 'Newfane'}
    ]
}]

Accuracy Evaluation

Run the tests and calculate the accuracy:

def execute_tests(cnx, context, prompt, use_few_shots, tables_vector_store, table_df, example_selector, example_prompt):
    tests_generated_sql = [(x, get_sql_from_text(
            context, 
            prompt, 
            user_input=x['input'], 
            use_few_shots=use_few_shots, 
            tables_vector_store=tables_vector_store, 
            table_df=table_df,
            example_selector=example_selector if use_few_shots else None, 
            example_prompt=example_prompt if use_few_shots else None,
        )) for x in deepcopy(tests)]
    
    tests_sql_executions = [(x[0], sql_result_equals(cnx, x[1], x[0]['expected'])) 
                            for x in tests_generated_sql]
    
    accuracy = sum(1 for i in tests_sql_executions if i[1] == True) / len(tests_sql_executions)
    print(f'Accuracy: {accuracy}')
    print('-'*80)

Results

# Accuracy tests for prompts executed **without** few shots
use_few_shots = False
execute_tests(
    cnx,
    context, 
    prompt, 
    use_few_shots, 
    tables_vector_store, 
    table_df, 
    example_selector, 
    example_prompt
)

    error on running query: 
    SELECT "EventDate", COUNT("EventId") as "TotalEvents"
    FROM Aviation.Event
    GROUP BY "EventDate"
    ORDER BY "TotalEvents" DESC
    TOP 3;
    --------------------------------------------------------------------------------
    error on running query: 
    SELECT "EventId", "EventDate", "C"."CrewNumber", "C"."Age", "C"."Sex"
    FROM "Aviation.Event" AS "E"
    JOIN "Aviation.Crew" AS "C" ON "E"."ID" = "C"."EventId"
    WHERE "E"."EventDate" >= '2013-01-01' AND "E"."EventDate" < '2014-01-01'
    --------------------------------------------------------------------------------
    result not expected for query: 
    SELECT TOP 3 "e"."EventId", "e"."EventDate", "e"."LocationCity", "e"."LocationState", "a"."AircraftManufacturer", "a"."AircraftModel"
    FROM "Aviation"."Event" AS "e"
    JOIN "Aviation"."Aircraft" AS "a" ON "e"."ID" = "a"."Event"
    WHERE "e"."EventDate" >= '2010-01-01' AND "e"."EventDate" < '2011-01-01'
    --------------------------------------------------------------------------------
    accuracy: 0.5714285714285714
    --------------------------------------------------------------------------------

# Accuracy tests for prompts executed **with** few shots
use_few_shots = True
execute_tests(
    cnx,
    context, 
    prompt, 
    use_few_shots, 
    tables_vector_store, 
    table_df, 
    example_selector, 
    example_prompt
)

    error on running query: 
    SELECT e.EventId, e.EventDate, e.LocationCity, e.LocationState, a.AircraftManufacturer, a.AircraftModel
    FROM Aviation.Event e
    JOIN Aviation.Aircraft a ON e.EventId = a.EventId
    WHERE Year(e.EventDate) = 2010 TOP 3
    --------------------------------------------------------------------------------
    accuracy: 0.8571428571428571
    --------------------------------------------------------------------------------

Conclusion

The accuracy of SQL queries generated with examples (few shots) is approximately 49% higher compared to those generated without examples (85% vs. 57%).

References

• https://python.langchain.com/v0.1/docs/expression_language/get_started/
• https://python.langchain.com/v0.1/docs/use_cases/sql/prompting/
• https://python.langchain.com/v0.1/docs/modules/model_io/prompts/composition/

Hello Team,

I got xDBC protocol is not compatible while executing python script. How to fix this error

C:\Users\ak\Desktop\lpyth\iris>C:/Users/ak/AppData/Local/Programs/Python/Python312/python.exe c:/Users/ak/Desktop/lpyth/iris/irisconn.py
An error occurred: connection failed: IRIS xDBC protocol is not compatible

py -m pip list
Package            Version
------------------ ---------
intersystems-iris  3.9.2

I'm exploring this right now: given a bunch of types defined as Pydantic models, how can I come up with an equivalent %RegisteredObject/%SerialObject and convert to/from (e.g., to support persistence and match validation as much as possible)?

People who know Python better than I do (e.g., your average undergraduate from this decade): is this a stupid idea or a cool idea? Has anyone else done this before?

You can learn a lot from a first impression; we want to hear about yours.

As a continuation of our User Insights Interview program (see this post for more details), we’re expanding our scope to include Python developers, particularly those of you who are new to working with InterSystems technology. We’re looking to conduct one-on-one interviews to hear your honest thoughts about getting started: What made sense, what didn’t, and where we could improve.

Interested in sharing your thoughts? Sign up to participate here.

I'm a huge sci-fi fan, but while I'm fully onboard the Star Wars train (apologies to my fellow Trekkies!), but I've always appreciated the classic episodes of Star Trek from my childhood. The diverse crew of the USS Enterprise, each masterminding their unique roles, is a perfect metaphor for understanding AI agents and their power in projects like Facilis. So, let's embark on an intergalactic mission, leveraging AI as our ship's crew and  boldly go where no man has gone before!  This teamwork concept is a wonderful analogy to illustrate how AI agents work and how we use them in our DC-Facilis project. So, let’s dive in and assume the role of a starship captain, leading an AI crew into unexplored territories!

Welcome to CrewAI!

To manage our AI crew, we use a fantastic framework called CrewAI. It's lean, lightning-fast, and operates as a multi-agent Python platform. One of the reasons we love it, besides the fact that it was created by another Brazilian, is its incredible flexibility and role-based design.

from crewai import Agent, Task, Crew

Meet the Planners

In Facilis, our AI agents are divided into two groups. Let's start with the first one I like to call "The Planners."

The Extraction Agent

The main role of Facilis is to take a natural language description of a REST service and auto-magically create all the necessary interoperability. So, our first crew member is the Extraction Agent. This agent is tasked with "extracting" API specifications from a user's prompt description.

Here's what the Extraction Agent looks out for:

• Host (required)
• Endpoint (required)
• Params (optional)
• Port (if available)
• JSON model (for POST/PUT/PATCH/DELETE)
• Authentication (if applicable)

    def create_extraction_agent(self) -> Agent:
        return Agent(
            role='API Specification Extractor',
            goal='Extract API specifications from natural language descriptions',
            backstory=dedent("""
                You are specialized in interpreting natural language descriptions
                and extracting structured API specifications.
            """),
            allow_delegation=True,
            llm=self.llm
        )

    def extract_api_specs(self, descriptions: List[str]) -> Task:
        return Task(
            description=dedent(f"""
                Extract API specifications from the following descriptions:
                {json.dumps(descriptions, indent=2)}
                
                For each description, extract:
                - host (required)
                - endpoint (required)
                - HTTP_Method (required)
                - params (optional)
                - port (if available)
                - json_model (for POST/PUT/PATCH/DELETE)
                - authentication (if applicable)
                
                Mark any missing required fields as 'missing'.
                Return results in JSON format as an array of specifications.
            """),
            expected_output="""A JSON array containing extracted API specifications with all required and optional fields""",
            agent=self.extraction_agent
        )

The Validation Agent

Next up, the Validation Agent! Their mission is to ensure that the API specifications gathered by the Extraction Agent are correct and consistent. They check:

1. Valid host format
2. Endpoint starting with '/'
3. Valid HTTP methods (GET, POST, PUT, DELETE, PATCH)
4. Valid port number (if provided)
5. JSON model presence for applicable methods.

    def create_validation_agent(self) -> Agent:
        return Agent(
            role='API Validator',
            goal='Validate API specifications for correctness and consistency',
            backstory=dedent("""
                You are an expert in API validation, ensuring all specifications
                meet the required standards and format.
            """),
            allow_delegation=False,
            llm=self.llm
        )

 def validate_api_spec(self, extracted_data: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Validate the following API specification:
                {json.dumps(extracted_data, indent=2)}
                
                Check for:
                1. Valid host format
                2. Endpoint starts with '/'
                3. Valid HTTP method (GET, POST, PUT, DELETE, PATCH)
                4. Valid port number (if provided)
                5. JSON model presence for POST/PUT/PATCH/DELETE methods
                
                Return validation results in JSON format.
            """),
            expected_output="""A JSON object containing validation results with any errors or confirmation of validity""",
            agent=self.validation_agent
        )

The Interaction Agent

Moving on, we meet the Interaction Agent, our User Interaction Specialist. Their role is to obtain any missing API specification fields that were marked by the Extraction Agent and validate them based on the Validation Agent's findings. They interact directly with users to fill any gaps.

The Production Agent

We need two crucial pieces of information to create the necessary interoperability: namespace and production name. The Production Agent engages with users to gather this information, much like the Interaction Agent.

The Documentation Transformation Agent

Once the specifications are ready, it's time to convert them into OpenAPI documentation. The Documentation Transformation Agent, an OpenAPI specialist, takes care of this.

    def create_transformation_agent(self) -> Agent:
        return Agent(
            role='OpenAPI Transformation Specialist',
            goal='Convert API specifications into OpenAPI documentation',
            backstory=dedent("""
                You are an expert in OpenAPI specifications and documentation.
                Your role is to transform validated API details into accurate
                and comprehensive OpenAPI 3.0 documentation.
            """),
            allow_delegation=False,
            llm=self.llm
        )

    def transform_to_openapi(self, validated_endpoints: List[Dict], production_info: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Transform the following validated API specifications into OpenAPI 3.0 documentation:
                
                Production Information:
                {json.dumps(production_info, indent=2)}
                
                Validated Endpoints:
                {json.dumps(validated_endpoints, indent=2)}
                
                Requirements:
                1. Generate complete OpenAPI 3.0 specification
                2. Include proper request/response schemas
                3. Document all parameters and request bodies
                4. Include authentication if specified
                5. Ensure proper path formatting
                
                Return the OpenAPI specification in both JSON and YAML formats.
            """),
            expected_output="""A JSON object containing the complete OpenAPI 3.0 specification with all endpoints and schemas""",
            agent=self.transformation_agent
        )

The Review Agent

After transformation, the OpenAPI documentation undergoes a meticulous review to ensure compliance and quality. The Review Agent follows this checklist:

1. OpenAPI 3.0 Compliance

• Correct version specification
• Required root elements
• Schema structure validation

2. Completeness

• All endpoints documented
• Fully specified parameters
• Defined request/response schemas
• Properly configured security schemes

3. Quality Checks

• Consistent naming conventions
• Clear descriptions
• Proper use of data types
• Meaningful response codes

4. Best Practices

• Proper tag usage
• Consistent parameter naming
• Appropriate security definitions

Finally, if everything looks good, the Review Agent reports a healthy JSON object with the following structure:

{
 "is_valid": boolean,
 "approved_spec": object (the reviewed and possibly corrected OpenAPI spec),
 "issues": [array of strings describing any issues found],
 "recommendations": [array of improvement suggestions]
}

    def create_reviewer_agent(self) -> Agent:
        return Agent(
            role='OpenAPI Documentation Reviewer',
            goal='Ensure OpenAPI documentation compliance and quality',
            backstory=dedent("""
                You are the final authority on OpenAPI documentation quality and compliance.
                With extensive experience in OpenAPI 3.0 specifications, you meticulously
                review documentation for accuracy, completeness, and adherence to standards.
            """),
            allow_delegation=True,
            llm=self.llm
        )


    def review_openapi_spec(self, openapi_spec: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Review the following OpenAPI specification for compliance and quality:
                
                {json.dumps(openapi_spec, indent=2)}
                
                Review Checklist:
                1. OpenAPI 3.0 Compliance
                - Verify correct version specification
                - Check required root elements
                - Validate schema structure
                
                2. Completeness
                - All endpoints properly documented
                - Parameters fully specified
                - Request/response schemas defined
                - Security schemes properly configured
                
                3. Quality Checks
                - Consistent naming conventions
                - Clear descriptions
                - Proper use of data types
                - Meaningful response codes
                
                4. Best Practices
                - Proper tag usage
                - Consistent parameter naming
                - Appropriate security definitions
                
                You must return a JSON object with the following structure:
                {{
                    "is_valid": boolean,
                    "approved_spec": object (the reviewed and possibly corrected OpenAPI spec),
                    "issues": [array of strings describing any issues found],
                    "recommendations": [array of improvement suggestions]
                }}
            """),
            expected_output="""A JSON object containing: is_valid (boolean), approved_spec (object), issues (array), and recommendations (array)""",
            agent=self.reviewer_agent
        )

The Iris Agent

The last agent in the planner group is the Iris Agent, who sends the finalized OpenAPI documentation to Iris.

    def create_iris_i14y_agent(self) -> Agent:
        return Agent(
            role='Iris I14y Integration Specialist',
            goal='Integrate API specifications with Iris I14y service',
            backstory=dedent("""
                You are responsible for ensuring smooth integration between the API
                documentation system and the Iris I14y service. You handle the
                communication with Iris, validate responses, and ensure successful
                integration of API specifications.
            """),
            allow_delegation=False,
            llm=self.llm
        )

    def send_to_iris(self, openapi_spec: Dict, production_info: Dict, review_result: Dict) -> Task:
        return Task(
            description=dedent(f"""
                Send the approved OpenAPI specification to Iris I14y service:

                Production Information:
                - Name: {production_info['production_name']}
                - Namespace: {production_info['namespace']}
                - Is New: {production_info.get('create_new', False)}

                Review Status:
                - Approved: {review_result['is_valid']}
                
                Return the integration result in JSON format.
            """),
            expected_output="""A JSON object containing the integration result with Iris I14y service, including success status and response details""",
            agent=self.iris_i14y_agent
        )

class IrisI14yService:
    def __init__(self):
        self.logger = logging.getLogger('facilis.IrisI14yService')
        self.base_url = os.getenv("FACILIS_URL", "http://dc-facilis-iris-1:52773") 
        self.headers = {
            "Content-Type": "application/json"
        }
        self.timeout = int(os.getenv("IRIS_TIMEOUT", "504"))  # in milliseconds
        self.max_retries = int(os.getenv("IRIS_MAX_RETRIES", "3"))
        self.logger.info("IrisI14yService initialized")

    async def send_to_iris_async(self, payload: Dict) -> Dict:
        """
        Send payload to Iris generate endpoint asynchronously
        """
        self.logger.info("Sending payload to Iris generate endpoint")
        if isinstance(payload, str):
            try:
                json.loads(payload)  
            except json.JSONDecodeError:
                raise ValueError("Invalid JSON string provided")
        
        retry_count = 0
        last_error = None

        # Create timeout for aiohttp
        timeout = aiohttp.ClientTimeout(total=self.timeout / 1000)  # Convert ms to seconds

        while retry_count < self.max_retries:
            try:
                self.logger.info(f"Attempt {retry_count + 1}/{self.max_retries}: Sending request to {self.base_url}/facilis/api/generate")
                
                async with aiohttp.ClientSession(timeout=timeout) as session:
                    async with session.post(
                        f"{self.base_url}/facilis/api/generate",
                        json=payload,
                        headers=self.headers
                    ) as response:
                        if response.status == 200:
                            return await response.json()
                        response.raise_for_status()

            except asyncio.TimeoutError as e:
                retry_count += 1
                last_error = e
                error_msg = f"Timeout occurred (attempt {retry_count}/{self.max_retries})"
                self.logger.warning(error_msg)
                
                if retry_count < self.max_retries:
                    wait_time = 2 ** (retry_count - 1)
                    self.logger.info(f"Waiting {wait_time} seconds before retry...")
                    await asyncio.sleep(wait_time)
                continue

            except aiohttp.ClientError as e:
                error_msg = f"Failed to send to Iris: {str(e)}"
                self.logger.error(error_msg)
                raise IrisIntegrationError(error_msg)

        error_msg = f"Failed to send to Iris after {self.max_retries} attempts due to timeout"
        self.logger.error(error_msg)
        raise IrisIntegrationError(error_msg, last_error)

Meet the Generators

Our second set of agents are - the Generators. They are here to transform the OpenAPI specifications into InterSystems IRIS interoperability. There are eight of them in this group.

The first one is the Analyzer Agent. He's like the planner, mapping out the route. Its job is to delve into the OpenAPI specs and figure out what IRIS Interoperability components are needed.

    def create_analyzer_agent():
        return Agent(
            role="OpenAPI Specification Analyzer",
            goal="Thoroughly analyze OpenAPI specifications and plan IRIS Interoperability components",
            backstory="""You are an expert in both OpenAPI specifications and InterSystems IRIS Interoperability. 
            Your job is to analyze OpenAPI documents and create a detailed plan for how they should be 
            implemented as IRIS Interoperability components.""",
            verbose=False,
            allow_delegation=False,
            tools=[analyze_openapi_tool],
            llm=get_facilis_llm()
        )

   analysis_task = Task(
        description="""Analyze the OpenAPI specification and plan the necessary IRIS Interoperability components. 
        Include a list of all components that should be in the Production class.""",
        agent=analyzer,
        expected_output="A detailed analysis of OpenAPI spec and plan for IRIS components, including Production components list",
        input={
            "openapi_spec": openApiSpec,
            "production_name": "${production_name}" 
        }
    )

Next up, the Business Services (BS) and Business Operations (BO) Agents take over. They generate the Business Services and Business Operations based on the OpenAPI endpoints. They use a handy tool called MessageClassTool to generate the perfect message classes, ensuring the communication.

    def create_bs_generator_agent():
        return Agent(
            role="IRIS Production and Business Service Generator",
            goal="Generate properly formatted IRIS Production and Business Service classes from OpenAPI specifications",
            backstory="""You are an experienced InterSystems IRIS developer specializing in Interoperability Productions.
            Your expertise is in creating Business Services and Productions that can receive and process incoming requests based on
            API specifications.""",
            verbose=False,
            allow_delegation=True,
            tools=[generate_production_class_tool, generate_business_service_tool],
            llm=get_facilis_llm()
        )

    def create_bo_generator_agent():
        return Agent(
            role="IRIS Business Operation Generator",
            goal="Generate properly formatted IRIS Business Operation classes from OpenAPI specifications",
            backstory="""You are an experienced InterSystems IRIS developer specializing in Interoperability Productions.
            Your expertise is in creating Business Operations that can send requests to external systems
            based on API specifications.""",
            verbose=False,
            allow_delegation=True,
            tools=[generate_business_operation_tool, generate_message_class_tool],
            llm=get_facilis_llm()
        )

    bs_generation_task = Task(
        description="Generate Business Service classes based on the OpenAPI endpoints",
        agent=bs_generator,
        expected_output="IRIS Business Service class definitions",
        context=[analysis_task]
    )

    bo_generation_task = Task(
        description="Generate Business Operation classes based on the OpenAPI endpoints",
        agent=bo_generator,
        expected_output="IRIS Business Operation class definitions",
        context=[analysis_task]
    )

    class GenerateMessageClassTool(BaseTool):
        name: str = "generate_message_class"
        description: str = "Generate an IRIS Message class"
        input_schema: Type[BaseModel] = GenerateMessageClassToolInput

        def _run(self, message_name: str, schema_info: Union[str, Dict[str, Any]]) -> str:
            writer = IRISClassWriter()
            try:
                if isinstance(schema_info, str):
                    try:
                        schema_dict = json.loads(schema_info)
                    except json.JSONDecodeError:
                        return "Error: Invalid JSON format for schema info"
                else:
                    schema_dict = schema_info

                class_content = writer.write_message_class(message_name, schema_dict)
                # Store the generated class
                writer.generated_classes[f"MSG.{message_name}"] = class_content
                return class_content
            except Exception as e:
                return f"Error generating message class: {str(e)}"

Once BS and BO have done their thing, it's time for the Production Agent to shine! This agent pulls everything together to create a cohesive production environment.

After everything is set up, next in line is the Validation Agent. This one makes comes in for a final checkup, making sure each Iris class is ok.

Then we have the Export Agent and the Collection Agent. The Export Agent generates the .cls files while the Collection Agent gathers all the file names. Everything gets passed along to the importer, which compiles everything into InterSystems Iris.

    def create_exporter_agent():
        return Agent(
            role="IRIS Class Exporter",
            goal="Export and validate IRIS class definitions to proper .cls files",
            backstory="""You are an InterSystems IRIS deployment specialist. Your job is to ensure 
            that generated IRIS class definitions are properly exported as valid .cls files that 
            can be directly imported into an IRIS environment.""",
            verbose=False,
            allow_delegation=False,
            tools=[export_iris_classes_tool, validate_iris_classes_tool],
            llm=get_facilis_llm()
        )
        
    def create_collector_agent():
        return Agent(
            role="IRIS Class Collector",
            goal="Collect all generated IRIS class files into a JSON collection",
            backstory="""You are a file system specialist responsible for gathering and 
            organizing generated IRIS class files into a structured collection.""",
            verbose=False,
            allow_delegation=False,
            tools=[CollectGeneratedFilesTool()],
            llm=get_facilis_llm()
        )

    export_task = Task(
        description="Export all generated IRIS classes as valid .cls files",
        agent=exporter,
        expected_output="Valid IRIS .cls files saved to output directory",
        context=[bs_generation_task, bo_generation_task],
        input={
            "output_dir": "/home/irisowner/dev/output/iris_classes"  # Optional
        }
    )

    collection_task = Task(
        description="Collect all generated IRIS class files into a JSON collection",
        agent=collector,
        expected_output="JSON collection of all generated .cls files",
        context=[export_task, validate_task],
        input={
            "directory": "./output/iris_classes"
        }
    )

Limitations and Challenges

Our project started as an exciting experiment, where my fellow musketeers and I aimed to create a fully automated tool using agents. It was a wild ride! Our main focus was on REST API integrations. It's always a joy to get a task with an OpenAPI specification to integrate; however, legacy systems can be a whole different story. We thought automating these tasks could be incredibly useful. But every adventure has its twists: One of the biggest challenges was instructing the AI to convert OpenAPI to Iris Interoperability. We started with openAI GPT3.5-turbo model, on initial tests proved difficult with debugging and preventing breaks. Switching to Anthropic Claude 3.7 Sonnet showed better results for the Generator group but not so much for the Planners... This led us to split our environment configurations, using different LLM providers for flexibility. We used GPT3.5-turbo for planning and Claude sonnet for generation, great combo! This combination worked well but we did encounter issues with hallucinations. Moving to GT4o improved results, yet we still faced hallucinations creating Iris classes and sometimes unnecessary OpenAPI specifications like the renowned Pet Store OpenAPI example. we had a blast learning along the way, and I'm super excited about the amazing future in this field with countless possibilities!

Summary of Alerts

Detail of Alerts

DP-439207 - AIX JSON Unicode Parsing Issue

Hi Community,
In this article, I will introduce my application iris-AgenticAI .

The rise of agentic AI marks a transformative leap in how artificial intelligence interacts with the world—moving beyond static responses to dynamic, goal-driven problem-solving. Powered by OpenAI’s Agentic SDK , The OpenAI Agents SDK enables you to build agentic AI apps in a lightweight, easy-to-use package with very few abstractions. It's a production-ready upgrade of our previous experimentation for agents, Swarm.
This application showcases the next generation of autonomous AI systems capable of reasoning, collaborating, and executing complex tasks with human-like adaptability.

Application Features

• Agent Loop 🔄 A built-in loop that autonomously manages tool execution, sends results back to the LLM, and iterates until task completion.
• Python-First 🐍 Leverage native Python syntax (decorators, generators, etc.) to orchestrate and chain agents without external DSLs.
• Handoffs 🤝 Seamlessly coordinate multi-agent workflows by delegating tasks between specialized agents.
• Function Tools ⚒️ Decorate any Python function with @tool to instantly integrate it into the agent’s toolkit.
• Vector Search (RAG) 🧠 Native integration of vector store (IRIS) for RAG retrieval.
• Tracing 🔍 Built-in tracing to visualize, debug, and monitor agent workflows in real time (think LangSmith alternatives).
• MCP Servers 🌐 Support for Model Context Protocol (MCP) via stdio and HTTP, enabling cross-process agent communication.
• Chainlit UI 🖥️ Integrated Chainlit framework for building interactive chat interfaces with minimal code.
• Stateful Memory 🧠 Preserve chat history, context, and agent state across sessions for continuity and long-running tasks.

Hello interface engineers and app developers,     

Did you know that you can use Python in productions or integrations?

The production configuration user interface is low-code, but in many projects, you may reach a point where you need to write some code. As discussed in the Integration Architecture course, business processes are full of places for inserting code, specifically the BPL editor (for BPL business processes) and the DTL editor (for data transformations).

Hi all.

I'm trying to create an indexed table with an vector field so I can search by the vector value.
I've been investigating and found that to get the vector value based on the text (token), use a Python method like the following:

In this article I will be discussing the use of an alternative LLM for generative IA. OpenIA is commonly used, in this article I will show you how to use it and the advantages of using Ollama

In the generative AI usage model that we are used to, we have the following flow:

• we take texts from a data source (a file, for example) and embedding that text into vectors
• we store the vectors in an IRIS database.
• we call an LLM (Large Language Model) that accesses these vectors as context to generate responses in human language.

InterSystems has been at the forefront of database technology since its inception, pioneering innovations that consistently outperform competitors like Oracle, IBM, and Microsoft. By focusing on an efficient kernel design and embracing a no-compromise approach to data performance, InterSystems has carved out a niche in mission-critical applications, ensuring reliability, speed, and scalability.

A History of Technical Excellence

After so many years of waiting, we finally got an official driver available on Pypi

Additionally, found JDBC driver finally available on Maven already for 3 months,  and .Net driver on Nuget more than a month.

 As an author of so many implementations of IRIS support for various Python libraries, I wanted to check it. Implementation of DB-API means that it should be replaceable and at least functions defined in the standard. The only difference should be in SQL.

And the beauty of using already existing libraries, that they already implemented other databases by using DB-API standard, and these libraries already expect how driver should work.

I decided to test InterSystems official driver by implementing its support in SQLAlchemy-iris library.

Introduction

As the health interoperability landscape expands to include data exchange across on-premise as well as hosted solutions, we are seeing an increased need to integrate with services such as cloud storage. One of the most prolifically used and well supported tools is the NoSQL database DynamoDB (Dynamo), provided by Amazon Web Services (AWS).

The challenge for IRIS implementations, as it relates to Dynamo and other modern web services, is that there is no native support for interacting with Dynamo and AWS within ObjectScript. Fortunately, we can utilize ObjectScript’s support for embedded Python to leverage AWS connectivity via Boto3 (Amazon’s Python SDK).

In this tutorial we are going to walk through the steps needed in order to put a single item into the Dynamo table.

Prerequisites and Assumptions

In order to connect to Dynamo you will need to have the following in place:

• An AWS Credentials (~/.aws/credentials) file in your .aws directory. This hidden file may be found in your home directory. (https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/SettingUp.DynamoWebService.html)
• An existing table in Dynamo to which you want to connect.
• The Embedded Python package loaded. (https://docs.intersystems.com/irislatest/csp/docbook/DocBook.UI.Page.cls?KEY=AFL_epython)

This tutorial assumes these are already in place and focuses on the actual connection to DynamoDB and writing to the table using ObjectScript and Embedded Python.

Use Case

Given a Dynamo table, “Employees”, we have the task of adding a single new employee to the table. We are going to pass in their “First_Name”, “Last_Name”, and “Employee_Number”. As we are new to the technology we are going to hard code the values, proving a successful connection and insertion into the table using Boto3.

An Overview

Once credentials have been provided the steps to connect to DynamoDB are as follows:

1. Create a calling ClassMethod in ObjectScript
2. Create a Method that will hold the connection logic using Embedded Python
3. Import boto3 and Create the client
4. Call the put_item method
5. Optional: Return a %Status

There are few other hurdles that you may encounter and the end of this article will also address some of those encountered by the author.

Connecting to DynamoDB

Step 1: Create a calling ClassMethod in ObjectScript

Begin by creating a ClassMethod in ObjectScript. Since we are going to hard code the item being put into the table this is all that is needed.

The purpose of this method is to wrap the Python Method so that the ClassMethod may be called by another upstream operation.

ClassMethod writeToDynamoDB(tableName As %String)

For this use case a TableName is being passed in as a %String. The return value of %Status is needed only when the upstream operation needs a success/failure response.

Step 2: Create a Method that will hold the connection logic using Embedded Python

This Python Method will hold the actual logic for connecting to Dynamo. The method signature is written in ObjectScript but the method body is written in Python and syntax is enforced by Python standards.

ClassMethod writeToDynamoDBpython(tableName As %String) [ Language = python ]

We begin by using ClassMethod as when the code is compiled and run ObjectScript won’t see it when it is called. The same “tableName” will be handed to this method for the Embedded Python to work with so we will use the same variable name here (since these are just Strings the program needs no additional adjustments/transformations). We must also indicate the language to follow using the square bracket notation.

Step 3: Import boto3 and Create the Client

Python syntax requires importing libraries at the outset of your implementations. In this case we will be doing it inside the method body. Importing a Python library is normally done at the top of the file but as this an ObjectScript file doing the import outside of the method body would cause an exception.

ClassMethod writeToDynamoDBpython(tableName As %String) [ Language = python ]
{
    import boto3

    client = boto3.client('dynamodb')
}

Next, the client needs to be created in order to connect to Dynamo. Boto3 provides a straightforward way of creating that client however note that the service required MUST be passed in, e.g. ‘dynamodb’, otherwise it won’t connect to the correct services (other examples are ‘s3’, ‘cloudwatch’, etc.)

For this use case we will create a client that points to a docker container that has been set up to run an image of Dynamo so that we don’t accrue too many calls to AWS:

client = boto3.client(
    service_name='dynamodb', 
    region_name='us-east-1', 
    endpoint_url='http://host.docker.internal:8080'
)

Step 4: Call the put_item method

AWS provides extensive documentation regarding the many fields that may be included in the call to put_item(). In this use case we need only to send it the “First_Name”, “Last_Name”, and “Employee_Number” as follows:

response = client.put_item(
            TableName=tableName,
            Item={
                'First_Name': {
                    'S': 'Vic'
                },
                'Last_Name': {
                    'S': 'Cordova'
                },
                'Employee_Number': {
                     'N': '012345',
                }
)

There are two required fields from the standpoint of AWS: tableName and the item being inserted into the table, e.g. “Item”. When using this implementation of the client the format for the object being inserted takes a key:value form. The name of the object, e.g. ‘First_Name’, indicates the table column name. The following ‘S’ inside of the ‘First_Name’ object indicates to AWS that the associated value is of type String (there are of course other data types that may be noted here, i.e. ‘SS’ for Lists, ‘N’ for Numbers, etc.). This marker is required when using the client.

The complete Embedded Python method:

ClassMethod writeToDynamoDBpython(tableName As %String) [ Language = python ]
{
    import boto3

    client = boto3.client(
        service_name='dynamodb', 
        region_name='us-east-1', 
        endpoint_url='http://host.docker.internal:8080'
    )

    response = client.put_item(
            TableName=tableName,
            Item={
                'First_Name': {
                    'S': 'Vic'
                },
                'Last_Name': {
                    'S': 'Cordova'
                },
                'Employee_Number': {
                     'N': '012345',
                }
    )
}

Step 5: Call the Embedded Python Method from the ObjectScript Method

Now that the logic has been completed we are able to call the Python method from the ObjectScript method thus making the connection possible. This will take in the tablename and run the logic through Python, writing to Dynamo.

ClassMethod writeToDynamoDB(tableName As %String)
{
    DO ..writeToDynamoDBpython(tableName)
}

As this stands it is usable however any upstream callers may require a returned status. In order to accomplish this the following step will suggest some ways to indicate a success/failure to any calling operations.

Step 6: Optional: Return a %Status

In order to pass a success/failure status upstream consider the following edits:

ClassMethod writeToDynamoDB(tableName As %String) As %Status
    {
        #dim tSC As %Status = $$$OK
    
        Set pyStatus = ..writeToDynamoDBpython(tableName)
    
        If (pyStatus '= "OK") {Set tSC = $$$ERROR($$$GeneralError,"failed to write to DynamoDB")}
    
        Quit tSC
    }

ClassMethod writeToDynamoDBpython(ByRef tableName As %String) [ Language = python ]
{
    import boto3

    client = boto3.client(
        service_name='dynamodb', 
        region_name='us-east-1', 
        endpoint_url='http://host.docker.internal:8080'
    )
    
    result = "OK"

    try:
    response = client.put_item(
            TableName=tableName,
            Item={
                'First_Name': {
                    'S': 'Vic'
                },
                'Last_Name': {
                    'S': 'Cordova'
                },
                'Employee_Number': {
                     'N': '012345',
                }
   )
       except:
            result = "BAD"
            
    return result
}

• Add %Status to the ObjectScript method’s definition.
• Set up a status variable (e.g. “tSC”) to hold the $$OK macro.
• Create a status variable in the Embedded Python method using a simple string, “OK”
• Wrap you put_item() in a try/except. The “except:” should update the status variable to something other than “OK” in the case of a failed write.
• Return the status string from the Python method.
• Set up a variable that stores the value returned by the Python method (e.g. “ Set pyStatus = ..writeToDynamoDBpython(tableName)”)
• Check the result of that response in ObjectScript using an if check. In the case of a failure, update the status variable (“tSC”) to be an $$ERROR($$GeneralError) adding an appropriate error message.
• Finally, terminate the method using Quit and return the value of tSC to the upstream caller.

Potential Hurdles and Caveats

This implementation shows a simple example in order to provide the foundation from which an implementer may continue to explore based on their individual use-case and requirements.

Communication between ObjectScript and AWS may prove to be the most tricky as it necessitates an intermediary passing statuses and values back and forth in potentially disparate data types.

Another thing to note is that this implementation is hard-coded. In order to have a more programmatic approach, for example passing in a JSON string, edits like passing the JSON as a %DynamicObject and utilizing the .toJSON() method in ObjectScript in conjunction with leveraging Python’s ability to convert a JSON to a dictionary object using the load() method can be one approach.

Raising exceptions from Embedded Python is also a hurdle as the data types don’t line up. This implementation used Strings to indicate success/failure in conjunction with try/except and if-statements (as noted in the code snippet above) in order to perform some approximation of error handling. It is also worth mentioning that Python exceptions will be printed out, for example if required data is missing, without having to be passed to ObjectScript. The propagation of exceptions is what may require some finagling based on your individual use-case.

Summary/In Conclusion

In summary, establishing a connection to AWS using a Dynamo client, creating a wrapper method in ObjectScript, housing the put_item() logic in a ClassMethod that uses Embedded Python, and making the necessary adjustments per individual requirements makes interacting with Dynamo possible.

Referencing the Documentation will also prove to be very fruitful as there exists sufficient support on both the part of InterSystems and AWS.

Resources

• Intersystems documentation for Embedded Python
• Intersystems course for Embedded Python
• Amazon’s Boto3 documentation
• Amazon’s documentation for the put_item() method

Using SQL Gateway with Python, Vector Search, and Interoperability in InterSystems Iris

Part 3 – REST and Interoperability

Now that we have finished the configuration of the SQL Gateway and we have been able to access the data from the external database via python, and we have set up our vectorized base, we can perform some queries. For this in this part of the article we will use an application developed with CSP, HTML and Javascript that will access an integration in Iris, which then performs the search for data similarity, sends it to LLM and finally returns the generated SQL. The CSP page calls an API in Iris that receives the data to be used in the query, calling the integration. For more information about REST in the Iris see the documentation available at https://docs.intersystems.com/irislatest/csp/docbook/DocBook.UI.Page.cl…

I'm proud to announce the new release of iris-pex-embedded-python (v2.3.1) with a new command line interface.

This command line is called iop for Interoperability On Python.

First I would like to present in few words the project the main changes since the version 1.

A breif history of the project

Version 1.0 was a proof of concept to show how the interoperability framework of IRIS can be used with a python first approach while remaining compatible with any existing ObjectScript code.

What does it mean? It means that any python developer can use the IRIS interoperability framework without any knowledge of ObjectScript.

Example :

from grongier.pex import BusinessOperation

class MyBusinessOperation(BusinessOperation):

    def on_message(self, request):
        self.log.info("Received request")

Great, isn't it?

With version 1.1, I added the possibilty to register those python classes to IRIS with an helper function.

from grongier.pex import Utils

Utils.register_file("/src/MyBusinessOperation.py")

Version 2.0 was a major release because now you can install this project with pip.

pip install iris-pex-embedded-python

What's new in version 2.3.1

Version 2.3.1 is a major release because it introduces a new command line interface.

This command line interface can be used with this python project based on this module or maybe with projects that doesn't use this module.

Let me introduce it and explain why it can be used in non python projects.

The command line interface

The command line is part of this project, to install it you just have to install this project with pip.

pip install iris-pex-embedded-python

Then you can use the command line iop to start the interoperability framework.

iop

output :

usage: iop [-h] [-d DEFAULT] [-l] [-s START] [-k] [-S] [-r] [-M MIGRATE] [-e EXPORT] [-x] [-v] [-L]
optional arguments:
  -h, --help            display help and default production name
  -d DEFAULT, --default DEFAULT
                        set the default production
  -l, --lists           list productions
  -s START, --start START
                        start a production
  -k, --kill            kill a production (force stop)
  -S, --stop            stop a production
  -r, --restart         restart a production
  -M MIGRATE, --migrate MIGRATE
                        migrate production and classes with settings file
  -e EXPORT, --export EXPORT
                        export a production
  -x, --status          status a production
  -v, --version         display version
  -L, --log             display log

default production: UnitTest.Production

Let's go in few examples.

help

The help command display the help and the default production name.

iop -h

output :

usage: python3 -m grongier.pex [-h] [-d DEFAULT] [-l] [-s START] [-k] [-S] [-r] [-M MIGRATE] [-e EXPORT] [-x] [-v] [-L]
...
default production: PEX.Production

default

The default command set the default production.

With no argument, it display the default production.

iop -d

output :

default production: PEX.Production

With an argument, it set the default production.

iop -d PEX.Production

lists

The lists command list productions.

iop -l

output :

{
    "PEX.Production": {
        "Status": "Stopped",
        "LastStartTime": "2023-05-31 11:13:51.000",
        "LastStopTime": "2023-05-31 11:13:54.153",
        "AutoStart": 0
    }
}

start

The start command start a production.

To exit the command, you have to press CTRL+C.

iop -s PEX.Production

If no argument is given, the start command start the default production.

iop -s

output :

2021-08-30 15:13:51.000 [PEX.Production] INFO: Starting production
2021-08-30 15:13:51.000 [PEX.Production] INFO: Starting item Python.FileOperation
2021-08-30 15:13:51.000 [PEX.Production] INFO: Starting item Python.EmailOperation
...

kill

The kill command kill a production (force stop).

Kill command is the same as stop command but with a force stop.

Kill command doesn't take an argument because only one production can be running.

iop -k 

stop

The stop command stop a production.

Stop command doesn't take an argument because only one production can be running.

iop -S 

restart

The restart command restart a production.

Restart command doesn't take an argument because only one production can be running.

iop -r 

migrate

The migrate command migrate a production and classes with settings file.

Migrate command must take the absolute path of the settings file.

Settings file must be in the same folder as the python code.

iop -M /tmp/settings.py

export

The export command export a production.

If no argument is given, the export command export the default production.

iop -e

If an argument is given, the export command export the production given in argument.

iop -e PEX.Production

output :

{
    "Production": {
        "@Name": "PEX.Production",
        "@TestingEnabled": "true",
        "@LogGeneralTraceEvents": "false",
        "Description": "",
        "ActorPoolSize": "2",
        "Item": [
            {
                "@Name": "Python.FileOperation",
                "@Category": "",
                "@ClassName": "Python.FileOperation",
                "@PoolSize": "1",
                "@Enabled": "true",
                "@Foreground": "false",
                "@Comment": "",
                "@LogTraceEvents": "true",
                "@Schedule": "",
                "Setting": [
                    {
                        "@Target": "Adapter",
                        "@Name": "Charset",
                        "#text": "utf-8"
                    },
                    {
                        "@Target": "Adapter",
                        "@Name": "FilePath",
                        "#text": "/irisdev/app/output/"
                    },
                    {
                        "@Target": "Host",
                        "@Name": "%settings",
                        "#text": "path=/irisdev/app/output/"
                    }
                ]
            }
        ]
    }
}

status

The status command status a production.

Status command doesn't take an argument because only one production can be running.

iop -x 

output :

{
    "Production": "PEX.Production",
    "Status": "stopped"
}

Status can be :

• stopped
• running
• suspended
• troubled

version

The version command display the version.

iop -v

output :

2.3.0

log

The log command display the log.

To exit the command, you have to press CTRL+C.

iop -L

output :

2021-08-30 15:13:51.000 [PEX.Production] INFO: Starting production
2021-08-30 15:13:51.000 [PEX.Production] INFO: Starting item Python.FileOperation
2021-08-30 15:13:51.000 [PEX.Production] INFO: Starting item Python.EmailOperation
...

Can it be used outside of iris-pex-embedded-python project ?

That's your choice.

But, before you leave, let me tell why I think it can be used outside of iris-pex-embedded-python project.

First, because it can interact with production without the need to use an iris shell. That means it's easier to use in a script.

Secoud, because settings.py can be used to import production and classes with environment variables.

Here is an example of settings.py :

import os

PRODUCTIONS = [
        {
            'UnitTest.Production': {
                "Item": [
                    {
                        "@Name": "Python.FileOperation",
                        "@ClassName": "Python.FileOperation",
                        "Setting": {
                            "@Target": "Host",
                            "@Name": "%settings",
                            "#text": os.environ['SETTINGS']
                        }
                    }
                ]
            }
        } 
    ]

Pay attention to #text value. It's an environment variable. Neat, isn't it ?

Do you see yourself using this command line tool, worth it to keep developing it ?

Thanks for reading and your feedbacks are welcome.

Introduction

To achieve optimized AI performance, robust explainability, adaptability, and efficiency in healthcare solutions, InterSystems IRIS serves as the core foundation for a project within the x-rAI multi-agentic framework. This article provides an in-depth look at how InterSystems IRIS empowers the development of a real-time health data analytics platform, enabling advanced analytics and actionable insights. The solution leverages the strengths of InterSystems IRIS, including dynamic SQL, native vector search capabilities, distributed caching (ECP), and FHIR interoperability. This innovative approach directly aligns with the contest themes of "Using Dynamic SQL & Embedded SQL," "GenAI, Vector Search," and "FHIR, EHR," showcasing a practical application of InterSystems IRIS in a critical healthcare context.

System Architecture

The Health Agent in x-rAI is built on a modular architecture that integrates multiple components:

Data Ingestion Layer: Fetches real-time health data from wearable devices using the Terra API.

Data Storage Layer: Utilizes InterSystems IRIS for storing and managing structured health data.

Analytics Engine: Leverages InterSystems IRIS's vector search capabilities for similarity analysis and insights generation.

Caching Layer: Implements distributed caching via InterSystems IRIS Enterprise Cache Protocol (ECP) to enhance scalability.

Interoperability Layer: Uses FHIR standards to integrate with external healthcare systems like EHRs.

Below is a high-level architecture diagram:

[Wearable Devices] --> [Terra API] --> [Data Ingestion] --> [InterSystems IRIS] --> [Analytics Engine]
                                                          ------[Caching Layer]------
                                                          ----[FHIR Integration]-----

Technical Implementation

1. Real-Time Data Integration Using Dynamic SQL

The Health Agent ingests real-time health metrics (e.g., heart rate, steps, sleep hours) from wearable devices via the Terra API. This data is stored in InterSystems IRIS using dynamic SQL for flexibility in query generation.

Dynamic SQL Implementation

Dynamic SQL allows the system to adaptively construct queries based on incoming data structures.

def index_health_data_to_iris(data):
    conn = iris_connect()
    if conn is None:
        raise ConnectionError("Failed to connect to InterSystems IRIS.")
    try:
        with conn.cursor() as cursor:
            query = """
                INSERT INTO HealthData (user_id, heart_rate, steps, sleep_hours)
                VALUES (?, ?, ?, ?)
            """
            cursor.execute(query, (
                data['user_id'],
                data['heart_rate'],
                data['steps'],
                data['sleep_hours']
            ))
            conn.commit()
            print("Data successfully indexed into IRIS.")
    except Exception as e:
        print(f"Error indexing health data: {e}")
    finally:
        conn.close()

Benefits of Dynamic SQL

Enables flexible query construction based on incoming data schemas.

Reduces development overhead by avoiding hardcoded queries.

Supports seamless integration of new health metrics without modifying the database schema.

2. Advanced Analytics with Vector Search

InterSystems IRIS’s native vector datatype and similarity functions were utilized to perform vector search on health data. This allowed the system to identify historical records similar to a user’s current health metrics.

Vector Search Workflow

Convert health metrics (e.g., heart rate, steps, sleep hours) into a vector representation.

Store vectors in a dedicated column in the HealthData table.

Perform similarity searches using VECTOR_SIMILARITY().

SQL Query for Vector Search

SELECT TOP 3 user_id, heart_rate, steps, sleep_hours,
       VECTOR_SIMILARITY(vec_data, ?) AS similarity
FROM HealthData
ORDER BY similarity DESC;

Python Integration

def iris_vector_search(query_vector):
    conn = iris_connect()
    if conn is None:
        raise ConnectionError("Failed to connect to InterSystems IRIS.")
    try:
        with conn.cursor() as cursor:
            query_vector_str = ",".join(map(str, query_vector))
            sql = """
                SELECT TOP 3 user_id, heart_rate, steps, sleep_hours,
                       VECTOR_SIMILARITY(vec_data, ?) AS similarity
                FROM HealthData
                ORDER BY similarity DESC;
            """
            cursor.execute(sql, (query_vector_str,))
            results = cursor.fetchall()
            return results
    except Exception as e:
        print(f"Error performing vector search: {e}")
        return []
    finally:
        conn.close()

Benefits of Vector Search

Enables personalized recommendations by identifying historical patterns.

Enhances explainability by linking current metrics to similar past cases.

Optimized for high-speed analytics through SIMD (Single Instruction Multiple Data) operations.

3. Distributed Caching for Scalability

To handle increasing volumes of health data efficiently, the Health Agent leverages InterSystems IRIS’s Enterprise Cache Protocol (ECP). This distributed caching mechanism reduces latency and enhances scalability.

Key Features of ECP

Local caching on application servers minimizes central database queries.

Automatic synchronization ensures consistency across all cache nodes.

Horizontal scaling enables dynamic addition of application servers.

Caching Workflow

Frequently accessed health records are cached locally on application servers.

Subsequent queries for the same records are served directly from the cache.

Updates to cached records trigger automatic synchronization with the central database.

Benefits of Caching

Reduces query response times by serving requests from local caches.

Improves system scalability by distributing workload across multiple nodes.

Minimizes infrastructure costs by reducing central server load.

4. FHIR Integration for Interoperability

InterSystems IRIS’s support for FHIR (Fast Healthcare Interoperability Resources) ensured seamless integration with external healthcare systems like EHRs.

FHIR Workflow Wearable device data is transformed into FHIR-compatible resources (e.g., Observation, Patient).

These resources are stored in InterSystems IRIS and made accessible via RESTful APIs.

External systems can query or update these resources using standard FHIR endpoints.

Benefits of FHIR Integration

Ensures compliance with healthcare interoperability standards.

Facilitates secure exchange of health data between systems.

Enables integration with existing healthcare workflows and applications.

Explainable AI Through Real-Time Insights

By combining InterSystems IRIS’s analytics capabilities with x-rAI’s multi-agentic reasoning framework, the Health Agent generates actionable and explainable insights. For example:

"User 123 had similar metrics (Heart Rate: 70 bpm; Steps: 9,800; Sleep: 7 hrs). Based on historical trends, maintaining your current activity levels is recommended."

This transparency builds trust in AI-driven healthcare applications by providing clear reasoning behind recommendations.

Conclusion The integration of InterSystems IRIS into x-rAI’s Health Agent showcases its potential as a robust platform for building intelligent and explainable AI systems in healthcare. By leveraging features like dynamic SQL, vector search, distributed caching, and FHIR interoperability, this project delivers real-time insights that are both actionable and transparent—paving the way for more reliable AI applications in critical domains like healthcare.

Hey, community! 👋

We are a team of Stanford students applying technology to make sense of climate action. AI excites us because we know we can quickly analyze huge amounts of text.

As we require more reports on sustainability, such as responsibility reports and financial statements, it can be challenging to cut through the noise of aspirations and get to the real action: what are companies doing

That’s why we built a tool to match companies with climate actions scraped from company sustainability reports.

Using embedded Python while building your InterSystems-based solution can add very powerful and deep capabilities to your toolbox.

I'd like to share one sample use-case I encountered - enabling a CDC (Change Data Capture) for a mongoDB Collection - capturing those changes, digesting them through an Interoperability flow, and eventually updating an EMR via a REST API.

In the world of APIs, REST is very extended. But what happens when you need more flexibility in your data-fetching strategies? For instance letting the client to choose what fields is going to receive. Enter GraphQL, a query language for your APIs that provides a flexible alternative to REST.

In this post, we will:

• Compare REST and GraphQL.
• Dive into the basics of GraphQL: Queries, Mutations, and HTTP.
• Build a simple GraphQL server implementation using Graphene, SQLAlchemy, and Flask over data in InterSystems IRIS.
• Explore how to deploy your GraphQL server as a WSGI application in IRIS.

Grab a coffee ☕ and let's go:

REST vs. GraphQL: The Key Differences

Let’s start by comparing REST and GraphQL side by side:

The good news is that both use HTTP so they are pretty easy to consume from client applications.

Getting Started with GraphQL

GraphQL is built around three main concepts:

1. Queries: Retrieve data from your API.

    query {
        allEmployees(hiredAfter: "2020-01-01") {
            name
            hiredon
        }
    }
   

• query keyword indicates a read operation (although it can be omitted)
• allEmployees is the field that represents a list of employees. In the server, a resolver function in the GraphQL schema will fetch the data. It takes (hiredAfter: "2020-01-01") as an argument that is used to filter data.
• name and hiredon are subfields that we are selecting to be returned

2. Mutations: Modify data

    mutation {
        createEmployee(name: "John", lastname: "Doe", position: "Developer", hiredon: "2021-11-30", departmentId: "1") {
            employee {
                name
                lastname
                department {
                    name
                }
            }
        }
    }
   

• mutation indicates an operation that will modify data
• createEmployee is the mutation field that handles the creation of a new employee (a resolver function in the GraphQL schema will handle that). It takes different arguments to define the new employee.
• After creating the new employee, the mutation will return an object that includes details about the newly created employee (this is defined by employee and structure containing name, lastname and department with some nested fields).

3. HTTP: Works over POST/GET requests to a single endpoint, typically /graphql.

For example, the previous query could be sent as:

curl -X POST \
  -H "Content-Type: application/json" \
  -d '{"query": "query { allEmployees(hiredAfter: \"2020-01-01\") { name hiredon } }"}' \
  http://localhost:5000/graphql

And the response data could be:

{"data":{"allEmployees":[{"name":"Alice","hiredon":"2020-06-15"},{"name":"Clara","hiredon":"2021-09-01"},{"name":"Eve","hiredon":"2022-01-05"},{"name":"Frank","hiredon":"2020-11-15"},{"name":"John","hiredon":"2021-11-30"},{"name":"Phileas","hiredon":"2023-11-27"}]}}

Building a GraphQL Server with InterSystems IRIS

We will use this tools ⚒️:

• Graphene: To define GraphQL schemas and queries.
• SQLAlchemy: For database interaction.
• Flask: As the web framework.
• InterSystems IRIS: Our data platform where we will storage data.

📝 You will find detailed instructions on setting up InterSystems IRIS Community and a Python virtual environment in iris-graphql-demo.

Now let's focus on the main steps you need to do.

Step 1: Define the Data Model

We will use a simple model with Employee and Department tables that will be created and populated in InterSystems IRIS.

Create a models.py file with:

from sqlalchemy import *
from sqlalchemy.orm import relationship, declarative_base, Session, scoped_session, sessionmaker, relationship, backref

engine = create_engine('iris://superuser:SYS@localhost:1972/USER')
db_session = scoped_session(sessionmaker(bind=engine))

Base = declarative_base()

Base.query = db_session.query_property()

class Department(Base):
    __tablename__ = 'departments'
    __table_args__ = {'schema': 'dc_graphql'}

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    description = Column(String)

    employees = relationship('Employee', back_populates='department')

class Employee(Base):
    __tablename__ = 'employees'
    __table_args__ = {'schema': 'dc_graphql'}

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    lastname = Column(String, nullable=False)
    hiredon = Column(Date)
    position = Column(String)
    department_id = Column(Integer, ForeignKey('dc_graphql.departments.id'))

    department = relationship('Department', back_populates='employees')

Step 2: Create the GraphQL Schema

Let's define the GraphQL Schema using Graphene using the models we have just defined.

In the related OpenExchange application you will find a version that also implements a mutation.

Create a schema.py file with this content:

import graphene
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from models import db_session, Employee, Department

class DepartmentType(SQLAlchemyObjectType):
    class Meta:
        model = Department
        interfaces = ()

class EmployeeType(SQLAlchemyObjectType):
    class Meta:
        model = Employee
        interfaces = ()

class Query(graphene.ObjectType):
    all_departments = graphene.List(DepartmentType)
    all_employees = graphene.List(
        EmployeeType,
        hired_after=graphene.Date(),
        position=graphene.String()
    )

    def resolve_all_departments(self, info):
        return db_session.query(Department).all()

    def resolve_all_employees(self, info, hired_after=None, position=None):
        query = db_session.query(Employee)
        if hired_after:
            query = query.filter(Employee.hiredon > hired_after)
        if position:
            query = query.filter(Employee.position == position)
        return query.all()

schema = graphene.Schema(query=Query)

Step 3: Set Up Flask

And we will need also a Flask application that sets up the /graphql endpoint.

Create an app.py file like this:

from flask import Flask, request, jsonify
from flask_graphql import GraphQLView
from schema import schema

app = Flask(__name__)

@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view(
        'graphql',
        schema=schema,
        graphiql=True,
    )
)

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

Step 4: Create some data

We could create the tables and insert some data using SQL or ObjectScript.

But, if you are exploring the capabilities of Python + InterSystems IRIS I recommend you to use Python 🐍 directly.

Open an interactive Python session:

python

And then:

from models import db_session, engine, Base, Department, Employee

# this will create the tables for you in InterSytems IRIS
Base.metadata.create_all(bind=engine)

# add departments
engineering = Department(name='Engineering', description='Handles product development and technology')
hr = Department(name='Human Resources', description='Manages employee well-being and recruitment')
sales = Department(name='Sales', description='Responsible for sales and customer relationships')
db_session.add_all([engineering, hr, sales])

# add employees
employees = [
    Employee(name='Alice', lastname='Smith', hiredon=date(2020, 6, 15), position='Software Engineer', department=engineering),
    Employee(name='Bob', lastname='Brown', hiredon=date(2019, 3, 10), position='QA Engineer', department=engineering),
    Employee(name='Clara', lastname='Johnson', hiredon=date(2021, 9, 1), position='Recruiter', department=hr),
    Employee(name='David', lastname='Davis', hiredon=date(2018, 7, 22), position='HR Manager', department=hr),
    Employee(name='Eve', lastname='Wilson', hiredon=date(2022, 1, 5), position='Sales Executive', department=sales),
    Employee(name='Frank', lastname='Taylor', hiredon=date(2020, 11, 15), position='Account Manager', department=sales)
]
db_session.add_all(employees)
db_session.commit()

Step 5: Test your server

First, you will need to run your server:

python app.py

Then you can start testing your server using an included GraphiQL UI in http://localhost:5000/graphql.

GraphQL Best Practices

If you need to work with very large datasets or you need pagination, consider using Relay. Relay introduces Edge and Node concepts for efficient pagination and handling of large datasets.

In the related OpenExchange application you will also find a GraphQL schema implementation using Relay.

Deploying Your WSGI Application in InterSystems IRIS

In case you don't know what WGSI is, just check this article WSGI Support Introduction and the related posts that explain how to deploy Flask, Django and FastAPI applications in IRIS by @Guillaume Rongier

In a nutshell, InterSystems IRIS can deploy WGSI applications now so you can deploy your GraphQL server.

You only need to set up a web application like this:

After that, your endpoint will be http://localhost:52773/graphqlserver/graphql

Conclusion

Congratulations! You’ve built a GraphQL server on top of InterSystems IRIS, explored its basic features, and even deployed it as a WSGI application. This demonstrates how flexible API approaches can integrate seamlessly with enterprise-grade data platforms like InterSystems IRIS.

If you are still awake, thanks for reading 😄! I'll give you a final fun fact:

• Did you know that GraphQL was developed at Facebook in 2012 and open-sourced in 2015? Now it powers sites like GitHub, Shopify, and Twitter.

I have a flask application, its working locally. I also have a iris for health 2024.3 docker container front-ended with the iris nginx container.

I can configure the WSGI application with iris for health, give it a "url" (/flask or /csp/flask) but I cannot access the flask application. It looks like the url is not found within the nginx configuration. Is there any documentation or suggestion for configuration/enabling a flask app with a IRIS container front ended with the nginx container (provided by intersystems container registry) 

Hi Community,

We're pleased to invite you to the upcoming webinar in Hebrew:

👉 Hebrew Webinar: GenAI + RAG - Leveraging Intersystems IRIS as your Vector DB👈

📅 Date & time: February 26th, 3:00 PM IDT