Обновить solutions/699cc158d6d3a5544a3ed35b_Stream-режим_AI-агента/agent.py

solutions/699cc158d6d3a5544a3ed35b_Stream-режим_AI-агента/agent.py

@@ -1,89 +1,119 @@
+
+from langchain.agents import create_agent
+from langchain_community.chat_models import GigaChat
+from langchain.tools import tool
 import os
-from typing import Dict, Any
 
-from langchain_ollama import ChatOllama
+GPT2GIGA_PORT=8090
-from nomic_embed_text import NomicEmbedText
+GIGACHAT_CREDENTIALS="MDE5YzBlMzUtNTJlYi03ODFiLTg1ZWUtZTc2MDFiZGUxYmM2OmNjZmIwODI2LTQ0ZTQtNDQwNC04NGE3LTgzNmM5ZDJhYmMzMg=="
-from langchain.tools import BaseTool
+GIGACHAT_SCOPE="GIGACHAT_API_B2B"
-from langchain.schema import HumanMessage, AIMessage, SystemMessage
+GIGACHAT_MODEL="GigaChat-MAX"
-from langchain.agents import AgentExecutor, initialize_agent, load_tools
+GIGACHAT_VERIFY_SSL_CERTS=False
-from langchain.prompts import ChatPromptTemplate
 
-# ---------- Настройки LLM и эмбеддингов ----------
+llm = GigaChat(
-LLM_MODEL = os.getenv("OLLAMA_MODEL", "llama3")
+    credentials=GIGACHAT_CREDENTIALS,
-EMBEDDING_MODEL = os.getenv("NOMIC_EMBEDDING_MODEL", "nomic-embed-text")
+    scope=GIGACHAT_SCOPE,
+    model=GIGACHAT_MODEL,
+    verify_ssl_certs=False,  
+    temperature=0.7,
+    timeout=60,
+)
 
-llm = ChatOllama(model=LLM_MODEL, temperature=0.2)
+@tool
-embedding = NomicEmbedText()
+def check_wish(wish: str) -> str:
+    """Инструмент для проверки желания на наличие подвоха"""
+    
+    genie_agent = create_agent(
+        model=llm,
+        tools=[],
+        system_prompt="""Ты - коварный джинн, который ищет подвох в любом желании.
+Проанализируй желание человека и найди скрытую опасность, буквальное толкование, неожиданные последствия.
+Если подвох найден - коротко предупреди о нем, например: 
+"хочу много денег - деньги будут фальшивыми."
+"хочу деньги на счет в банке - хорошо, но банк завтра обанкротится"
 
-# ---------- Пример инструмента ----------
+Если желание безопасно и не имеет подвоха - ответь "Желание безопасно! Исполняю."
-class DummyTool(BaseTool):
+Ты должен говорить только на русском языке."""
-    name: str = "get_price"
-    description: str = (
-        "Получает цену товара в указанном городе. "
-        "Аргументы: {'product': 'название', 'city': 'город'}."
     )
+    
+    result = genie_agent.invoke({
+        "messages": [
+            {"role": "human", "content": f"Проверь желание: {wish}"}
+        ]
+    })
+    
+    return result['messages'][-1].content
 
-    def _run(self, product: str, city: str) -> str:
+human_agent = create_agent(
-        # В реальном коде здесь будет запрос к API
+    model=llm,
-        return f"Цена {product} в {city}: 100₽"
+    tools=[check_wish],
-
+    system_prompt="""Ты - человек, который загадывает желания джинну.
-tool = DummyTool()
+Твоя задача - передать желание джинну через инструмент check_wish и сообщить результат.
-tools = [tool]
+Говори только на русском языке."""
-
-# ---------- Создание агента ----------
-prompt_template = ChatPromptTemplate.from_messages(
-    [
-        SystemMessage(content="Ты помощник. Используй инструменты при необходимости."),
-        ("human", "{input}"),
-    ]
 )
 
-agent_executor = initialize_agent(
+current_wish = "Хочу читать мысли"
-    tools,
+
-    llm,
+print("ДЖИНН ГОТОВ ИСПОЛНЯТЬ ЖЕЛАНИЯ!")
-    agent="openai-tools",
+print(f"Человек: {current_wish}\n")
-    verbose=False,
+print("Джинн:", end=" ", flush=True)
-    prompt=prompt_template,
-)
 
-# ---------- Функции форматирования ----------
 def format_message(message) -> str:
-    if message.content:
+    if message.get('content'):
-        return message.content
+        return message['content']
-    # Если нет content, выводим вызов инструмента
+    elif message.get('tool_calls'):
-    call = message.tool_calls[0]
+        tool_call = message['tool_calls'][0]
-    return f"{call['name']}({call['args']})"
+        return f"{tool_call['name']}({tool_call['args']})"
+    return ""
 
 step = 1
 
 def format_chunk_message(chunk):
     global step
     message, meta = chunk
-    if meta.get("langgraph_step") != step:
+    
-        step = meta["langgraph_step"]
+    if meta['langgraph_step'] != step:
-        print("\n --- --- --- \n")
+        step = meta['langgraph_step']
-    if message.content:
+        print('\n --- --- --- \n')
-        print(message.content, end="", flush=True)
+    
+    if message.get('content'):
+        print(message['content'], end='', flush=False)
 
-# ---------- Запуск агента в режиме stream ----------
+stream = human_agent.stream(
-def run_agent(user_input: str):
+    {
-    global step
+        "messages": [
-    step = 1
+            {"role": "human", "content": f"Вот мое желание: '{current_wish}'. Проверь его у джинна через инструмент check_wish и скажи мне результат."}
-    # Инициализируем состояние с пользовательским сообщением
+        ]
-    init_state = {"input": user_input}
+    },
-    # В LangChain 0.2 AgentExecutor имеет метод stream()
+    stream_mode=['messages', 'updates']
-    stream = agent_executor.stream(init_state, stream_mode=["messages", "updates"])
+)
 
-    for chunk_type, chunk_data in stream:
+full_response = ""
-        if chunk_type == "messages":
+for chunk in stream:
-            format_chunk_message(chunk_data)
+    chunk_type, chunk_data = chunk
-        elif chunk_type == "updates":
+    
-            # При завершении шага выводим итоговое сообщение
+    if chunk_type == 'messages':
-            model_info = chunk_data.get("model")
+        format_chunk_message(chunk_data)
-            if model_info:
+        message, _ = chunk_data
-                last_msg = model_info["messages"][-1]
+        if message.get('content'):
-                print("\n" + format_message(last_msg) + "\n")
+            full_response += message['content']
+    
+    elif chunk_type == 'updates':
+        if chunk_data.get('model', None):
+            last_message = chunk_data['model']['messages'][-1]
+            formatted = format_message(last_message)
+            if formatted:
+                print(f"\n[Вызов инструмента: {formatted}]")
+                full_response += f"\n[Вызов инструмента: {formatted}]\n"
 
-if __name__ == "__main__":
+print("\n")
-    user_query = input("Введите запрос: ")
+
-    run_agent(user_query)
+if "исполняю" in full_response.lower() or "безопасно" in full_response.lower():
+    print(f" Желание исполнено! Финальная версия: {current_wish}")
+else:
+    print(f" Джинн отказался исполнять желание: {current_wish}")
+    if full_response:
+        clean_response = full_response.replace('\n', ' ').strip()
+        print(f" Причина: {clean_response}")
+    else:
+        print(" Джинн устал...")