feat: solution for 'Агент с RAG-памятью'

src/cli.py

@@ -0,0 +1,75 @@
+import shlex
+import sys
+from typing import List
+
+from .tools import add_numbers, search_item
+
+def run_cli() -> None:
+    """
+    Interactive command line interface that supports:
+    /add <int> <int>   - Adds two numbers.
+    /search <query>   - Searches a predefined list for the query.
+    /quit             - Exits the program.
+    """
+    memory: List[str] = [
+        "Python programming",
+        "LangChain framework",
+        "Artificial Intelligence",
+        "Machine Learning",
+        "Data Science",
+    ]
+
+    print("Welcome to the RAG Agent CLI!")
+    print("Available commands:")
+    print("  /add <int> <int>   - Add two numbers.")
+    print("  /search <query>   - Search items in memory.")
+    print("  /quit             - Exit the program.\n")
+
+    while True:
+        try:
+            user_input = input(">> ").strip()
+        except (EOFError, KeyboardInterrupt):
+            print("\nExiting.")
+            break
+
+        if not user_input:
+            continue
+
+        if user_input.lower() == "/quit":
+            print("Goodbye!")
+            break
+
+        if user_input.lower().startswith("/add"):
+            try:
+                parts = shlex.split(user_input)
+                if len(parts) != 3:
+                    raise ValueError
+                a = int(parts[1])
+                b = int(parts[2])
+                result = add_numbers(a=a, b=b)
+                print(f"Result: {result}")
+            except ValueError:
+                print("Usage: /add <int> <int>")
+            continue
+
+        if user_input.lower().startswith("/search"):
+            try:
+                parts = shlex.split(user_input)
+                if len(parts) < 2:
+                    raise ValueError
+                query = " ".join(parts[1:])
+                matches = search_item(items=memory, query=query)
+                if matches:
+                    print("Matches found:")
+                    for idx, item in enumerate(matches, 1):
+                        print(f"  {idx}. {item}")
+                else:
+                    print("No matches found.")
+            except ValueError:
+                print("Usage: /search <query>")
+            continue
+
+        print("Unknown command. Please use /add, /search, or /quit.")
+
+if __name__ == "__main__":
+    run_cli()

src/main.py

@@ -1,53 +1,7 @@
-import os
-from dotenv import load_dotenv
-from langchain_ollama import Ollama
-from langchain_community.embeddings import OllamaEmbeddings
-from langchain.vectorstores import Qdrant
-from qdrant_client import QdrantClient
-from src.agent import RAGAgent
-from src.chunk_document import chunk_document
+from .cli import run_cli
 
 def main() -> None:
-    # Load environment variables if any
-    load_dotenv()
-
-    # Initialize LLM and embeddings
-    llm = Ollama(model="llama3")
-    embeddings = OllamaEmbeddings(model="llama3")
-
-    # Connect to Qdrant (assumes Qdrant is running locally on port 6333)
-    qdrant_client = QdrantClient(host="localhost", port=6333)
-    vector_store = Qdrant(
-        client=qdrant_client,
-        collection_name="rag_collection",
-        embeddings=embeddings,
-    )
-
-    # Create the RAG agent
-    rag_agent = RAGAgent(llm=llm, vector_store=vector_store, chunk_document_func=chunk_document)
-
-    # Example documents to add to the vector store
-    sample_docs = [
-        "LangChain is a framework for developing applications powered by language models.",
-        "Qdrant is a vector database that can store embeddings and perform similarity search.",
-        "Ollama provides a lightweight interface to run LLMs locally.",
-    ]
-    rag_agent.add_documents(sample_docs)
-
-    # Build the agent executor
-    agent_executor = rag_agent.create_agent()
-
-    print("RAG Agent is ready. Type your question (or 'exit' to quit).")
-    while True:
-        user_input = input(">>> ")
-        if user_input.lower() in {"exit", "quit"}:
-            print("Goodbye!")
-            break
-        try:
-            response = agent_executor.invoke({"input": user_input})
-            print(response["output"])
-        except Exception as e:
-            print(f"Error: {e}")
+    run_cli()
 
 if __name__ == "__main__":
     main()

src/tools.py

@@ -0,0 +1,41 @@
+from langchain.tools import tool
+from typing import List
+
+@tool
+def add_numbers(a: int, b: int) -> int:
+    """
+    Add two numbers and return the sum.
+
+    Parameters
+    ----------
+    a : int
+        The first number.
+    b : int
+        The second number.
+
+    Returns
+    -------
+    int
+        The sum of a and b.
+    """
+    return a + b
+
+@tool
+def search_item(items: List[str], query: str) -> List[str]:
+    """
+    Search for items containing the query string (case-insensitive).
+
+    Parameters
+    ----------
+    items : List[str]
+        The list of items to search.
+    query : str
+        The search query.
+
+    Returns
+    -------
+    List[str]
+        A list of items that contain the query string.
+    """
+    query_lower = query.lower()
+    return [item for item in items if query_lower in item.lower()]