Merge pull request #1660 from oracle-devrel/lsa-custom-rag-agent2

added custom_rag_agent demo to genai-agents

Author: anshoracle

ai/gen-ai-agents/custom_rag_agent/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 Luigi Saetta
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

ai/gen-ai-agents/custom_rag_agent/README.md

@@ -0,0 +1,37 @@
+![UI](images/ui_image.png)
+
+# Custom RAG agent
+This repository contains the code for the development of a **custom RAG Agent**, based on OCI Generative AI, Oracle 23AI DB and **LangGraph**
+
+## Design and implementation
+* The agent is implemented using **LangGraph**
+* Vector Search is implemented, using Langchain, on top of Oracle 23AI
+* A **reranker** can be used to refine the search
+
+Design decisions:
+* For every node of the graph there is a dedicated Python class (a **Runnable**, as QueryRewriter...)
+* Reranker is implemented using a LLM. As other option, it is easy to plug-in, for example, Cohere reranker
+* The agent is integrated with **OCI APM**, for **Observability**; Integration using **py-zipkin**
+* UI implemented using **Streamlit**
+
+Streaming:
+* Support for streaming events from the agent: as soon as a step is completed (Vector Search, Reranking, ...) the UI is updated.
+For example, links to the documentation' chunks are displayed before the final answer is ready.
+* Streaming of the final answer.
+
+## Status
+It is **wip**.
+
+## References
+* [Integration with OCI APM](https://luigi-saetta.medium.com/enhancing-observability-in-rag-solutions-with-oracle-cloud-6f93b2675f40)
+
+## Advantages of the Agentic approach
+One of the primary advantages of the agentic approach is its modularity. 
+Customer requirements often surpass the simplicity of typical Retrieval-Augmented Generation (RAG) demonstrations. Implementing a framework like **LangGraph** necessitates organizing code into a modular sequence of steps, facilitating the seamless integration of additional features at appropriate places.​
+
+For example, to ensure that final responses do not disclose Personally Identifiable Information (PII) present in the knowledge base, one can simply append a node at the end of the graph. This node would process the generated answer, detect any PII, and anonymize it accordingly.
+
+## Configuration
+* use Python 3.11
+* use the requirements.txt
+* create your config_private.py using the template provided

ai/gen-ai-agents/custom_rag_agent/agent_state.py

@@ -0,0 +1,52 @@
+"""
+File name: agent_state.py
+Author: Luigi Saetta
+Date last modified: 2025-03-31
+Python Version: 3.11
+
+Description:
+    This module defines the class that handles the agent's State
+
+
+Usage:
+    Import this module into other scripts to use its functions.
+    Example:
+        from agent_state import State
+
+License:
+    This code is released under the MIT License.
+
+Notes:
+    This is a part of a demo showing how to implement an advanced
+    RAG solution as a LangGraph agent.
+
+Warnings:
+    This module is in development, may change in future versions.
+"""
+
+from typing_extensions import TypedDict, Optional
+
+
+class State(TypedDict):
+    """
+    The state of the graph
+    """
+
+    # the original user request
+    user_request: str
+    chat_history: list = []
+
+    # the question reformulated using chat_history
+    standalone_question: str = ""
+
+    # similarity_search
+    retriever_docs: Optional[list] = []
+    # reranker
+    reranker_docs: Optional[list] = []
+    # Answer
+    final_answer: str
+    # Citations
+    citations: list = []
+
+    # if any step encounter an error
+    error: Optional[str] = None

ai/gen-ai-agents/custom_rag_agent/answer_generator.py

@@ -0,0 +1,135 @@
+"""
+File name: answer_generator.py
+Author: Luigi Saetta
+Date last modified: 2025-03-31
+Python Version: 3.11
+
+Description:
+    This module implements the last step in the workflow: generation
+    of the answer form the LLM:
+
+
+Usage:
+    Import this module into other scripts to use its functions.
+    Example:
+        from answer_generator import AnswerGenerator
+
+License:
+    This code is released under the MIT License.
+
+Notes:
+    This is a part of a demo showing how to implement an advanced
+    RAG solution as a LangGraph agent.
+
+Warnings:
+    This module is in development, may change in future versions.
+"""
+
+from langchain_core.runnables import Runnable
+from langchain_core.messages import SystemMessage, HumanMessage
+from langchain.prompts import PromptTemplate
+
+# integration with APM
+from py_zipkin.zipkin import zipkin_span
+
+from agent_state import State
+from oci_models import get_llm
+from prompts import (
+    ANSWER_PROMPT_TEMPLATE,
+)
+
+from utils import get_console_logger
+from config import AGENT_NAME, DEBUG
+
+logger = get_console_logger()
+
+
+class AnswerGenerator(Runnable):
+    """
+    Takes the user request and the chat history and rewrite the user query
+    in a standalone question that is used for the semantic search
+    """
+
+    def __init__(self):
+        """
+        Init
+        """
+        self.dict_languages = {
+            "en": "English",
+            "fr": "French",
+            "it": "Italian",
+            "es": "Spanish",
+        }
+
+    def build_context_for_llm(self, docs: list):
+        """
+        Build the context for the final answer from LLM
+
+        docs: list[Documents]
+        """
+        _context = ""
+
+        for doc in docs:
+            _context += doc.page_content + "\n\n"
+
+        return _context
+
+    @zipkin_span(service_name=AGENT_NAME, span_name="answer_generation")
+    def invoke(self, input: State, config=None, **kwargs):
+        """
+        Generate the final answer
+        """
+        # get the config
+        model_id = config["configurable"]["model_id"]
+
+        if config["configurable"]["main_language"] in self.dict_languages:
+            # want to change language
+            main_language = self.dict_languages.get(
+                config["configurable"]["main_language"]
+            )
+        else:
+            # "same as the question" (default)
+            # answer will be in the same language as the question
+            main_language = None
+
+        if DEBUG:
+            logger.info("AnswerGenerator, model_id: %s", model_id)
+            logger.info("AnswerGenerator, main_language: %s", main_language)
+
+        final_answer = ""
+        error = None
+
+        try:
+            llm = get_llm(model_id=model_id)
+
+            _context = self.build_context_for_llm(input["reranker_docs"])
+
+            system_prompt = PromptTemplate(
+                input_variables=["context"],
+                template=ANSWER_PROMPT_TEMPLATE,
+            ).format(context=_context)
+
+            messages = [
+                SystemMessage(content=system_prompt),
+            ]
+            # add the chat history
+            for msg in input["chat_history"]:
+                messages.append(msg)
+
+            # to force the answer in the selected language
+            if main_language is not None:
+                the_question = f"{input['user_request']}. Answer in {main_language}."
+            else:
+                # no cross language
+                the_question = input["user_request"]
+
+            messages.append(HumanMessage(content=the_question))
+
+            # here we invoke the LLM and we return the generator
+            final_answer = llm.stream(messages)
+
+        except Exception as e:
+            logger.error("Error in generate_answer: %s", e)
+            error = str(e)
+
+        return {"final_answer": final_answer, "error": error}