Создание агентов LangChain для LLM на Python

from langchain_community.tools import DuckDuckGoSearchResults
from langchain.chains import LLMMathChain
from langchain.agents import Tool
 
# Создаем инструмент для поиска
search_tool = DuckDuckGoSearchResults()
 
# Создаем инструмент для математических вычислений
math_chain = LLMMathChain.from_llm(llm=my_llm)
math_tool = Tool.from_function(
    name="Calculator",
    func=math_chain.run,
    description="Выполняет математические вычисления"
)

from langchain.agents import AgentType, initialize_agent
 
agent = initialize_agent(
    tools=[search_tool, math_tool],
    llm=my_llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# Создание агента с включенным режимом подробного вывода
verbose_agent = initialize_agent(
    tools=[search_tool, math_tool],
    llm=my_llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True  # Включаем подробный вывод процесса рассуждений
)

# Установка необходимых пакетов
[H2]pip install langchain langchain-google-genai duckduckgo-search[/H2]
 
from langchain_google_genai import ChatGoogleGenerativeAI
import os
 
# Установка ключа API
os.environ['GOOGLE_API_KEY'] = "ваш_API_ключ"
 
# Создание экземпляра модели
llm = ChatGoogleGenerativeAI(model="gemini-pro")

from langchain_community.tools import DuckDuckGoSearchResults
 
# Создаем инструмент поиска
search_tool = DuckDuckGoSearchResults()

from langchain.chains import LLMMathChain
from langchain.agents import Tool
 
# Создаем цепочку для математических вычислений
math_chain = LLMMathChain.from_llm(llm=llm)
 
# Оборачиваем ее в инструмент
math_tool = Tool.from_function(
    name="Calculator",
    func=math_chain.run,
    description="Используй этот инструмент для математических операций. Вводи только математические выражения."
)

from langchain.agents import AgentType, initialize_agent
 
# Создаем агента
agent = initialize_agent(
    tools=[search_tool, math_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True  # Включаем подробный вывод
)
 
# Запускаем агента с вопросом
result = agent.invoke("Какой департамент возглавляет Илон Маск и сколько будет 234 * 15?")
print(result['output'])

from langchain.prompts import PromptTemplate
 
# Создаем шаблон промпта
template = """Ты - полезный ассистент, который отвечает на вопросы, используя доступные инструменты.
 
Доступные инструменты:
{tools}
 
Используй следующий формат:
Вопрос: вопрос пользователя
Мысли: подумай, как лучше ответить на вопрос
Действие: название инструмента (одно из [{tool_names}])
Входные данные: данные для инструмента
Наблюдение: результат действия
... (повторяй Мысли/Действие/Входные данные/Наблюдение сколько нужно)
Мысли: теперь я знаю финальный ответ
Финальный ответ: ответ на вопрос пользователя
 
Вопрос: {input}
Мысли:"""
 
prompt = PromptTemplate.from_template(template)

from langchain.agents import create_react_agent
 
# Создаем агента с кастомным промптом
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

   search_tool = DuckDuckGoSearchResults(num_results=3, snippets_count=3)

import requests
 
def get_weather(location):
    """Получает текущую погоду для указанного местоположения."""
    api_key = "ваш_ключ_API_погоды"
    url = f"https://api.weatherapi.com/v1/current.json?key={api_key}&q={location}"
    response = requests.get(url)
    data = response.json()
    return f"Погода в {location}: {data['current']['temp_c']}°C, {data['current']['condition']['text']}"
 
# Создаем инструмент
weather_tool = Tool.from_function(
    name="WeatherInfo",
    func=get_weather,
    description="Получает текущую погоду для указанного местоположения. Вводи название города или региона."
)

agent = initialize_agent(
    tools=[search_tool, math_tool, weather_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

def analyze_stock(ticker):
    """Получает и анализирует данные акции."""
    # Получаем текущую цену
    current_price = get_current_price(ticker)
    # Получаем исторические данные
    historical_data = get_historical_data(ticker)
    # Проводим анализ
    analysis = perform_simple_analysis(current_price, historical_data)
    return f"Анализ акции {ticker}: {analysis}"
 
stock_tool = Tool.from_function(
    name="StockAnalyzer",
    func=analyze_stock,
    description="Анализирует данные акции по её тикеру (например, AAPL для Apple)."
)

def get_weather(location):
    """Получает текущую погоду для указанного местоположения."""
    # Валидация входных данных
    if not location or not isinstance(location, str):
        return "Ошибка: укажите корректное местоположение в виде строки."
    
    try:
        api_key = "ваш_ключ_API_погоды"
        url = f"https://api.weatherapi.com/v1/current.json?key={api_key}&q={location}"
        response = requests.get(url)
        response.raise_for_status()  # Вызовет исключение при HTTP-ошибках
        data = response.json()
        return f"Погода в {location}: {data['current']['temp_c']}°C, {data['current']['condition']['text']}"
    except requests.exceptions.RequestException as e:
        return f"Ошибка при запросе погоды: {str(e)}"
    except (KeyError, ValueError) as e:
        return f"Ошибка при обработке данных погоды: {str(e)}"

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10  # Ограничиваем число итераций
)

def logged_tool(func, name):
    def wrapper(*args, **kwargs):
        print(f"[DEBUG] Вызов инструмента {name}")
        print(f"[DEBUG] Аргументы: {args}, {kwargs}")
        result = func(*args, **kwargs)
        print(f"[DEBUG] Результат: {result[:100]}...")  # Показываем только начало для длинных результатов
        return result
    return wrapper
 
# Оборачиваем инструмент в логгер
weather_tool.func = logged_tool(weather_tool.func, weather_tool.name)

from langchain.callbacks import get_openai_callback
 
# Фиксируем все запросы к OpenAI
with get_openai_callback() as cb:
    result = agent.invoke("Ваш запрос")
    print(f"Токенов использовано: {cb.total_tokens}")
    print(f"Стоимость: ${cb.total_cost}")

from langchain.tools import HumanInputRun
 
human_input_tool = HumanInputRun()
 
tools = [search_tool, math_tool, human_input_tool]
 
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

from langchain.agents import Tool, AgentExecutor
from langchain.agents.agent_toolkits import create_spark_dataframe_agent
from langchain.agents.agent_types import AgentType
from langchain.memory import ConversationBufferMemory
 
# Создаем память для агента
memory = ConversationBufferMemory(memory_key="chat_history")
 
# Включаем возврат промежуточных шагов
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    return_intermediate_steps=True  # Это позволит нам анализировать шаги
)

from functools import lru_cache
 
@lru_cache(maxsize=100)
def cached_math_operation(expression):
    # Вычисление математического выражения
    return eval(expression)
 
# Создаем инструмент с кэшированием
cached_math_tool = Tool.from_function(
    name="CachedCalculator",
    func=cached_math_operation,
    description="Выполняет математические вычисления с кэшированием результатов."
)

from langchain.agents import AgentType, initialize_agent
 
zero_shot_agent = initialize_agent(
    tools=[search_tool, math_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentType, initialize_agent
 
memory = ConversationBufferMemory(memory_key="chat_history")
 
conversational_agent = initialize_agent(
    tools=[search_tool, math_tool],
    llm=llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True
)

from langchain.agents.agent_toolkits import create_self_ask_with_search_agent
 
self_ask_agent = create_self_ask_with_search_agent(
    llm=llm,
    search_tool=search_tool,
    verbose=True
)

from langchain.agents import Tool
from langchain.agents.agent_toolkits import create_retrieval_agent
 
# Создаем инструмент для работы с документами
retriever = your_vector_db.as_retriever()
retrieval_tool = Tool(
    name="DocSearch",
    func=retriever.get_relevant_documents,
    description="Ищет информацию в базе документов. Используй это для поиска конкретных фактов."
)
 
# Создаем агента для работы с этим инструментом
retrieval_agent = create_retrieval_agent(
    llm=llm,
    tools=[retrieval_tool],
    verbose=True
)

from langchain.agents import load_tools
from langchain.agents.agent_toolkits import create_plan_and_execute_agent
 
# Загружаем набор стандартных инструментов
tools = load_tools(["serpapi", "llm-math"], llm=llm)
 
# Создаем агента с планированием
planner_agent = create_plan_and_execute_agent(
    llm=llm,
    tools=tools,
    verbose=True
)

from langchain.agents import AgentType, initialize_agent
from langchain.prompts import StructuredChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List
 
# Определяем структуру ответа
class MovieRecommendation(BaseModel):
    title: str = Field(description="Название фильма")
    year: int = Field(description="Год выпуска")
    rating: float = Field(description="Рейтинг от 0 до 10")
    reasons: List[str] = Field(description="Причины рекомендации")
 
# Создаем агента с структурированным выводом
structured_agent = initialize_agent(
    tools=[search_tool],
    llm=llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

from langchain.agents import AgentType, initialize_agent
 
openai_functions_agent = initialize_agent(
    tools=[search_tool, math_tool],
    llm=llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True
)

from langchain.agents import ZeroShotAgent, AgentExecutor
from langchain.memory import ConversationBufferMemory
 
# Создаем шаблон промпта для кастомного агента
prefix = """Ты - эксперт по анализу данных. Используй следующие инструменты для анализа данных клиентов:"""
suffix = """Начинай!
 
Вопрос: {input}
{agent_scratchpad}"""
 
# Определяем промпт для агента
prompt = ZeroShotAgent.create_prompt(
    tools=[data_analysis_tool, visualization_tool],
    prefix=prefix,
    suffix=suffix,
    input_variables=["input", "agent_scratchpad"]
)
 
# Создаем агента с кастомным промптом
agent = ZeroShotAgent(llm_chain=LLMChain(llm=llm, prompt=prompt), tools=[data_analysis_tool, visualization_tool])
agent_executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=[data_analysis_tool, visualization_tool], verbose=True)

# Создаем агента для поиска информации
research_agent = initialize_agent(
    tools=[search_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)
 
# Создаем агента для анализа данных
analysis_agent = initialize_agent(
    tools=[math_tool, data_analysis_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)
 
# Создаем агента-координатора
def coordinator(task):
    """Распределяет задачи между специализированными агентами"""
    if "найди" in task.lower() or "поиск" in task.lower():
        return research_agent.run(task)
    elif "анализ" in task.lower() or "рассчитай" in task.lower():
        return analysis_agent.run(task)
    else:
        # Если не ясно, какой агент нужен, используем эвристику или спрашиваем у модели
        task_type = llm.predict(f"Определи тип задачи: '{task}'. Это поиск информации или анализ данных?")
        if "поиск" in task_type.lower():
            return research_agent.run(task)
        else:
            return analysis_agent.run(task)

# Функция для оценки производительности агента
def evaluate_agent(agent, tasks, metrics=["accuracy", "steps", "time"]):
 results = {}
 for metric in metrics:
     results[metric] = []
 
 for task in tasks:
     start_time = time.time()
     response = agent.invoke(task)
     end_time = time.time()
     
     # Оцениваем точность (это субъективная метрика в реальности)
     accuracy = evaluate_accuracy(response, task)
     
     # Количество шагов (для агентов с verbose=True)
     steps = len(response.get("intermediate_steps", []))
     
     results["accuracy"].append(accuracy)
     results["steps"].append(steps)
     results["time"].append(end_time - start_time)
 
 # Возвращаем средние значения
 return {k: sum(v)/len(v) for k, v in results.items()}

def dynamic_agent_selector(query, llm):
 """Выбирает подходящий тип агента на основе анализа запроса"""
 
 # Анализируем запрос с помощью LLM
 analysis = llm.predict(
     f"""Проанализируй запрос и определи, какой тип агента лучше использовать:
     Запрос: {query}
     
     1. Если это простой фактический вопрос - "basic"
     2. Если это многоступенчатый сложный вопрос - "self_ask"
     3. Если это диалог, требующий контекста - "conversational"
     4. Если это задача, требующая планирования - "planner"
     
     Верни только одно слово - тип агента."""
 )
 
 # Создаем соответствующий агент
 if "basic" in analysis.lower():
     return initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)
 elif "self_ask" in analysis.lower():
     return create_self_ask_with_search_agent(llm, search_tool)
 elif "conversational" in analysis.lower():
     memory = ConversationBufferMemory(memory_key="chat_history")
     return initialize_agent(tools, llm, agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION, memory=memory)
 elif "planner" in analysis.lower():
     return create_plan_and_execute_agent(llm, tools)
 else:
     # По умолчанию используем универсальный агент
     return initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION)

# Создаем инструменты для работы с документами
retriever = your_medical_db.as_retriever()
doc_tool = Tool(
 name="MedicalKnowledge",
 func=retriever.get_relevant_documents,
 description="Ищет информацию в медицинской базе знаний."
)
 
# Создаем кастомный промпт для планирования
planning_prompt = """Ты - медицинский ИИ-ассистент.
Составь план диагностики для пациента с описанными симптомами.
План должен содержать шаги, каждый с конкретным диагностическим вопросом или проверкой.
 
Симптомы: {input}
 
План:"""
 
# Функция планирования
def create_diagnostic_plan(symptoms):
 return llm.predict(planning_prompt.format(input=symptoms))
 
# Функция выполнения плана с поиском информации
def execute_diagnostic_step(step):
 # Ищем информацию для этого шага
 documents = doc_tool.run(step)
 # Анализируем найденную информацию
 analysis = llm.predict(f"Шаг диагностики: {step}\n\nИнформация: {documents}\n\nАнализ:")
 return analysis
 
# Основная функция диагностики
def medical_diagnosis(symptoms):
 # Создаем план
 plan = create_diagnostic_plan(symptoms).split("\n")
 results = []
 
 # Выполняем каждый шаг плана
 for step in plan:
     if step.strip():
         step_result = execute_diagnostic_step(step)
         results.append(f"- {step}: {step_result}")
 
 # Формируем итоговое заключение
 final_diagnosis = llm.predict(f"""Симптомы: {symptoms}
     
Результаты диагностики:
{''.join(results)}
 
Предположительный диагноз и рекомендации:""")
 
 return final_diagnosis

# Предотвращение зацикливания
agent_executor = AgentExecutor(
 agent=agent,
 tools=tools,
 verbose=True,
 max_iterations=10,  # Ограничиваем число итераций
 max_execution_time=60,  # Ограничиваем время выполнения в секундах
 early_stopping_method="generate"  # Генерируем ответ, если достигнут лимит
)
 
# Валидация финального ответа
def validate_final_answer(agent_output):
 """Проверяет финальный ответ агента на соответствие собранной информации"""
 steps = agent_output.get("intermediate_steps", [])
 final_answer = agent_output.get("output", "")
 
 # Собираем всю информацию из промежуточных шагов
 collected_info = "\n".join([f"{action}: {observation}" for action, observation in steps])
 
 # Проверяем соответствие финального ответа собранной информации
 validation = llm.predict(f"""
 Собранная информация:
 {collected_info}
 
 Финальный ответ:
 {final_answer}
 
 Проверь, соответствует ли финальный ответ собранной информации.
 Если есть несоответствия или ошибки, напиши исправленный ответ.
 Если ответ корректен, напиши "Ответ корректен".
 """)
 
 if "ответ корректен" in validation.lower():
     return final_answer
 else:
     return validation

from langchain.agents import Tool
from pydantic import BaseModel, Field
from typing import Optional
import requests
 
class WeatherParams(BaseModel):
    """Параметры для запроса погоды."""
    location: str = Field(..., description="Город или координаты")
    days: Optional[int] = Field(1, description="Количество дней для прогноза")
    
class WeatherAPI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.weatherapi.com/v1"
        
    def get_weather(self, params: WeatherParams):
        """Получает информацию о погоде с валидацией и обработкой ошибок."""
        try:
            # Валидация параметров
            if params.days < 1 or params.days > 10:
                return "Ошибка: количество дней должно быть от 1 до 10."
                
            url = f"{self.base_url}/forecast.json"
            response = requests.get(url, params={
                "key": self.api_key,
                "q": params.location,
                "days": params.days
            }, timeout=5)
            
            if response.status_code == 400:
                return f"Не удалось найти местоположение '{params.location}'"
            
            response.raise_for_status()
            data = response.json()
            
            # Форматирование результата для удобства агента
            result = f"Погода в {data['location']['name']}, {data['location']['country']}:\n"
            for day in data['forecast']['forecastday']:
                result += f"- {day['date']}: {day['day']['condition']['text']}, "
                result += f"Температура: {day['day']['avgtemp_c']}°C\n"
            
            return result
            
        except requests.exceptions.Timeout:
            return "Ошибка: превышено время ожидания ответа от сервера погоды."
        except requests.exceptions.RequestException as e:
            return f"Ошибка при запросе погоды: {str(e)}"
        except (KeyError, ValueError, TypeError) as e:
            return f"Ошибка при обработке данных погоды: {str(e)}"
 
# Создаем экземпляр API
weather_api = WeatherAPI(api_key="ваш_ключ_API")
 
# Создаем инструмент
weather_tool = Tool.from_function(
    name="WeatherForecast",
    func=lambda x: weather_api.get_weather(WeatherParams.parse_raw(x)),
    description="Получает прогноз погоды. Вводи JSON с полями: location (строка) и days (число от 1 до 10)."
)

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory, ConversationBufferWindowMemory
 
# Простая буферная память - хранит всю историю
buffer_memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
 
# Память с окном - хранит только последние N взаимодействий
window_memory = ConversationBufferWindowMemory(
    memory_key="chat_history", 
    k=5,  # Хранить только последние 5 обменов
    return_messages=True
)
 
# Суммаризирующая память - сохраняет сжатое представление длинных диалогов
summary_memory = ConversationSummaryMemory(
    llm=llm,  # Нужна модель для суммаризации
    memory_key="chat_history",
    return_messages=True
)

from langchain.memory import CombinedMemory
 
# Комбинируем суммаризацию с буфером последних сообщений
combined_memory = CombinedMemory(
    memories=[
        ConversationSummaryMemory(llm=llm, memory_key="chat_summary"),
        ConversationBufferWindowMemory(k=3, memory_key="recent_messages")
    ]
)
 
agent = initialize_agent(
    tools=[search_tool, math_tool],
    llm=llm,
    agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
    memory=combined_memory,
    verbose=True
)

from langchain.smith import RunEvalConfig, run_on_dataset
import time
 
# Обертка для обработки ошибок инструментов
def tool_with_fallback(primary_tool, fallback_tool=None, max_retries=2):
    def wrapper(*args, **kwargs):
        attempts = 0
        while attempts <= max_retries:
            try:
                return primary_tool(*args, **kwargs)
            except Exception as e:
                attempts += 1
                if attempts > max_retries:
                    if fallback_tool:
                        try:
                            return fallback_tool(*args, **kwargs)
                        except:
                            return f"Все методы получения данных не сработали. Ошибка: {str(e)}"
                    else:
                        return f"Не удалось выполнить операцию после {max_retries} попыток. Ошибка: {str(e)}"
                time.sleep(1)  # Пауза перед повторной попыткой
    return wrapper
 
# Применяем обертку к инструменту
weather_tool_with_fallback = Tool.from_function(
    name="WeatherForecast",
    func=tool_with_fallback(
        weather_api.get_weather,
        fallback_tool=lambda params: "Сегодня ожидается типичная для этого времени года погода"
    ),
    description="Получает прогноз погоды"
)

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader
 
# Загружаем и подготавливаем документы
documents = TextLoader("./data/documentation.txt").load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
splits = text_splitter.split_documents(documents)
 
# Создаем векторное хранилище
embedding_model = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents=splits, embedding=embedding_model)
 
# Создаем инструмент для поиска по документам
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
doc_tool = Tool.from_function(
    name="DocumentSearch",
    func=retriever.get_relevant_documents,
    description="Ищет информацию в технической документации. Используй для поиска конкретных технических деталей."
)

from langchain.prompts import MessagesPlaceholder
from langchain.agents.format_scratchpad import format_to_openai_function_messages
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
 
llm = ChatOpenAI(
    temperature=0.2,  # Низкая температура для более детерминированных ответов
    model="gpt-4",
    top_p=0.9,  # Слегка ограничиваем разнообразие
    max_tokens=1000  # Контролируем длину ответов
)

cot_template = """Ты - эксперт по решению сложных проблем.
Для каждой задачи действуй следующим образом:
1. Внимательно проанализируй условие задачи
2. Раздели задачу на подзадачи или шаги
3. Решай каждый шаг последовательно, объясняя свои рассуждения
4. Проверь свое решение
5. Сформулируй окончательный ответ
 
Задача: {input}
 
Размышление:"""
 
cot_prompt = PromptTemplate(
    template=cot_template,
    input_variables=["input"]
)
 
# Используем этот промпт в агенте
cot_chain = LLMChain(llm=llm, prompt=cot_prompt)

role_prompt = """Ты - опытный аналитик данных с экспертизой в финансах.
Ты умеешь:
Интерпретировать финансовые показатели
Выявлять тренды в данных
Делать обоснованные прогнозы
Объяснять сложные концепции простым языком
 
Используй свои навыки для ответа на следующий вопрос: {input}"""

few_shot_template = """Вот примеры, как нужно анализировать финансовые данные:
 
Вопрос: Как изменилась прибыль компании в Q2?
Мысли: Мне нужно найти данные о прибыли за Q2 и сравнить с предыдущим периодом.
Действие: FinancialDataTool
Входные данные: {"company": "ACME Corp", "metric": "profit", "period": "Q2"}
Наблюдение: Прибыль в Q2: $1.2M, Q1: $0.8M
Ответ: Прибыль компании ACME Corp выросла на 50% в Q2 по сравнению с Q1, с $0.8M до $1.2M.
 
Теперь твоя очередь:
Вопрос: {input}
Мысли:"""

reflection_prompt = """После каждого действия оцени его результат:
1. Получил ли я полезную информацию?
2. Приблизился ли я к ответу на вопрос?
3. Какое действие будет наиболее полезным дальше?
 
Вопрос: {input}
Мысли:"""

# Пример Docker файла для агента
# Dockerfile
FROM python:3.9-slim
 
WORKDIR /app
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
 
CMD ["python", "agent_service.py"]

# Пример AWS Lambda хендлера
def lambda_handler(event, context):
    query = event.get('query', '')
    
    # Инициализируем агента (можно использовать кеширование для ускорения)
    agent = initialize_agent(...)
    
    # Выполняем запрос
    result = agent.invoke(query)
    
    return {
        'statusCode': 200,
        'body': result['output']
    }

Agent Gateway (API) ─┬─ Spe******t Agent 1 ─── Tool Service 1
                     ├─ Spe******t Agent 2 ─┬─ Tool Service 2
                     │                      └─ External API
                     └─ Coordinator Agent ─── Vector Database

from functools import lru_cache
import hashlib
import json
 
# Кеширование вызовов API с учетом всех параметров
def cache_key_builder(*args, **kwargs):
    key = {
        'args': args,
        'kwargs': {k: v for k, v in kwargs.items() if k != 'api_key'}
    }
    return hashlib.md5(json.dumps(key, sort_keys=True).encode()).hexdigest()
 
@lru_cache(maxsize=100)
def cached_api_call(cache_key, *args, **kwargs):
    # Реальный вызов API
    return actual_api_function(*args, **kwargs)
 
def weather_with_cache(*args, **kwargs):
    cache_key = cache_key_builder(*args, **kwargs)
    return cached_api_call(cache_key, *args, **kwargs)

import logging
from datetime import datetime
 
class AgentMonitor:
   def __init__(self, log_file="agent_logs.jsonl"):
       self.log_file = log_file
       self.logger = logging.getLogger("agent_monitor")
       self.logger.setLevel(logging.INFO)
       
   def log_interaction(self, query, response, intermediate_steps=None, metadata=None):
       log_entry = {
           "timestamp": datetime.now().isoformat(),
           "query": query,
           "response": response,
           "steps": intermediate_steps or [],
           "metadata": metadata or {}
       }
       
       with open(self.log_file, "a") as f:
           f.write(json.dumps(log_entry) + "
")
           
   def analyze_logs(self, days=7):
       """Анализирует логи за последние N дней."""
       # Анализ логов, выявление паттернов, проблем и т.д.
       pass
 
# Использование монитора
monitor = AgentMonitor()
 
def agent_with_monitoring(query, metadata=None):
   result = agent.invoke(query)
   monitor.log_interaction(
       query=query,
       response=result["output"],
       intermediate_steps=result.get("intermediate_steps"),
       metadata=metadata
   )
   return result

from sklearn.cluster import KMeans
import numpy as np
 
def analyze_failure_patterns(logs, embedding_model):
   # Извлекаем проблемные запросы
   failed_queries = [log["query"] for log in logs if "error" in log["metadata"]]
   
   # Получаем эмбеддинги
   embeddings = [embedding_model.embed_query(q) for q in failed_queries]
   
   # Кластеризуем для выявления паттернов
   kmeans = KMeans(n_clusters=5)
   clusters = kmeans.fit_predict(np.array(embeddings))
   
   # Анализируем каждый кластер
   for i in range(5):
       cluster_queries = [q for q, c in zip(failed_queries, clusters) if c == i]
       print(f"Кластер ошибок #{i}:")
       for q in cluster_queries[:5]:  # Первые 5 примеров
           print(f"  - {q}")

from langchain.callbacks.base import BaseCallbackHandler
 
class SecurityCallbackHandler(BaseCallbackHandler):
   """Обработчик для проверки безопасности запросов и ответов."""
   
   def __init__(self, blocked_terms=None, sensitive_actions=None):
       self.blocked_terms = blocked_terms or []
       self.sensitive_actions = sensitive_actions or {}
       
   def on_agent_action(self, action, **kwargs):
       """Проверяет безопасность действия агента."""
       tool_name = action.tool
       tool_input = action.tool_input
       
       # Проверка на запрещенные термины
       for term in self.blocked_terms:
           if term.lower() in str(tool_input).lower():
               raise ValueError(f"Действие содержит запрещенный термин: {term}")
               
       # Дополнительные проверки для чувствительных действий
       if tool_name in self.sensitive_actions:
           validator = self.sensitive_actions[tool_name]
           if not validator(tool_input):
               raise ValueError(f"Действие не прошло проверку безопасности: {tool_name}")
               
       return action
 
# Валидаторы для конкретных инструментов
def validate_db_query(query):
   """Проверяет SQL-запрос на безопасность."""
   dangerous_patterns = ["DROP", "DELETE", "UPDATE", "--", ";"]
   return not any(pattern in query.upper() for pattern in dangerous_patterns)
 
# Настройка обработчика безопасности
security_handler = SecurityCallbackHandler(
   blocked_terms=["пароль", "кредитная карта", "секретный"],
   sensitive_actions={
       "DatabaseTool": validate_db_query,
       "ApiTool": lambda x: "admin" not in x.lower()
   }
)
 
# Применение к агенту
agent_executor = AgentExecutor.from_agent_and_tools(
   agent=agent,
   tools=tools,
   callbacks=[security_handler],
   verbose=True
)

class PermissionSystem:
   def __init__(self, default_policy="deny"):
       self.default_policy = default_policy
       self.permissions = {}
       
   def add_permission(self, tool_name, condition_func=None):
       """Добавляет разрешение для инструмента."""
       self.permissions[tool_name] = condition_func
       
   def check_permission(self, tool_name, tool_input, user_id=None, context=None):
       """Проверяет разрешение на использование инструмента."""
       if tool_name not in self.permissions:
           return self.default_policy == "allow"
           
       condition_func = self.permissions[tool_name]
       if condition_func is None:
           return True  # Разрешено безусловно
           
       return condition_func(tool_input, user_id, context)
 
# Пример использования
permissions = PermissionSystem(default_policy="deny")
permissions.add_permission("SearchTool")  # Разрешено всем
permissions.add_permission("DatabaseTool", 
   lambda input, user_id, ctx: user_id in ["admin", "manager"])  # Только для админов и менеджеров

class AdaptiveAgent:
   def __init__(self, base_agent, user_preferences=None):
       self.base_agent = base_agent
       self.user_preferences = user_preferences or {}
       self.user_history = {}
       
   def update_preferences(self, user_id, preferences):
       """Обновляет предпочтения пользователя."""
       if user_id not in self.user_preferences:
           self.user_preferences[user_id] = {}
       self.user_preferences[user_id].update(preferences)
       
   def invoke(self, query, user_id):
       """Выполняет запрос с учетом предпочтений пользователя."""
       # Получаем предпочтения пользователя
       prefs = self.user_preferences.get(user_id, {})
       
       # Модифицируем запрос с учетом предпочтений
       if prefs.get("verbose_explanations"):
           query = f"{query} Пожалуйста, дай подробное объяснение."
           
       if prefs.get("preferred_sources"):
           sources = ", ".join(prefs["preferred_sources"])
           query = f"{query} Приоритизируй информацию из следующих источников: {sources}."
           
       # Выполняем запрос
       result = self.base_agent.invoke(query)
       
       # Сохраняем историю взаимодействий
       if user_id not in self.user_history:
           self.user_history[user_id] = []
       self.user_history[user_id].append((query, result))
       
       return result

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
 
def recommend_tools_based_on_history(user_history, all_tools):
   """Рекомендует инструменты на основе истории запросов пользователя."""
   if not user_history:
       return all_tools[:3]  # Возвращаем топ-3 по умолчанию
       
   # Получаем тексты запросов
   queries = [q for q, _ in user_history]
   
   # Создаем TF-IDF векторы
   vectorizer = TfidfVectorizer()
   query_vectors = vectorizer.fit_transform(queries)
   
   # Вычисляем сходство с описаниями инструментов
   tool_descriptions = [tool.description for tool in all_tools]
   tool_vectors = vectorizer.transform(tool_descriptions)
   
   # Находим средний вектор запросов пользователя
   avg_query_vector = query_vectors.mean(axis=0)
   
   # Вычисляем сходство с каждым инструментом
   similarities = cosine_similarity(avg_query_vector, tool_vectors)
   
   # Сортируем инструменты по релевантности
   ranked_tools = [all_tools[i] for i in similarities.argsort()[0][::-1]]
   
   return ranked_tools

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
async def parallel_tools_execution(tools, query):
   """Выполняет запросы к инструментам параллельно."""
   with ThreadPoolExecutor() as executor:
       loop = asyncio.get_event_loop()
       tasks = [loop.run_in_executor(executor, tool.run, query) for tool in tools]
       return await asyncio.gather(*tasks)

from diskcache import Cache
 
cache = Cache("./prompt_cache")
 
def cached_embedding(text, model, ttl=86400):
   """Кеширует эмбеддинги для повторного использования."""
   key = f"emb_{model}_{hash(text)}"
   if key in cache:
       return cache[key]
   
   embedding = model.get_embedding(text)
   cache.set(key, embedding, expire=ttl)
   return embedding

# Для Flask или FastAPI приложений
llm = None
tools = None
 
def initialize():
   global llm, tools
   llm = ChatOpenAI(model="gpt-4")
   tools = [
       SearchTool(),
       CalculatorTool(),
       WeatherTool()
   ]
 
# Вызываем при старте сервера
initialize()

assistant/
  ├── tools/
  │   ├── search.py
  │   ├── calculator.py
  │   ├── weather.py
  │   └── scheduler.py
  ├── memory/
  │   └── conversation_store.py
  ├── monitoring/
  │   └── logger.py
  ├── coordinator.py
  └── main.py

# tools/search.py
from langchain_community.tools import DuckDuckGoSearchResults
from functools import lru_cache
import hashlib
import json
 
class CachedSearch:
    def __init__(self, max_cache_size=100):
        self.search_tool = DuckDuckGoSearchResults()
        self.search = lru_cache(maxsize=max_cache_size)(self._search)
        
    def _search(self, query):
        try:
            results = self.search_tool.run(query)
            return results
        except Exception as e:
            return f"Ошибка поиска: {str(e)}"
    
    def get_tool(self):
        from langchain.agents import Tool
        return Tool.from_function(
            name="InternetSearch",
            func=self.search,
            description="Ищет актуальную информацию в интернете. Используй для вопросов о текущих событиях, фактах и данных, которые могут быть не в твоей базе знаний."
        )

# tools/weather.py
import requests
from langchain.agents import Tool
from pydantic import BaseModel, Field
from typing import Optional
import json
 
class WeatherRequest(BaseModel):
    location: str = Field(..., description="Город или координаты")
    days: Optional[int] = Field(1, description="Количество дней прогноза")
 
class WeatherTool:
    def __init__(self, api_key):
        self.api_key = api_key
        
    def get_weather(self, request_str):
        try:
            # Парсим входные данные
            request = json.loads(request_str) if isinstance(request_str, str) else request_str
            location = request.get("location", "")
            days = request.get("days", 1)
            
            # Проверяем валидность
            if not location:
                return "Необходимо указать местоположение"
                
            # Делаем запрос к API
            url = f"https://api.weatherapi.com/v1/forecast.json"
            params = {
                "key": self.api_key,
                "q": location,
                "days": days
            }
            
            response = requests.get(url, params=params, timeout=5)
            response.raise_for_status()
            data = response.json()
            
            # Форматируем результат
            result = f"Погода в {data['location']['name']}, {data['location']['country']}:\n"
            for day in data['forecast']['forecastday']:
                result += f"- {day['date']}: {day['day']['condition']['text']}, "
                result += f"Температура: {day['day']['avgtemp_c']}°C\n"
                
            return result
            
        except json.JSONDecodeError:
            return "Ошибка: неверный формат запроса. Ожидается JSON."
        except requests.exceptions.RequestException as e:
            return f"Ошибка при запросе погоды: {str(e)}"
        except (KeyError, ValueError, TypeError) as e:
            return f"Ошибка при обработке данных: {str(e)}"
    
    def get_tool(self):
        return Tool.from_function(
            name="WeatherInfo",
            func=self.get_weather,
            description="Получает прогноз погоды для указанного местоположения. Принимает JSON с полями: location (строка) и days (число дней, по умолчанию 1)."
        )

# memory/conversation_store.py
from langchain.memory import ConversationSummaryBufferMemory
from datetime import datetime
 
class EnhancedMemory:
    def __init__(self, llm, max_token_limit=2000):
        self.internal_memory = ConversationSummaryBufferMemory(
            llm=llm,
            max_token_limit=max_token_limit,
            memory_key="chat_history",
            return_messages=True
        )
        self.metadata = {
            "session_start": datetime.now().isoformat(),
            "user_preferences": {},
            "conversation_topics": []
        }
    
    def add_user_message(self, message):
        self.internal_memory.chat_memory.add_user_message(message)
        
    def add_ai_message(self, message):
        self.internal_memory.chat_memory.add_ai_message(message)
    
    def get_memory_variables(self):
        return self.internal_memory.load_memory_variables({})
        
    def update_metadata(self, key, value):
        if key not in self.metadata:
            self.metadata[key] = value
        elif isinstance(self.metadata[key], list):
            self.metadata[key].append(value)
        else:
            self.metadata[key] = value

# coordinator.py
from langchain.agents import AgentType, initialize_agent
from langchain_google_genai import ChatGoogleGenerativeAI
import os
 
class AssistantCoordinator:
    def __init__(self, api_key, weather_api_key):
        # Настраиваем LLM
        os.environ['GOOGLE_API_KEY'] = api_key
        self.llm = ChatGoogleGenerativeAI(model="gemini-pro")
        
        # Инициализируем инструменты
        from tools.search import CachedSearch
        from tools.calculator import EnhancedCalculator
        from tools.weather import WeatherTool
        from tools.scheduler import ScheduleTool
        
        self.search_tool = CachedSearch().get_tool()
        self.calc_tool = EnhancedCalculator(self.llm).get_tool()
        self.weather_tool = WeatherTool(weather_api_key).get_tool()
        self.schedule_tool = ScheduleTool().get_tool()
        
        self.tools = [
            self.search_tool,
            self.calc_tool,
            self.weather_tool,
            self.schedule_tool
        ]
        
        # Создаем память
        from memory.conversation_store import EnhancedMemory
        self.memory = EnhancedMemory(self.llm)
        
        # Инициализируем агента
        self.agent = initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
            memory=self.memory.internal_memory,
            verbose=True
        )
        
        # Настраиваем логирование
        from monitoring.logger import AssistantLogger
        self.logger = AssistantLogger()
    
    def process_query(self, user_query, user_id=None):
        # Логируем запрос
        self.logger.log_request(user_query, user_id)
        
        # Обрабатываем запрос
        try:
            response = self.agent.invoke({"input": user_query})
            output = response["output"]
            
            # Логируем ответ
            self.logger.log_response(output, user_id)
            
            return {
                "status": "success",
                "response": output
            }
        except Exception as e:
            error_msg = f"Ошибка при обработке запроса: {str(e)}"
            self.logger.log_error(error_msg, user_id)
            
            return {
                "status": "error",
                "response": "Извините, произошла ошибка при обработке вашего запроса.",
                "error": str(e)
            }

# main.py
from flask import Flask, request, jsonify
from coordinator import AssistantCoordinator
import os
 
app = Flask(__name__)
 
# Инициализируем ассистента при запуске
assistant = AssistantCoordinator(
    api_key=os.environ.get("GEMINI_API_KEY"),
    weather_api_key=os.environ.get("WEATHER_API_KEY")
)
 
@app.route('/api/chat', methods=['POST'])
def chat():
    data = request.json
    user_query = data.get('query', '')
    user_id = data.get('user_id', 'anonymous')
    
    if not user_query:
        return jsonify({"status": "error", "message": "Query is required"}), 400
    
    result = assistant.process_query(user_query, user_id)
    return jsonify(result)
 
if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
async def parallel_tool_execution(tools, query):
    with ThreadPoolExecutor() as executor:
        loop = asyncio.get_event_loop()
        tasks = [loop.run_in_executor(executor, tool.run, query) for tool in tools]
        return await asyncio.gather(*tasks)

def preanalyze_query(self, query):
    """Предварительно анализирует запрос для выбора подходящих инструментов"""
    analysis = self.llm.predict(f"""
    Проанализируй запрос пользователя и определи, какие инструменты могут понадобиться:
    "{query}"
    
    Варианты инструментов:
    - search: для поиска информации
    - calculator: для математических расчетов
    - weather: для информации о погоде
    - scheduler: для управления расписанием
    
    Верни только список необходимых инструментов через запятую.
    """)
    
    needed_tools = []
    if "search" in analysis.lower():
        needed_tools.append(self.search_tool)
    if "calculator" in analysis.lower():
        needed_tools.append(self.calc_tool)
    if "weather" in analysis.lower():
        needed_tools.append(self.weather_tool)
    if "scheduler" in analysis.lower():
        needed_tools.append(self.schedule_tool)
    
    # Если ничего не определили, вернем все инструменты
    return needed_tools if needed_tools else self.tools