エージェンティック AI RAG アプリケーションの構築方法: 手順ガイド

コミュニティの皆さん、こんにちは。
従来のキーワードベースの検索では、ニュアンスのあるドメイン固有のクエリには対応できません。 ベクトル検索であれば、セマンティック認識を利用して、キーワードだけでなくコンテキストにも基づいたレスポンスを AI エージェントで検索して生成することができます。
この記事では、エージェンティック AI RAG（検索拡張生成）アプリケーションを作成手順を紹介します。

実装手順:

1. エージェントツールを作成する
   • インジェスト機能の追加: ドキュメント（例: InterSystems IRIS 2025.1 リリースノート）を自動的にインジェストしてインデックス作成を行います。
   • ベクトル検索機能の実装
2. ベクトル検索エージェントを作成する
3. Triage（メインエージェント）に渡す
4. エージェントを実行する

1. エージェントツールを作成する
1.1 - ドキュメントの取り込みを実装します。ドキュメントの取り込みとインデックス作成を自動化します。

インジェストツールは以下のコードで実装します。

defingestDoc(self):#Check if document is defined, by selecting from table#If not defined then INGEST document, Otherwise back
        embeddings = OpenAIEmbeddings()	
        #Load the document based on the fle type
        loader = TextLoader("/irisdev/app/docs/IRIS2025-1-Release-Notes.txt", encoding='utf-8')      
    documents = loader.load()        
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=<span class="hljs-number">400</span>, chunk_overlap=<span class="hljs-number">0</span>)
    
    texts = text_splitter.split_documents(documents)
                   
    <span class="hljs-comment">#COLLECTION_NAME = "rag_document"</span>
    db = IRISVector.from_documents(
        embedding=embeddings,
        documents=texts,
        collection_name = self.COLLECTION_NAME,
        connection_string=self.CONNECTION_STRING,
    )

    db = IRISVector.from_documents(embedding=embeddings,documents=texts, collection_name = self.COLLECTION_NAME, connection_string=self.CONNECTION_STRING,)</code></pre>
ベクトル検索エージェントは、指定されたリポジトリフォルダから「New in InterSystems IRIS 2025.1」を IRIS Vector Store に自動的に取り込んでインデックスを作成します。この操作はそのデータがまだ存在しない場合にのみ実行されます。

以下のクエリを実行して、ベクトルストアから必要なデータを取得します。
SELECTid, embedding, document, metadata
FROM SQLUser.AgenticAIRAG

1.2 - ベクトル検索機能の実装
以下のコードはエージェントの検索機能を実装します。
defragSearch(self,prompt):#Check if collections are defined or ingested done.# if not then call ingest method
        embeddings = OpenAIEmbeddings()	
        db2 = IRISVector (
            embedding_function=embeddings,    
            collection_name=self.COLLECTION_NAME,
            connection_string=self.CONNECTION_STRING,
        )
        docs_with_score = db2.similarity_search_with_score(prompt)
        relevant_docs = ["".join(str(doc.page_content)) + " "for doc, _ in docs_with_score]
    <span class="hljs-comment">#Generate Template</span>
    template = <span class="hljs-string">f"""
    Prompt: <span class="hljs-subst">{prompt}</span>
    Relevant Docuemnts: <span class="hljs-subst">{relevant_docs}</span>
    """</span>
    <span class="hljs-keyword">return</span> template</code></pre>

Triage エージェントは、受信したユーザークエリを処理し、それを Vector Search Agent に委任します。このエージェントは、セマンティック検索を実行して、最も関連性の高い情報を取得します。

2 - ベクトルストアエージェントを作成する
以下のコードは、以下の要素を含む vector_search_agent を実装します。
エージェントを調整するためのカスタム handoff_descriptions
明確な演算 instructions
iris_RAG_search ツール（ドキュメントの取り込みとベクトル検索操作に irisRAG.py を使用）
@function_tool    @cl.step(name = "Vector Search Agent (RAG)", type="tool", show_input = False)asyncdefiris_RAG_search():"""Provide IRIS Release Notes details,IRIS 2025.1 Release Notes, IRIS Latest Release Notes, Release Notes"""ifnot ragOprRef.check_VS_Table():
                 #Ingest the document first
                 msg = cl.user_session.get("ragclmsg")
                 msg.content = "Ingesting Vector Data..."await msg.update()
                 ragOprRef.ingestDoc()
        <span class="hljs-keyword">if</span> ragOprRef.check_VS_Table():
             msg = cl.user_session.get(<span class="hljs-string">"ragclmsg"</span>)
             msg.content = <span class="hljs-string">"Searching Vector Data..."</span>
             <span class="hljs-keyword">await</span> msg.update()                 
             <span class="hljs-keyword">return</span> ragOprRef.ragSearch(cl.user_session.get(<span class="hljs-string">"ragmsg"</span>))   
        <span class="hljs-keyword">else</span>:
             <span class="hljs-keyword">return</span> <span class="hljs-string">"Error while getting RAG data"</span>
vector_search_agent = Agent(
        name=<span class="hljs-string">"RAGAgent"</span>,
        handoff_description=<span class="hljs-string">"Specialist agent for Release Notes"</span>,
        instructions=<span class="hljs-string">"You provide assistance with Release Notes. Explain important events and context clearly."</span>,
        tools=[iris_RAG_search]
)</code></pre>

3 - Triage（メインエージェント）に渡す
以下のコードは、処理済みのクエリを Triage エージェント（メインコーディネーター）に渡すためのハンドオフプロトコルを実装します。
 triage_agent = Agent(
        name="Triage agent",
        instructions=(
            "Handoff to appropriate agent based on user query.""if they ask about Release Notes, handoff to the vector_search_agent.""If they ask about production, handoff to the production agent.""If they ask about dashboard, handoff to the dashboard agent.""If they ask about process, handoff to the processes agent.""use the WebSearchAgent tool to find information related to the user's query and do not use this agent is query is about Release Notes.""If they ask about order, handoff to the order_agent."
        ),
        handoffs=[vector_search_agent,production_agent,dashboard_agent,processes_agent,order_agent,web_search_agent]
    )

4 - エージェントを実行する
以下のコードは次の内容を実行します。
ユーザー入力を受け取る
triage_agent  を呼び出す
クエリを Vector_Search_Agent に送信して処理する
@cl.on_message
async def main(message: cl.Message):
    """Process incoming messages and generate responses."""
    # Send a thinking message
    msg = cl.Message(content="Thinking...")
    await msg.send()
agent: Agent = cast(Agent, cl.user_session.get(<span class="hljs-string">"agent"</span>))
config: RunConfig = cast(RunConfig, cl.user_session.get(<span class="hljs-string">"config"</span>))

# Retrieve the chat history from the session.
history = cl.user_session.get(<span class="hljs-string">"chat_history"</span>) or []

# Append the user'<span class="hljs-keyword">s</span> message to the history.
history.append({<span class="hljs-string">"role"</span>: <span class="hljs-string">"user"</span>, <span class="hljs-string">"content"</span>: message.content})

# Used by RAG agent
cl.user_session.<span class="hljs-keyword">set</span>(<span class="hljs-string">"ragmsg"</span>, message.content)
cl.user_session.<span class="hljs-keyword">set</span>(<span class="hljs-string">"ragclmsg"</span>, msg)

<span class="hljs-keyword">try</span>:
    <span class="hljs-keyword">print</span>(<span class="hljs-string">"\n[CALLING_AGENT_WITH_CONTEXT]\n"</span>, history, <span class="hljs-string">"\n"</span>)
    result = Runner.run_sync(agent, history, run_config=config)
           
    response_content = result.final_output
    
    # Update the thinking message with the actual response
    msg.content = response_content
    await msg.update()

    # Append the assistant'<span class="hljs-keyword">s</span> response to the history.
    history.append({<span class="hljs-string">"role"</span>: <span class="hljs-string">"developer"</span>, <span class="hljs-string">"content"</span>: response_content})
    # NOTE: Here we are appending the response to the history <span class="hljs-keyword">as</span> a developer message.
    # This is a BUG in the agents library.
    # The expected behavior is to append the response to the history <span class="hljs-keyword">as</span> an assistant message.

    # Update the session with the <span class="hljs-keyword">new</span> history.
    cl.user_session.<span class="hljs-keyword">set</span>(<span class="hljs-string">"chat_history"</span>, history)
    
    # Optional: Log the interaction
    <span class="hljs-keyword">print</span>(f<span class="hljs-string">"User: {message.content}"</span>)
    <span class="hljs-keyword">print</span>(f<span class="hljs-string">"Assistant: {response_content}"</span>)
    
except Exception <span class="hljs-keyword">as</span> e:
    msg.content = f<span class="hljs-string">"Error: {str(e)}"</span>
    await msg.update()
    <span class="hljs-keyword">print</span>(f<span class="hljs-string">"Error: {str(e)}"</span>)</code></pre>

実際の動作をご覧ください
詳細については、iris-AgenticAI の Open Excahnge アプリケーションページをご覧ください。
以上です