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

コミュニティの皆さん、こんにちは。

従来のキーワードベースの検索では、ニュアンスのあるドメイン固有のクエリには対応できません。 ベクトル検索であれば、セマンティック認識を利用して、キーワードだけでなくコンテキストにも基づいたレスポンスを AI エージェントで検索して生成することができます。

この記事では、エージェンティック AI RAG（検索拡張生成）アプリケーションを作成手順を紹介します。

実装手順:

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

1. エージェントツールを作成する

1.1 - ドキュメントの取り込みを実装します。ドキュメントの取り込みとインデックス作成を自動化します。

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

    def ingestDoc(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=400, chunk_overlap=0)
        
        texts = text_splitter.split_documents(documents)
                       
        #COLLECTION_NAME = "rag_document"
        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,)

ベクトル検索エージェントは、指定されたリポジトリフォルダから「New in InterSystems IRIS 2025.1」を IRIS Vector Store に自動的に取り込んでインデックスを作成します。この操作はそのデータがまだ存在しない場合にのみ実行されます。

以下のクエリを実行して、ベクトルストアから必要なデータを取得します。

SELECT
id, embedding, document, metadata
FROM SQLUser.AgenticAIRAG

1.2 - ベクトル検索機能の実装

以下のコードはエージェントの検索機能を実装します。

 def ragSearch(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]
        
        #Generate Template
        template = f"""
        Prompt: {prompt}
        Relevant Docuemnts: {relevant_docs}
        """
        return template

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)
    async def iris_RAG_search():
            """Provide IRIS Release Notes details,IRIS 2025.1 Release Notes, IRIS Latest Release Notes, Release Notes"""
            if not ragOprRef.check_VS_Table():
                 #Ingest the document first
                 msg = cl.user_session.get("ragclmsg")
                 msg.content = "Ingesting Vector Data..."
                 await msg.update()
                 ragOprRef.ingestDoc()
            
            if ragOprRef.check_VS_Table():
                 msg = cl.user_session.get("ragclmsg")
                 msg.content = "Searching Vector Data..."
                 await msg.update()                 
                 return ragOprRef.ragSearch(cl.user_session.get("ragmsg"))   
            else:
                 return "Error while getting RAG data"
    vector_search_agent = Agent(
            name="RAGAgent",
            handoff_description="Specialist agent for Release Notes",
            instructions="You provide assistance with Release Notes. Explain important events and context clearly.",
            tools=[iris_RAG_search]
    )

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 - エージェントを実行する

以下のコードは次の内容を実行します。

1. ユーザー入力を受け取る
2. triage_agent  を呼び出す
3. クエリを 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("agent"))
    config: RunConfig = cast(RunConfig, cl.user_session.get("config"))

    # Retrieve the chat history from the session.
    history = cl.user_session.get("chat_history") or []
    
    # Append the user's message to the history.
    history.append({"role": "user", "content": message.content})
    
    # Used by RAG agent
    cl.user_session.set("ragmsg", message.content)
    cl.user_session.set("ragclmsg", msg)

    try:
        print("\n[CALLING_AGENT_WITH_CONTEXT]\n", history, "\n")
        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's response to the history.
        history.append({"role": "developer", "content": response_content})
        # NOTE: Here we are appending the response to the history as a developer message.
        # This is a BUG in the agents library.
        # The expected behavior is to append the response to the history as an assistant message.
    
        # Update the session with the new history.
        cl.user_session.set("chat_history", history)
        
        # Optional: Log the interaction
        print(f"User: {message.content}")
        print(f"Assistant: {response_content}")
        
    except Exception as e:
        msg.content = f"Error: {str(e)}"
        await msg.update()
        print(f"Error: {str(e)}")

実際の動作をご覧ください

詳細については、iris-AgenticAI の Open Excahnge アプリケーションページをご覧ください。

以上です