Создайте интеллектуальный интерфейс агента AI с несколькими инструментами, используя стримку для беспроблемного взаимодействия в реальном времени

!pip install -q streamlit langchain langchain-google-genai langchain-community
!pip install -q pyngrok python-dotenv wikipedia duckduckgo-search
!npm install -g localtunnel


import streamlit as st
import os
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool, WikipediaQueryRun, DuckDuckGoSearchRun
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import PromptTemplate
from langchain.callbacks.streamlit import StreamlitCallbackHandler
from langchain_community.utilities import WikipediaAPIWrapper, DuckDuckGoSearchAPIWrapper
import asyncio
import threading
import time
from datetime import datetime
import json

GOOGLE_API_KEY = "Use Your API Key Here" 
NGROK_AUTH_TOKEN = "Use Your Auth Token Here" 
os.environ("GOOGLE_API_KEY") = GOOGLE_API_KEY

class InnovativeAgentTools:
   """Advanced tool collection for the multi-agent system"""
  
   @staticmethod
   def get_calculator_tool():
       def calculate(expression: str) -> str:
           """Calculate mathematical expressions safely"""
           try:
               allowed_chars = set('0123456789+-*/.() ')
               if all(c in allowed_chars for c in expression):
                   result = eval(expression)
                   return f"Result: {result}"
               else:
                   return "Error: Invalid mathematical expression"
           except Exception as e:
               return f"Calculation error: {str(e)}"
      
       return Tool(
           name="Calculator",
           func=calculate,
           description="Calculate mathematical expressions. Input should be a valid math expression."
       )
  
   @staticmethod
   def get_memory_tool(memory_store):
       def save_memory(key_value: str) -> str:
           """Save information to memory"""
           try:
               key, value = key_value.split(":", 1)
               memory_store(key.strip()) = value.strip()
               return f"Saved '{key.strip()}' to memory"
           except:
               return "Error: Use format 'key: value'"
      
       def recall_memory(key: str) -> str:
           """Recall information from memory"""
           return memory_store.get(key.strip(), f"No memory found for '{key}'")
      
       return (
           Tool(name="SaveMemory", func=save_memory,
                description="Save information to memory. Format: 'key: value'"),
           Tool(name="RecallMemory", func=recall_memory,
                description="Recall saved information. Input: key to recall")
       )
  
   @staticmethod
   def get_datetime_tool():
       def get_current_datetime(format_type: str = "full") -> str:
           """Get current date and time"""
           now = datetime.now()
           if format_type == "date":
               return now.strftime("%Y-%m-%d")
           elif format_type == "time":
               return now.strftime("%H:%M:%S")
           else:
               return now.strftime("%Y-%m-%d %H:%M:%S")
      
       return Tool(
           name="DateTime",
           func=get_current_datetime,
           description="Get current date/time. Options: 'date', 'time', or 'full'"
       )

class MultiAgentSystem:
   """Innovative multi-agent system with specialized capabilities"""
  
   def __init__(self, api_key: str):
       self.llm = ChatGoogleGenerativeAI(
           model="gemini-pro",
           google_api_key=api_key,
           temperature=0.7,
           convert_system_message_to_human=True
       )
       self.memory_store = {}
       self.conversation_memory = ConversationBufferWindowMemory(
           memory_key="chat_history",
           k=10,
           return_messages=True
       )
       self.tools = self._initialize_tools()
       self.agent = self._create_agent()
  
   def _initialize_tools(self):
       """Initialize all available tools"""
       tools = ()
      
       tools.extend((
           DuckDuckGoSearchRun(api_wrapper=DuckDuckGoSearchAPIWrapper()),
           WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
       ))
      
       tools.append(InnovativeAgentTools.get_calculator_tool())
       tools.append(InnovativeAgentTools.get_datetime_tool())
       tools.extend(InnovativeAgentTools.get_memory_tool(self.memory_store))
      
       return tools
  
   def _create_agent(self):
       """Create the ReAct agent with advanced prompt"""
       prompt = PromptTemplate.from_template("""
🤖 You are an advanced AI assistant with access to multiple tools and persistent memory.


AVAILABLE TOOLS:
{tools}


TOOL USAGE FORMAT:
- Think step by step about what you need to do
- Use Action: tool_name
- Use Action Input: your input
- Wait for Observation
- Continue until you have a final answer


MEMORY CAPABILITIES:
- You can save important information using SaveMemory
- You can recall previous information using RecallMemory
- Always try to remember user preferences and context


CONVERSATION HISTORY:
{chat_history}


CURRENT QUESTION: {input}


REASONING PROCESS:
{agent_scratchpad}


Begin your response with your thought process, then take action if needed.
""")
      
       agent = create_react_agent(self.llm, self.tools, prompt)
       return AgentExecutor(
           agent=agent,
           tools=self.tools,
           memory=self.conversation_memory,
           verbose=True,
           handle_parsing_errors=True,
           max_iterations=5
       )
  
   def chat(self, message: str, callback_handler=None):
       """Process user message and return response"""
       try:
           if callback_handler:
               response = self.agent.invoke(
                   {"input": message},
                   {"callbacks": (callback_handler)}
               )
           else:
               response = self.agent.invoke({"input": message})
           return response("output")
       except Exception as e:
           return f"Error processing request: {str(e)}"

def create_streamlit_app():
   """Create the innovative Streamlit application"""
  
   st.set_page_config(
       page_title="🚀 Advanced LangChain Agent with Gemini",
       page_icon="🤖",
       layout="wide",
       initial_sidebar_state="expanded"
   )
  
   st.markdown("""
   
   """, unsafe_allow_html=True)
  
   st.markdown("""
   

       
       
Powered by LangChain + Gemini API + Streamlit
   
   """, unsafe_allow_html=True)
  
   with st.sidebar:
       st.header("🔧 Configuration")
      
       api_key = st.text_input(
           "🔑 Google AI API Key",
           type="password",
           value=GOOGLE_API_KEY if GOOGLE_API_KEY != "your-gemini-api-key-here" else "",
           help="Get your API key from https://ai.google.dev/"
       )
      
       if not api_key:
           st.error("Please enter your Google AI API key to continue")
           st.stop()
      
       st.success("✅ API Key configured")
      
       st.header("🤖 Agent Capabilities")
       st.markdown("""
       - 🔍 **Web Search** (DuckDuckGo)
       - 📚 **Wikipedia Lookup**
       - 🧮 **Mathematical Calculator**
       - 🧠 **Persistent Memory**
       - 📅 **Date & Time**
       - 💬 **Conversation History**
       """)
      
       if 'agent_system' in st.session_state:
           st.header("🧠 Memory Store")
           memory = st.session_state.agent_system.memory_store
           if memory:
               for key, value in memory.items():
                   st.markdown(f"""
                   

                       {key}: {value}
                   
                   """, unsafe_allow_html=True)
           else:
               st.info("No memories stored yet")
  
   if 'agent_system' not in st.session_state:
       with st.spinner("🔄 Initializing Advanced Agent System..."):
           st.session_state.agent_system = MultiAgentSystem(api_key)
       st.success("✅ Agent System Ready!")
  
   st.header("💬 Interactive Chat")
  
   if 'messages' not in st.session_state:
       st.session_state.messages = ({
           "role": "assistant",
           "content": """🤖 Hello! I'm your advanced AI assistant powered by Gemini. I can:


• Search the web and Wikipedia for information
• Perform mathematical calculations
• Remember important information across our conversation
• Provide current date and time
• Maintain conversation context


Try asking me something like:
- "Calculate 15 * 8 + 32"
- "Search for recent news about AI"
- "Remember that my favorite color is blue"
- "What's the current time?"
"""
       })
  
   for message in st.session_state.messages:
       with st.chat_message(message("role")):
           st.markdown(message("content"))
  
   if prompt := st.chat_input("Ask me anything..."):
       st.session_state.messages.append({"role": "user", "content": prompt})
       with st.chat_message("user"):
           st.markdown(prompt)
      
       with st.chat_message("assistant"):
           callback_handler = StreamlitCallbackHandler(st.container())
          
           with st.spinner("🤔 Thinking..."):
               response = st.session_state.agent_system.chat(prompt, callback_handler)
          
           st.markdown(f"""
           

               {response}
           
           """, unsafe_allow_html=True)
          
           st.session_state.messages.append({"role": "assistant", "content": response})
  
   st.header("💡 Example Queries")
   col1, col2, col3 = st.columns(3)
  
   with col1:
       if st.button("🔍 Search Example"):
           example = "Search for the latest developments in quantum computing"
           st.session_state.example_query = example
  
   with col2:
       if st.button("🧮 Math Example"):
           example = "Calculate the compound interest on $1000 at 5% for 3 years"
           st.session_state.example_query = example
  
   with col3:
       if st.button("🧠 Memory Example"):
           example = "Remember that I work as a data scientist at TechCorp"
           st.session_state.example_query = example
  
   if 'example_query' in st.session_state:
       st.info(f"Example query: {st.session_state.example_query}")

def setup_ngrok_auth(auth_token):
   """Setup ngrok authentication"""
   try:
       from pyngrok import ngrok, conf
      
       conf.get_default().auth_token = auth_token
      
       try:
           tunnels = ngrok.get_tunnels()
           print("✅ Ngrok authentication successful!")
           return True
       except Exception as e:
           print(f"❌ Ngrok authentication failed: {e}")
           return False
          
   except ImportError:
       print("❌ pyngrok not installed. Installing...")
       import subprocess
       subprocess.run(('pip', 'install', 'pyngrok'), check=True)
       return setup_ngrok_auth(auth_token)


def get_ngrok_token_instructions():
   """Provide instructions for getting ngrok token"""
   return """
🔧 NGROK AUTHENTICATION SETUP:


1. Sign up for an ngrok account:
  - Visit: https://dashboard.ngrok.com/signup
  - Create a free account


2. Get your authentication token:
  - Go to: https://dashboard.ngrok.com/get-started/your-authtoken
  - Copy your authtoken


3. Replace 'your-ngrok-auth-token-here' in the code with your actual token


4. Alternative methods if ngrok fails:
  - Use Google Colab's built-in public URL feature
  - Use localtunnel: !npx localtunnel --port 8501
  - Use serveo.net: !ssh -R 80:localhost:8501 serveo.net
"""

def main():
   """Main function to run the application"""
   try:
       create_streamlit_app()
   except Exception as e:
       st.error(f"Application error: {str(e)}")
       st.info("Please check your API key and try refreshing the page")

def run_in_colab():
   """Run the application in Google Colab with proper ngrok setup"""
  
   print("🚀 Starting Advanced LangChain Agent Setup...")
  
   if NGROK_AUTH_TOKEN == "your-ngrok-auth-token-here":
       print("⚠️  NGROK_AUTH_TOKEN not configured!")
       print(get_ngrok_token_instructions())
      
       print("🔄 Attempting alternative tunnel methods...")
       try_alternative_tunnels()
       return
  
   print("📦 Installing required packages...")
   import subprocess
  
   packages = (
       'streamlit',
       'langchain',
       'langchain-google-genai',
       'langchain-community',
       'wikipedia',
       'duckduckgo-search',
       'pyngrok'
   )
  
   for package in packages:
       try:
           subprocess.run(('pip', 'install', package), check=True, capture_output=True)
           print(f"✅ {package} installed")
       except subprocess.CalledProcessError:
           print(f"⚠️  Failed to install {package}")
  
   app_content=""'
import streamlit as st
import os
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool, WikipediaQueryRun, DuckDuckGoSearchRun
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import PromptTemplate
from langchain.callbacks.streamlit import StreamlitCallbackHandler
from langchain_community.utilities import WikipediaAPIWrapper, DuckDuckGoSearchAPIWrapper
from datetime import datetime


# Configuration - Replace with your actual keys
GOOGLE_API_KEY = "''' + GOOGLE_API_KEY + '''"
os.environ("GOOGLE_API_KEY") = GOOGLE_API_KEY


class InnovativeAgentTools:
   @staticmethod
   def get_calculator_tool():
       def calculate(expression: str) -> str:
           try:
               allowed_chars = set('0123456789+-*/.() ')
               if all(c in allowed_chars for c in expression):
                   result = eval(expression)
                   return f"Result: {result}"
               else:
                   return "Error: Invalid mathematical expression"
           except Exception as e:
               return f"Calculation error: {str(e)}"
      
       return Tool(name="Calculator", func=calculate,
                  description="Calculate mathematical expressions. Input should be a valid math expression.")
  
   @staticmethod
   def get_memory_tool(memory_store):
       def save_memory(key_value: str) -> str:
           try:
               key, value = key_value.split(":", 1)
               memory_store(key.strip()) = value.strip()
               return f"Saved '{key.strip()}' to memory"
           except:
               return "Error: Use format 'key: value'"
      
       def recall_memory(key: str) -> str:
           return memory_store.get(key.strip(), f"No memory found for '{key}'")
      
       return (
           Tool(name="SaveMemory", func=save_memory, description="Save information to memory. Format: 'key: value'"),
           Tool(name="RecallMemory", func=recall_memory, description="Recall saved information. Input: key to recall")
       )
  
   @staticmethod
   def get_datetime_tool():
       def get_current_datetime(format_type: str = "full") -> str:
           now = datetime.now()
           if format_type == "date":
               return now.strftime("%Y-%m-%d")
           elif format_type == "time":
               return now.strftime("%H:%M:%S")
           else:
               return now.strftime("%Y-%m-%d %H:%M:%S")
      
       return Tool(name="DateTime", func=get_current_datetime,
                  description="Get current date/time. Options: 'date', 'time', or 'full'")


class MultiAgentSystem:
   def __init__(self, api_key: str):
       self.llm = ChatGoogleGenerativeAI(
           model="gemini-pro",
           google_api_key=api_key,
           temperature=0.7,
           convert_system_message_to_human=True
       )
       self.memory_store = {}
       self.conversation_memory = ConversationBufferWindowMemory(
           memory_key="chat_history", k=10, return_messages=True
       )
       self.tools = self._initialize_tools()
       self.agent = self._create_agent()
  
   def _initialize_tools(self):
       tools = ()
       try:
           tools.extend((
               DuckDuckGoSearchRun(api_wrapper=DuckDuckGoSearchAPIWrapper()),
               WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
           ))
       except Exception as e:
           st.warning(f"Search tools may have limited functionality: {e}")
      
       tools.append(InnovativeAgentTools.get_calculator_tool())
       tools.append(InnovativeAgentTools.get_datetime_tool())
       tools.extend(InnovativeAgentTools.get_memory_tool(self.memory_store))
       return tools
  
   def _create_agent(self):
       prompt = PromptTemplate.from_template("""
🤖 You are an advanced AI assistant with access to multiple tools and persistent memory.


AVAILABLE TOOLS:
{tools}


TOOL USAGE FORMAT:
- Think step by step about what you need to do
- Use Action: tool_name
- Use Action Input: your input
- Wait for Observation
- Continue until you have a final answer


CONVERSATION HISTORY:
{chat_history}


CURRENT QUESTION: {input}


REASONING PROCESS:
{agent_scratchpad}


Begin your response with your thought process, then take action if needed.
""")
      
       agent = create_react_agent(self.llm, self.tools, prompt)
       return AgentExecutor(agent=agent, tools=self.tools, memory=self.conversation_memory,
                          verbose=True, handle_parsing_errors=True, max_iterations=5)
  
   def chat(self, message: str, callback_handler=None):
       try:
           if callback_handler:
               response = self.agent.invoke({"input": message}, {"callbacks": (callback_handler)})
           else:
               response = self.agent.invoke({"input": message})
           return response("output")
       except Exception as e:
           return f"Error processing request: {str(e)}"


# Streamlit App
st.set_page_config(page_title="🚀 Advanced LangChain Agent", page_icon="🤖", layout="wide")


st.markdown("""

""", unsafe_allow_html=True)


st.markdown('
Powered by LangChain + Gemini API
', unsafe_allow_html=True)


with st.sidebar:
   st.header("🔧 Configuration")
   api_key = st.text_input("🔑 Google AI API Key", type="password", value=GOOGLE_API_KEY)
  
   if not api_key:
       st.error("Please enter your Google AI API key")
       st.stop()
  
   st.success("✅ API Key configured")
  
   st.header("🤖 Agent Capabilities")
   st.markdown("- 🔍 Web Search\\n- 📚 Wikipedia\\n- 🧮 Calculator\\n- 🧠 Memory\\n- 📅 Date/Time")
  
   if 'agent_system' in st.session_state and st.session_state.agent_system.memory_store:
       st.header("🧠 Memory Store")
       for key, value in st.session_state.agent_system.memory_store.items():
           st.markdown(f'
{key}: {value}
', unsafe_allow_html=True)


if 'agent_system' not in st.session_state:
   with st.spinner("🔄 Initializing Agent..."):
       st.session_state.agent_system = MultiAgentSystem(api_key)
   st.success("✅ Agent Ready!")


if 'messages' not in st.session_state:
   st.session_state.messages = ({
       "role": "assistant",
       "content": "🤖 Hello! I'm your advanced AI assistant. I can search, calculate, remember information, and more! Try asking me to: calculate something, search for information, or remember a fact about you."
   })


for message in st.session_state.messages:
   with st.chat_message(message("role")):
       st.markdown(message("content"))


if prompt := st.chat_input("Ask me anything..."):
   st.session_state.messages.append({"role": "user", "content": prompt})
   with st.chat_message("user"):
       st.markdown(prompt)
  
   with st.chat_message("assistant"):
       callback_handler = StreamlitCallbackHandler(st.container())
       with st.spinner("🤔 Thinking..."):
           response = st.session_state.agent_system.chat(prompt, callback_handler)
       st.markdown(f'
{response}
', unsafe_allow_html=True)
       st.session_state.messages.append({"role": "assistant", "content": response})


# Example buttons
st.header("💡 Try These Examples")
col1, col2, col3 = st.columns(3)
with col1:
   if st.button("🧮 Calculate 15 * 8 + 32"):
       st.rerun()
with col2:
   if st.button("🔍 Search AI news"):
       st.rerun()
with col3:
   if st.button("🧠 Remember my name is Alex"):
       st.rerun()
'''
  
   with open('streamlit_app.py', 'w') as f:
       f.write(app_content)
  
   print("✅ Streamlit app file created successfully!")
  
   if setup_ngrok_auth(NGROK_AUTH_TOKEN):
       start_streamlit_with_ngrok()
   else:
       print("❌ Ngrok authentication failed. Trying alternative methods...")
       try_alternative_tunnels()

def start_streamlit_with_ngrok():
   """Start Streamlit with ngrok tunnel"""
   import subprocess
   import threading
   from pyngrok import ngrok
  
   def start_streamlit():
       subprocess.run(('streamlit', 'run', 'streamlit_app.py', '--server.port=8501', '--server.headless=true'))
  
   print("🚀 Starting Streamlit server...")
   thread = threading.Thread(target=start_streamlit)
   thread.daemon = True
   thread.start()
  
   time.sleep(5)
  
   try:
       print("🌐 Creating ngrok tunnel...")
       public_url = ngrok.connect(8501)
       print(f"🔗 SUCCESS! Access your app at: {public_url}")
       print("✨ Your Advanced LangChain Agent is now running publicly!")
       print("📱 You can share this URL with others!")
      
       print("⏳ Keeping tunnel alive... Press Ctrl+C to stop")
       try:
           ngrok_process = ngrok.get_ngrok_process()
           ngrok_process.proc.wait()
       except KeyboardInterrupt:
           print("👋 Shutting down...")
           ngrok.kill()
          
   except Exception as e:
       print(f"❌ Ngrok tunnel failed: {e}")
       try_alternative_tunnels()


def try_alternative_tunnels():
   """Try alternative tunneling methods"""
   print("🔄 Trying alternative tunnel methods...")
  
   import subprocess
   import threading
  
   def start_streamlit():
       subprocess.run(('streamlit', 'run', 'streamlit_app.py', '--server.port=8501', '--server.headless=true'))
  
   thread = threading.Thread(target=start_streamlit)
   thread.daemon = True
   thread.start()
  
   time.sleep(3)
  
   print("🌐 Streamlit is running on http://localhost:8501")
   print("\n📋 ALTERNATIVE TUNNEL OPTIONS:")
   print("1. localtunnel: Run this in a new cell:")
   print("   !npx localtunnel --port 8501")
   print("\n2. serveo.net: Run this in a new cell:")
   print("   !ssh -R 80:localhost:8501 serveo.net")
   print("\n3. Colab public URL (if available):")
   print("   Use the 'Public URL' button in Colab's interface")
  
   try:
       while True:
           time.sleep(60)
   except KeyboardInterrupt:
       print("👋 Shutting down...")


if __name__ == "__main__":
   try:
       get_ipython()
       print("🚀 Google Colab detected - starting setup...")
       run_in_colab()
   except NameError:
       main()