Guide pas par pas : Configuration de RAG pour les agents d'IA Gen en utilisant la base de données vectorielle IRIS en Python

Comment configurer le RAG pour les agents OpenAI à l'aide d'InterSystems IRIS Vector DB en Python

Dans cet article, je vous présenterai un exemple d'utilisation d'InterSystems IRIS Vector DB pour stocker des représentations vectorielles et les intégrer à un agent OpenAI.

Pour cette démonstration, nous créerons un agent OpenAI au moyen de connaissances sur la technologie InterSystems. Pour ce faire, nous stockerons les représentations de certains documents InterSystems dans IRIS, puis, à l'aide de la recherche vectorielle IRIS, nous récupérerons le contenu pertinent, ce qui permettra de mettre en place un processus de Retrieval-Augmented Generation (RAG)

Note: Remarque : vous trouverez dans la section 1 des détails sur la manière de transformer du texte en embeddings. Si vous ne vous intéressez qu'à la recherche vectorielle IRIS, vous pouvez passer directement à la section 2.

Section 1: embedding des données

La qualité de vos embeddings dépend de celle de vos données ! Pour obtenir les meilleurs résultats, vous devez préparer vos données avec soin. Voici les étapes à suivre :

• Nettoyage du texte (suppression des caractères spéciaux ou des espaces superflus)
• Segmentation des données en blocs plus petits
• Autres techniques de prétraitement

Dans cet exemple, la documentation est stockée dans de simples fichiers texte qui ne nécessitent qu'un nettoyage minimal. Cependant, nous allons diviser le texte en blocs afin de permettre un RAG plus efficace et plus précis.

Étape 1 : Segmentation des fichiers texte

La segmentation du texte en blocs gérables présente deux avantages pour les systèmes RAG :

1. Une recherche plus précise : les embeddings représentent des sections de texte plus petites et plus spécifiques.
2. Une recherche plus efficace : la réduction de la quantité de texte traitée par requête diminue les coûts et améliore les performances.

Dans cet exemple, les blocs du texte segmenté seront stockés dans des fichiers Parquet avant d’être téléchargés vers IRIS (mais vous pouvez utiliser n’importe quelle méthode, y compris le téléchargement direct).

Fonction de segmentation

Nous utiliserons RecursiveCharacterTextSplitter de langchain_text_splitters pour segmenter le texte de manière stratégique en fonction des limites des paragraphes, des phrases et des mots.

• Taille du bloc : 300 jetons (des segments plus grands offrent plus de contexte mais augmentent le coût de la recherche)
• Chevauchement de blocs : 50 jetons (permet de préserver le contexte entre les blocs)

from langchain_text_splitters import RecursiveCharacterTextSplitter

def chunk_text_by_tokens(text: str, chunk_size: int, chunk_overlap: int) -> list[str]:
"""
Chunk text prioritizing paragraph and sentence boundaries using
RecursiveCharacterTextSplitter. Returns a list of chunk strings.
"""
splitter = RecursiveCharacterTextSplitter(
# En priorité, les unités sémantiques les plus grandes, puis les plus petites
separators=["\n\n", "\n", ". ", " ", ""],
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
is_separator_regex=False,
)
return splitter.split_text(text)
 

Nous allons ensuite utiliser la fonction de segmentation pour traiter un fichier texte à la fois et appliquons un encodeur tiktoken afin de calculer le nombre de jetons et de générer des métadonnées. Ces métadonnées nous seront utiles ultérieurement pour créer des représentations vectorielles et les stocker dans IRIS.

from pathlib import Path

import tiktoken

def chunk_file(path: Path, chunk_size: int, chunk_overlap: int, encoding_name: str = "cl100k_base") -> list[dict]:
"""
Read a file, split its contents into token-aware chunks, and return metadata for each chunk.
Returns a list of dicts with keys:
- filename
- relative_path
- absolute_path
- chunk_index
- chunk_text
- token_count
- modified_time
- size_bytes
"""
p = Path(path)
if not p.exists() or not p.is_file():
raise FileNotFoundError(f"File not found: {path}")
try:
text = p.read_text(encoding="utf-8", errors="replace")
except Exception as e:
raise RuntimeError(f"Failed to read file {p}: {e}")
# Préparation du tokenizer pour un comptage précis des jetons
try:
encoding = tiktoken.get_encoding(encoding_name)
except Exception as e:
raise ValueError(f"Invalid encoding name '{encoding_name}': {e}")
# Création des blocs à l'aide du « chunker » fourni
chunks = chunk_text_by_tokens(text, chunk_size, chunk_overlap)
# Métadonnées du fichier
stat = p.stat()
from datetime import datetime, timezone
modified_time = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat()
absolute_path = str(p.resolve())
try:
relative_path = str(p.resolve().relative_to(Path.cwd()))
except Exception:
relative_path = p.name
# Création de lignes
rows: list[dict] = []
for idx, chunk in enumerate(chunks):
token_count = len(encoding.encode(chunk))
rows.append({
"filename": p.name,
"relative_path": relative_path,
"absolute_path": absolute_path,
"chunk_index": idx,
"chunk_text": chunk,
"token_count": token_count,
"modified_time": modified_time,
"size_bytes": stat.st_size,
})
return rows
 

Étape 2 : Création d'embeddings

Vous pouvez générer des embeddings à l'aide de fournisseurs de services cloud (par exemple, OpenAI) ou de modèles locaux via Ollama (par exemple, nomic-embed-text). Dans cet exemple, nous utiliserons le modèle « text-embedding-3-small » d'OpenAI pour incorporer chaque bloc et enregistrer les résultats au format Parquet en vue d'une intégration ultérieure dans la base de données IRIS Vector DB.

from openai import OpenAI

import pandas as pd

def embed_and_save_parquet(input_parquet_path: str, output_parquet_path: str):
"""
Loads a Parquet file, creates embeddings for the 'chunk_text' column using
OpenAI's small embedding model, and saves the result to a new Parquet file.
Args:
input_parquet_path (str): Path to the input Parquet file containing 'chunk_text'.
output_parquet_path (str): Path to save the new Parquet file with embeddings.
openai_api_key (str): Your OpenAI API key.
"""
key = os.getenv("OPENAI_API_KEY")
if not key:
print("ERROR: OPENAI_API_KEY environment variable is not set.", file=sys.stderr)
sys.exit(1)
try:
# Téléchargement du fichier Parquet
df = pd.read_parquet(input_parquet_path)
# Initiation du client OpenAI
client = OpenAI(api_key=key)
# Generation des embeddings pour chaque chunk_text
embeddings = []
for text in df['chunk_text']:
response = client.embeddings.create(
input=text,
model="text-embedding-3-small"  # Using the small embedding model
)
embeddings.append(response.data[0].embedding)
# Ajout des embeddings au DataFrame
df['embedding'] = embeddings
# Enregistrement du nouveau DataFrame dans un fichier Parquet
df.to_parquet(output_parquet_path, index=False)
print(f"Embeddings generated and saved to {output_parquet_path}")
except FileNotFoundError:
print(f"Error: Input file not found at {input_parquet_path}")
except KeyError:
print("Error: 'chunk_text' column not found in the input Parquet file.")
except Exception as e:
print(f"An unexpected error occurred: {e}")

Étape 3 : Regroupement du traitement des données

Maintenant, il est temps d'exécuter le pipeline. Dans cet exemple, nous allons charger et segmenter la documentation de Business Service, générer des vecteurs d'encodage et enregistrer les résultats au format Parquet en vue de leur ingestion dans IRIS.

CHUNK_SIZE_TOKENS = 300

CHUNK_OVERLAP_TOKENS = 50

ENCODING_NAME="cl100k_base"

current_file_path = Path(file).resolve()
load_documentation_to_parquet(input_dir=current_file_path.parent / <span class="mention">"Documentation"</span> / <span class="mention">"BusinessService"</span>, 
                              output_file=current_file_path.parent / <span class="mention">"BusinessService.parquet"</span>, 
                              chunk_size=CHUNK_SIZE_TOKENS, 
                              chunk_overlap=CHUNK_OVERLAP_TOKENS, 
                              encoding_name=ENCODING_NAME)
embed_and_save_parquet(input_parquet_path=current_file_path.parent / <span class="mention">"BusinessService.parquet"</span>, 
                        output_parquet_path=current_file_path.parent / <span class="mention">"BusinessService_embedded.parquet"</span>)

Une ligne de notre fichier Parquet final pour le service métier ressemblera à ceci :

{"filename":"FileInboundAdapters.txt","relative_path":"Documentation\BusinessService\Adapters\FileInboundAdapters.txt","absolute_path":"C:\Users\…\Documentation\BusinessService\Adapters\FileInboundAdapters.txt","chunk_index":0,"chunk_text":"Settings for the File Inbound Adapter\nProvides reference information for settings of the file inbound adapter, EnsLib.File.InboundAdapterOpens in a new tab. You can configure these settings after you have added a business service that uses this adapter to your production.\nSummary","token_count":52,"modified_time":"2025-11-25T18:34:16.120336+00:00","size_bytes":13316,"embedding":[-0.02851865254342556,0.01860344596207142,…,0.0135544464207155]}

Section 2 : Utilisation de la recherche vectorielle IRIS

Étape 4 : Téléchargement de vos embeddings vers IRIS

Sélectionnez l'espace de noms IRIS et le nom de la table que vous utiliserez pour stocker les embeddings. (Le script ci-dessous créera la table si elle n'existe pas encore.) Utilisez ensuite le pilote Python DB-API d'InterSystems IRIS pour insérer les blocs et leurs embeddings.

La fonction ci-dessous permet de lire un fichier Parquet contenant le texte segmenté et les embeddings, de normaliser la colonne d’embedding en une liste de nombres flottants sérialisable au format JSON, de se connecter à IRIS, de créer la table de destination si elle n’existe pas (au moyen d’une colonne VECTOR(FLOAT, 1536), où 1536 correspond au nombre de dimensions dans l’embedding) puis insère chaque ligne à l'aide de TO_VECTOR(?) dans une instruction SQL paramétrée. Elle confirme la transaction en cas de succès, consigne la progression et nettoie la connexion, tout en annulant les modifications en cas d'erreurs de base de données.

import iris  # Le pilote Python DB-API d'InterSystems IRIS 

import pandas as pd

import numpy as np

import json

from pathlib import Path
# --- Configuration ---

PARQUET_FILE_PATH = "your_embeddings.parquet"

IRIS_HOST = "localhost"

IRIS_PORT = 8881

IRIS_NAMESPACE = "VECTOR"

IRIS_USERNAME = "superuser"

IRIS_PASSWORD = "sys"

TABLE_NAME = "AIDemo.Embeddings" # D'être conforme à la table créée dans IRIS

EMBEDDING_DIMENSIONS = 1536 # D'être conformes aux dimensions des embeddings utilisés

def upload_embeddings_to_iris(parquet_path: str):

"""

Reads a Parquet file with 'chunk_text' and 'embedding' columns

and uploads them to an InterSystems IRIS vector database table.

"""

# 1. Téléchargement des données du fichier Parquet à l'aide de pandas

try:

df = pd.read_parquet(parquet_path)

if 'chunk_text' not in df.columns or 'embedding' not in df.columns:

print("Error: Parquet file must contain 'chunk_text' and 'embedding' columns.")

return

except FileNotFoundError:

print(f"Error: The file at {parquet_path} was not found.")

return

# Les embeddings doivent être dans un format compatible avec la fonction TO_VECTOR (liste de flottants)

# Parquet enregistre souvent les tables numpy sous forme de listes

if isinstance(df['embedding'].iloc[0], np.ndarray):

df['embedding'] = df['embedding'].apply(lambda x: x.tolist())

print(f"Loaded {len(df)} records from {parquet_path}.")

# 2. Établissement d'une connexion à InterSystems IRIS

connection = None

try:

conn_string = f"{IRIS_HOST}:{IRIS_PORT}/{IRIS_NAMESPACE}"

connection = iris.connect(conn_string, IRIS_USERNAME, IRIS_PASSWORD)

cursor = connection.cursor()

print("Successfully connected to InterSystems IRIS.")

# Création de la table d'embedding si elle n'existe pas

cursor.execute(f"""

CREATE TABLE IF NOT EXISTS  {TABLE_NAME} (

ID INTEGER IDENTITY PRIMARY KEY,

chunk_text VARCHAR(2500), embedding VECTOR(FLOAT, {EMBEDDING_DIMENSIONS})

)"""

)

# 3. Préparation de l'instruction de SQL INSERT

# InterSystems IRIS utilise la fonction TO_VECTOR à l'aide de SQL pour insérer des données vectorielles

insert_sql = f"""

INSERT INTO {TABLE_NAME} (chunk_text, embedding)

VALUES (?, TO_VECTOR(?))

"""

# 4. Itération et insertion des données

count = 0

for index, row in df.iterrows():

text = row['chunk_text']

# Conversion de la liste de flottants en une chaîne JSON, ce qui est requis par la fonction TO_VECTOR lors de l'utilisation de la DB-API

vector_json_str = json.dumps(row['embedding'])
        cursor.execute(insert_sql, (text, vector_json_str))
        count += <span class="mention">1</span>
        <span class="mention">if</span> count % <span class="mention">100</span> == <span class="mention">0</span>:
            print(<span class="mention">f"Inserted </span><span class="mention">{count}</span> rows...")
    
    <span class="mention"># Validation de la transaction</span>
    connection.commit()
    print(<span class="mention">f"Data upload complete. Total rows inserted: </span><span class="mention">{count}</span>.")
<span class="mention">except</span> iris.DBAPIError <span class="mention">as</span> e:
    print(<span class="mention">f"A database error occurred: </span><span class="mention">{e}</span>")
    <span class="mention">if</span> connection:
        connection.rollback()
<span class="mention">except</span> Exception <span class="mention">as</span> e:
    print(<span class="mention">f"An unexpected error occurred: </span><span class="mention">{e}</span>")
<span class="mention">finally</span>:
    <span class="mention">if</span> connection:
        connection.close()
        print(<span class="mention">"Database connection closed."</span>)

Exemple d'utilisation :

current_file_path = Path(file).resolve()

upload_embeddings_to_iris(current_file_path.parent / "BusinessService_embedded.parquet")

Étape 5 : Création de votre fonctionnalité de recherche incorporée

Nous créerons ensuite une fonction de recherche qui intègre la requête de l’utilisateur, effectue une recherche par similarité vectorielle dans InterSystems IRIS via l’API Python DB‑et renvoie les meilleurs‑résultats (top-k) de notre table d’embeddings.

La fonction d’exemple ci-dessous lit un fichier Parquet contenant des segments de texte et leurs embeddings correspondants, puis charge ces données dans la table de stockage vectoriel d’InterSystems IRIS. Elle valide d'abord le fichier Parquet et normalise le format d'encodage en une chaîne de type tableau JSON compatible avec la fonction TO_VECTOR d'IRIS. Après avoir établi une connexion à IRIS, la fonction crée la table cible si elle n'existe pas, prépare une instruction SQL INSERT paramétrée et parcourt chaque ligne pour insérer le texte du segment et son encodage. Enfin, elle confirme la transaction, consigne la progression et assure une gestion correcte des erreurs ainsi que le nettoyage de la connexion à la base de données.

import iris

from typing import List

import os

from openai import OpenAI
# --- Configuration ---

PARQUET_FILE_PATH = "your_embeddings.parquet"

IRIS_HOST = "localhost"

IRIS_PORT = 8881

IRIS_NAMESPACE = "VECTOR"

IRIS_USERNAME = "superuser"

IRIS_PASSWORD = "sys"

TABLE_NAME = "AIDemo.Embeddings" # D'être conforme à la table créée dans IRIS

EMBEDDING_DIMENSIONS = 1536

MODEL = "text-embedding-3-small"

def get_embedding(text: str, model: str, client) -> List[float]:

# Normalisation des sauts de ligne et conversion en chaîne de caractères

payload = [("" if text is None else str(text)).replace("\n", " ") for _ in range(1)]

resp = client.embeddings.create(model=model, input=payload, encoding_format="float")

return resp.data[0].embedding

def search_embeddings(search: str, top_k: int):

print("-------RAG--------")

print(f"Searching IRIS vector store for: ", search)

key = os.getenv("OPENAI_API_KEY")

client = OpenAI(api_key=key)

# 2. Établissement d'une connexion à InterSystems IRIS

connection = None

try:

conn_string = f"{IRIS_HOST}:{IRIS_PORT}/{IRIS_NAMESPACE}"

connection = iris.connect(conn_string, IRIS_USERNAME, IRIS_PASSWORD)

cursor = connection.cursor()

print("Successfully connected to InterSystems IRIS.")

# Intégration d'une requête de recherche

#emb_raw = str(test_embedding) # FOR TESTING

emb_raw = get_embedding(search, model=MODEL, client=client)

emb_raw = str(emb_raw)

#print("EMB_RAW:", emb_raw)

emb_values = []

for x in emb_raw.replace('[', '').replace(']', '').split(','):

try:

emb_values.append(str(float(x.strip())))

except ValueError:

continue

emb_str = ", ".join(emb_values)

# Préparation de l'instruction SQL SELECT

search_sql = f"""

SELECT TOP {top_k} ID, chunk_text FROM {TABLE_NAME}

ORDER BY VECTOR_DOT_PRODUCT((embedding), TO_VECTOR(('{emb_str}'), FLOAT)) DESC

"""

cursor.execute(search_sql)

results = []

row = cursor.fetchone()

while row is not None:

results.append(row[:])

row = cursor.fetchone()
<span class="mention">except</span> iris.DBAPIError <span class="mention">as</span> e:
    print(<span class="mention">f"A database error occurred: </span><span class="mention">{e}</span>")
    <span class="mention">if</span> connection:
        connection.rollback()
<span class="mention">except</span> Exception <span class="mention">as</span> e:
    print(<span class="mention">f"An unexpected error occurred: </span><span class="mention">{e}</span>")
<span class="mention">finally</span>:
    <span class="mention">if</span> connection:
        connection.close()
        print(<span class="mention">"Database connection closed."</span>)
    print(<span class="mention">"------------RAG Finished-------------"</span>)
    <span class="mention">return</span> results

Étape 6 : Ajout du contexte RAG à votre agent

Maintenant que vous avez:

• segmenté et intégré votre documentation,
• téléchargé les embeddings vers IRIS et créé un index vectoriel,
• développé une fonction de recherche pour les requêtes vectorielles IRIS,

il est temps de rassembler le tout dans un chat interactif de type « Retrieval-Augmented Generation » (RAG) à l'aide de l'API OpenAI Responses. Dans cet exemple, nous donnerons à l'agent un accès direct à la fonction de recherche (pour un contrôle plus précis de l'agent), mais cela peut également être réalisé en utilisant une bibliothèque telle que langchain.

Tout d'abord, vous devrez créer vos instructions pour l'agent, en veillant à lui donner accès à la fonction de recherche :

import os

---------------------------- Configuration ----------------------------

MODEL = os.getenv("OPENAI_RESPONSES_MODEL", "gpt-5-nano")

SYSTEM_INSTRUCTIONS = (

"You are a helpful assistant that answers questions about InterSystems "

"business services and related integration capabilities. You have access "

"to a vector database of documentation chunks about business services. "

"\n\n"

"Use the search_business_docs tool whenever the user asks about specific "

"settings, configuration options, or how to perform tasks with business "

"services. Ground your answers in the retrieved context, quoting or "

"summarizing relevant chunks. If nothing relevant is found, say so "

"clearly and answer from your general knowledge with a disclaimer."

)

---------------------------- Définition des outils ----------------------------

TOOLS = [
{
"type": "function",
"name": "search_business_docs",
"description": (
"Searches a vector database of documentation chunks related to "
"business services and returns the most relevant snippets."
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": (
"Natural language search query describing what you want "
"to know about business services."
),
},
"top_k": {
"type": "integer",
"description": (
"Maximum number of results to retrieve from the vector DB."
),
"minimum": 1,
"maximum": 10,
},
},
"required": ["query", "top_k"],
"additionalProperties": False,
},
"strict": True,
}
]
 

Nous avons maintenant besoin d'une petite méthode de « routage » pour permettre au modèle d'utiliser notre outil RAG.

La fonction call_rag_tool(name, args) reçoit un appel de fonction émis par l’API OpenAI Responses et le redirige vers notre implémentation locale (l’outil search_business_docs qui encapsule Search.search_embeddings). Elle prend la requête du modèle et top_k, exécute la recherche vectorielle IRIS, puis renvoie une charge utile encodée en JSON‑contenant les meilleurs résultats (identifiants et snippets). Ce JSON converti en chaîne de caractères est important car l’API Responses attend que les sorties des outils soient sous forme de chaînes de caractères ; en formatant les résultats de manière prévisible, nous permettons au modèle d’ancrer facilement sa réponse finale dans la documentation récupérée. Si un nom d’outil inconnu est demandé, la fonction renvoie une charge utile d’erreur afin que le modèle puisse la gérer de manière appropriée.

def call_rag_tool(name: str, args: Dict[str, Any]) -> str:

"""Route function calls from the model to our local Python implementations.

Currently only supports the search_business_docs tool, which wraps

Search.search_embeddings.

The return value must be a string. We will JSON-encode a small structure

so the model can consume the results reliably.

"""

if name == "search_business_docs":

query = args.get("query", "")

top_k = args.get("top_k", "")

results = search_embeddings(query, top_k)

# Chaque ligne devrait se présenter sous la forme suivante (ID, chunk_text)

formatted: List[Dict[str, Any]] = []

for row in results:

if not row:

continue

# Il convient de prévoir une solution de secours si la longueur/structure des lignes est modifiée

doc_id = row[0] if len(row) > 0 else None

text = row[1] if len(row) > 1 else None

formatted.append({"id": doc_id, "text": text})

payload = {"query": query, "results": formatted}

return json.dumps(payload, ensure_ascii=False)

# Outil inconnu ; renvoie un message d'erreur structuré

return json.dumps({"error": f"Unknown tool name: {name}"})

Maintenant que nous avons notre outil RAG, nous pouvons commencer à travailler sur la logique de la boucle de chat. Tout d’abord, il nous faut un assistant pour extraire de manière fiable la réponse finale du modèle ainsi que les résultats des outils à partir de l’API OpenAI Responses. La fonction extract_answer_and_sources(response) parcourt les éléments response.output contenant les résultats du modèle et les concatène en une seule chaîne de réponse. Elle collecte également les charges utiles function_call_output (le JSON renvoyé par notre outil RAG), les analyse et les expose sous la forme de tool_context pour plus de transparence et de facilité de débogage. La fonction analyse la sortie du modèle en une structure compacte : {« answer »: ..., « tool_context »: [...]}.

def extract_answer_and_sources(response: Any) -> Dict[str, Any]:

"""Extract a structured answer and optional sources from a Responses API object.

We don't enforce a global JSON response schema here. Instead, we:

- Prefer the SDK's output_text convenience when present

- Fall back to concatenating any output_text content parts

- Also surface any tool-call-output payloads we got back this turn as

tool_context for debugging/inspection.

"""

answer_text = ""

# Solution privilégiée : facilité d'utilisation du SDK

if hasattr(response, "output_text") and response.output_text:

answer_text = response.output_text

else:

# Solution de secours : parcourir les éléments de sortie

parts: List[str] = []

for item in getattr(response, "output", []) or []:

if getattr(item, "type", None) == "message":

for c in getattr(item, "content", []) or []:

if getattr(c, "type", None) == "output_text":

parts.append(getattr(c, "text", ""))

answer_text = "".join(parts)

# Récupération de tous les éléments de type function_call_output pour les rendre visibles

tool_context: List[Dict[str, Any]] = []

for item in getattr(response, "output", []) or []:

if getattr(item, "type", None) == "function_call_output":

try:

tool_context.append({

"call_id": getattr(item, "call_id", None),

"output": json.loads(getattr(item, "output", "")),

})

except Exception:

tool_context.append({

"call_id": getattr(item, "call_id", None),

"output": getattr(item, "output", ""),

})

return {"answer": answer_text.strip(), "tool_context": tool_context}

Avec extract_answer_and_sources, nous pouvons construire la boucle de chat complète pour orchestrer une conversation en deux‑phases, impliquant l'appel d'outils‑, avec l'API OpenAI Responses. La fonction chat_loop() exécute une interface CLI interactive : elle recueille la question de l'utilisateur, envoie une première requête contenant des instructions système et l'outil search_business_docs, puis examine tous les éléments de type function_call émis par le modèle. Pour chaque appel de fonction, elle exécute notre outil RAG local (call_rag_tool, qui encapsule search_embeddings) et ajoute le résultat à la conversation sous la forme d’un élément function_call_output. Elle effectue ensuite une deuxième requête demandant au modèle d’utiliser ces résultats d’outils à l’aide de ceux-ci pour produire une réponse étayée, analyse cette réponse via extract_answer_and_sources, puis l’affiche. La boucle conserve le contexte d'exécution dans input_items afin que chaque tour puisse s'appuyer sur les messages précédents et les résultats des outils.

def chat_loop() -> None:

"""Run an interactive CLI chat loop using the OpenAI Responses API.

The loop supports multi-step tool-calling:

- First call may return one or more function_call items

- We execute those locally (e.g., call search_embeddings)

- We send the tool outputs back in a second responses.create call

- Then we print the model's final, grounded answer

"""

key = os.getenv("OPENAI_API_KEY")

if not key:

raise RuntimeError("OPENAI_API_KEY is not set in the environment.")

client = OpenAI(api_key=key)

print("\nBusiness Service RAG Chat")

print("Type 'exit' or 'quit' to stop.\n")

# Liste récapitulative des entrées (messages + appels d'outils + sorties d'outils) pour le contexte

input_items: List[Dict[str, Any]] = []

while True:

user_input = input("You: ").strip()

if not user_input:

continue

if user_input.lower() in {"exit", "quit"}:

print("Goodbye.")

break

# Ajout d'un message d'utilisateur

input_items.append({"role": "user", "content": user_input})

# Premier appel : le modèle décide s'il convient d'appeler des outils

response = client.responses.create(

model=MODEL,

instructions=SYSTEM_INSTRUCTIONS,

tools=TOOLS,

input=input_items,

)

# Les éléments de sortie du modèle sont enregistrés dans notre conversation en cours

input_items += response.output

# 2) Exécution des appels de fonction

# L'API Responses renvoie des éléments function_call dans response.output.

for item in response.output:

if getattr(item, "type", None) != "function_call":

continue

name = getattr(item, "name", None)

raw_args = getattr(item, "arguments", "{}")

try:

args = json.loads(raw_args) if isinstance(raw_args, str) else raw_args

except json.JSONDecodeError:

args = {"query": user_input}

result_str = call_rag_tool(name, args or {})

# Le résultat de l'outil est ajouté en tant que function_call_output

input_items.append(

{

"type": "function_call_output",

"call_id": getattr(item, "call_id", None),

"output": result_str,

}

)

# 3) Deuxième appel : le modèle est invité à répondre en utilisant les résultats des outils

followup = client.responses.create(

model=MODEL,

instructions=(

SYSTEM_INSTRUCTIONS

+ "\n\nYou have just received outputs from your tools. "

+ "Use them to give a concise, well-structured answer."

),

tools=TOOLS,

input=input_items,

)

structured = extract_answer_and_sources(followup)

print("Agent:\n" + structured["answer"] + "\n")

Ça y est ! Vous venez de créer un pipeline RAG complet propulsé par IRIS Vector Search. Bien que cet exemple se soit concentré sur un cas d'utilisation simple, IRIS Vector Search ouvre la voie à de nombreuses autres possibilités:

• Base de connaissances pour des agents de service client plus complexes
• Stockage du contexte conversationnel pour des agents hyper-personnalisés 
• Détection d'anomalies dans les données textuelles
• Analyse par regroupement pour les données textuelles

J'espère que ce guide vous a fourni un point de départ solide pour explorer la recherche vectorielle et créer vos propres applications basées sur l'IA avec InterSystems IRIS !

Le code source complet est disponible ici :

• https://openexchange.intersystems.com/portal/products/IRISVectorSearchRAGExample
• https://github.com/isc-epolakie/IRISVectorSearchRAGExample