feat: solution for '8. Самописный поисковый агент на основе deep agents from scratch'

src/index.py

@@ -0,0 +1,139 @@
+import os
+import asyncio
+from typing import Any
+
+import requests
+from dotenv import load_dotenv
+from langchain.chat_models import ChatOpenAI
+from langchain.agents import initialize_agent, AgentType
+from langchain.memory import ConversationBufferMemory
+from langchain.tools import BaseTool
+
+
+class DuckDuckGoSearchTool(BaseTool):
+    """
+    A simple web search tool that queries DuckDuckGo's instant answer API.
+    """
+
+    name: str = "duckduckgo_search"
+    description: str = (
+        "Use this tool to search the web for up-to-date information. "
+        "Input should be a search query."
+    )
+
+    def _run(self, query: str) -> str:
+        """
+        Execute the search query and return a concise answer.
+
+        Parameters
+        ----------
+        query : str
+            The search query string.
+
+        Returns
+        -------
+        str
+            A short answer extracted from the search results.
+        """
+        if not query:
+            return "No query provided."
+
+        url = "https://api.duckduckgo.com/"
+        params = {
+            "q": query,
+            "format": "json",
+            "no_html": 1,
+            "skip_disambig": 1,
+        }
+        try:
+            response = requests.get(url, params=params, timeout=10)
+            response.raise_for_status()
+            data = response.json()
+        except Exception as exc:
+            return f"Error during search: {exc}"
+
+        # Prefer abstract text if available
+        abstract = data.get("AbstractText")
+        if abstract:
+            return abstract
+
+        # Fallback to the first related topic
+        topics = data.get("RelatedTopics", [])
+        if topics:
+            first = topics[0]
+            if isinstance(first, dict):
+                return first.get("Text", "No relevant information found.")
+        return "No relevant information found."
+
+    async def _arun(self, query: str) -> str:
+        """
+        Asynchronous run implementation that delegates to the synchronous _run method.
+        """
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._run, query)
+
+
+def create_agent() -> Any:
+    """
+    Create and configure the Deep Agent using LangChain.
+
+    Returns
+    -------
+    Any
+        The initialized agent executor.
+    """
+    # Load environment variables (e.g., OPENAI_API_KEY)
+    load_dotenv()
+
+    # Initialize the LLM
+    llm = ChatOpenAI(temperature=0)
+
+    # Memory to keep conversation context
+    memory = ConversationBufferMemory(memory_key="chat_history")
+
+    # Instantiate the custom search tool
+    search_tool = DuckDuckGoSearchTool()
+
+    # Initialize the agent with the REACT description template
+    agent = initialize_agent(
+        tools=[search_tool],
+        llm=llm,
+        agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
+        memory=memory,
+        verbose=True,
+    )
+    return agent
+
+
+def main() -> None:
+    """
+    Simple CLI to interact with the Deep Agent.
+    """
+    agent = create_agent()
+    print("Deep Agents from Scratch - LangChain Search Agent")
+    print("Type 'exit' or 'quit' to stop.\n")
+
+    while True:
+        try:
+            query = input("Enter your question: ").strip()
+        except (EOFError, KeyboardInterrupt):
+            print("\nExiting.")
+            break
+
+        if query.lower() in {"exit", "quit"}:
+            print("Goodbye!")
+            break
+
+        if not query:
+            print("Please enter a non-empty query.")
+            continue
+
+        try:
+            result = agent.run(query)
+            print("\nAnswer:\n", result)
+        except Exception as exc:
+            print(f"Error: {exc}")
+
+
+if __name__ == "__main__":
+    main()