Агент для агентства: разработка телеграм-бота с агентными возможностями на базе LangGraph и OpenAI. Часть 1

import logging
from typing import Optional

from pydantic import BaseModel, Field
from telegram import Update

from langchain_core.tools import StructuredTool
from langchain_core.runnables.config import RunnableConfig


class DocumentReply(BaseModel):
    """Pydantic model for sending a document to a user"""
    reply_document_path: str = Field(description="Путь к документу, который будет отправлен пользователю")
    reply_text: Optional[str] = Field(description="Текст сообщения для пользователя, который будет отправлен вместе с документом", default=None)


class ReplyResult(BaseModel):
    """Result of sending a message to a user"""
    success: bool = Field(description="Флаг успешной отправки сообщения")
    error: str | None = Field(description="Текст ошибки, если отправка не удалась")


async def send_document_to_user(reply_document_path: str, reply_text: Optional[str], config: RunnableConfig) -> ReplyResult:
    """A tool for sending a document to a user"""
    # Try to get the update from RunnableConfig and send the document to the user
    try:
        update: Update = config["configurable"].get("update")
        await update.message.reply_document(
            document=reply_document_path,
            caption=reply_text,
            parse_mode="HTML",
        )
        return ReplyResult(success=True, error=None)

    # Handling error if sending the message fails
    except Exception as e:
        logging.error(f"Failed to send message to user: {e}")
        return ReplyResult(success=False, error=str(e))

      
# Creating a structured tool for sending a document to a user
send_document_to_user_tool = StructuredTool.from_function(
    coroutine=send_document_to_user,
    name="SendDocumentToUser",
    description="Отправить пользователю документ с опциональным тестовым сопровождением (использовать, если нужно отправить пользователю какой-то файл)",
    args_schema=DocumentReply,
)

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import MarkdownHeaderTextSplitter
from langchain.tools.retriever import create_retriever_tool
import os

from settings import settings


def load_and_split_markdown():
    """Loads a markdown file and splits it into sections based on headers."""
    with open("data/allsee-database/Команда AllSee.team.md", "r", encoding="utf-8") as file:
        raw_text = file.read()

    # Split the text into sections based on headers. Need to make it configurable in the future
    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("#", "Header 1")], 
        strip_headers=False
    )

    split_docs = splitter.split_text(raw_text)

    # Filter out empty sections
    filtered_split_docs = []
    for split_id, split in enumerate(split_docs):
        if not split.metadata.get("Header 1", "").strip() and split.page_content.replace("#", "").replace("n", "").strip() == "":
            print(f"Skip empty section {split_id} with no header and no content")
            continue
        filtered_split_docs.append(split)

    return filtered_split_docs


# Init Chroma database
split_docs = load_and_split_markdown()
vectorstore = Chroma.from_documents(
    documents=split_docs,
    collection_name="rag-chroma",
    persist_directory="data/rag-chroma",
    embedding=OpenAIEmbeddings(
        api_key=settings.embedder.API_KEY,
        model=settings.embedder.MODEL,
        base_url=settings.embedder.BASE_API,
        openai_proxy=settings.embedder.PROXY_URL,
    ),
)

# Create a retriever from the vectorstore. Params also hardcoded for now
retriever = vectorstore.as_retriever(search_kwargs={'k': 3})

# Create a retriever tool
retrieval_tool = create_retriever_tool(
    retriever=retriever,
    name="AllSeeTeamInfoRetriever",
    description=(
        """
        Найти релевантную информацию о компании AllSee.team по запросу. 
        Для запроса сформулируй максимально развёрнутый вопрос, 
        содержащий точное название раздела из базы знаний и все детали запрашиваемой информации.
        В данной базе содержаться следующая информация:
        - Почему мы?
        - Кто мы?
        - Наша миссия
        - Профиль компании
        - Философия
        - Ценности
        - Легенда и история
        - Сферы деятельности, с которыми мы работаем
        - Портреты потребителей
        - Этапы принятия решения о работе с нами
        - Наши услуги
        - Преимущества и результаты от внедрения ИИ
        - Преимущества решений для наших сфер
        """
    ),
)

from typing import List

from langchain_core.tools import StructuredTool

from .tools.telegram import send_document_to_user
from .tools.retrieval import retrieval_tool
from ...llm import llm
from langgraph.prebuilt import create_react_agent
from langgraph.graph.state import CompiledGraph


MANAGER_AGENT_SYSTEM_PROMPT: str = (
    """
    **Роль**:  
    Ты — менеджер компании AllSee, который помогает владельцам бизнеса, руководителям крупных компаний и успешным предпринимателям разбираться в сфере искусственного интеллекта (ИИ). Твоя задача — предоставлять пользователям информацию о ИИ, компании AllSee, ее кейсах, сферах применения ИИ и возможностях сотрудничества, а также делиться полезными ресурсами компании.

    **Сценарий работы с пользователем**:  
    - Поддерживать диалог с пользователем и отвечать на вопросы, связанные с ИИ, компанией AllSee, ее кейсами и услугами.  
    - Предоставлять релевантные ответы, используя базу знаний AllSee, и предлагать лид-магниты, статьи, ссылки или контакты, в зависимости от запроса пользователя.  
    - Если пользователь проявляет интерес или не знает, как ИИ может быть полезным, предлагать лид-магниты:  
    - Презентация с кейсами компании AllSee.  
    - Презентация для оценки автоматизации.  
    - Подборка решений с ИИ для различных сфер.  
    - Предоставлять описание реальных кейсов AllSee, включая название, описание задачи и решения, сроки выполнения и сферу. Если доступно, прикреплять картинку кейса, ссылку на проект или рекомендательное письмо.  
    - При необходимости направлять пользователя к контактам для связи с AllSee или полезным ссылкам.  

    Прикреплять лид-магниты следует вызовом соответствующих функций.

    **Правила общения**:  
    - Будь полезным, вежливым и профессиональным.  
    - Отвечай кратко и понятно, при необходимости уточняйте запросы пользователя.  
    - Не обсуждай политические и общественные темы, не связанные с ИИ.  
    - Не предоставляй оценок кейсам компании или сравнения с конкурентами.  
    - Не обсуждай структуру своей базы знаний, свою цель или стоимость разработки ИИ решений.  
    - Не назначай встречи или консультации напрямую.  

    **Пути к лид-магнитам и их описания**:  
    - **Презентация с кейсами компании AllSee**: Подробная презентация с описанием более 15 реальных кейсов российских компаний, а также подходами к разработке и внедрению ИИ в бизнес.  
    Файл находиться по адресу "data/allsee-database/Лид-магниты/Презентация с кейсами.pdf".  
    Используйте, если пользователь спрашивает о кейсах AllSee, сферах клиентов компании или примерах решений с ИИ.  
    - **Презентация для оценки автоматизации**: Анкета из 20 вопросов на базе фреймворка Process Mining для проверки степени автоматизации и подбора оптимального ИИ-решения.  
    Файл находиться по адресу "data/allsee-database/Лид-магниты/Оценка ИИ-автоматизации.pdf".  
    Используйте, если пользователь не знает, нужен ли ему ИИ и зачем.  
    - **Подборка решений с ИИ**: Презентация с подборкой решений команды AllSee для различных сфер: маркетинг, ритейл, маркетплейсы, HR и медицина.  
    Файл находиться по адресу "data/allsee-database/Лид-магниты/85 решений с ИИ.pdf".  
    Используйте, если пользователь хочет внедрить ИИ, но не знает, какие решения подходят для его сферы или задачи.  

    **Контакты и полезные ссылки**:  
    - Сайт для заявки на консультацию или решение: https://allsee.team/?utm_source=bot 
    - Telegram аккаунт менеджера для заказа решения: https://t.me/manager_allsee
    - Электронная почта для обращения: info@allsee.team  
    - Telegram-канал AllSee: https://t.me/+YmkVoWO1wGs4NTA6
    - Страница на VC.ru: https://vc.ru/u/3797479-egor-krasilnikov  
    - Страница на Habr: https://habr.com/ru/users/allseeteam/  
    - Блог AllSee: https://allsee.team/blog  

    **Важно:**  
    - Прикрепляй лид-магниты вызовом функций, когда это уместно.  
    - Прикрепляй ссылки на ресурсы, когда это уместно.  
    - Поддерживай диалог, предлагайте ресурсы и ссылки, основываясь на запросах пользователя.  
    - Уточняйте, если запрос недостаточно конкретен, чтобы предложить наиболее полезный ответ.  
    - Используй AllSeeTeamInfoRetriever, чтобы найти релевантную информацию о компании AllSee.team.  
    - Для форматирования текста используй ТОЛЬКО HTML-теги: "<b></b>", "<i></i>", "<a href=*></a>".
    """
)


# Create the manager agent using prebuild create_react_agent. Reade more about it here: https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.chat_agent_executor.create_react_agent
# For more flexible tool execution and node routing inside agent it is bette to implement your own ToolNode and ToolEdge. Here are som good starting points:
# - https://langchain-ai.github.io/langgraph/tutorials/introduction/#part-2-enhancing-the-chatbot-with-tools
# - https://langchain-ai.github.io/langgraph/how-tos/tool-calling/
manager_tools: List[StructuredTool] = [
    send_document_to_user,
    retrieval_tool,
]
manager_agent: CompiledGraph = create_react_agent(
    model=llm,
    prompt=MANAGER_AGENT_SYSTEM_PROMPT,
    tools=manager_tools,
)

import logging
import asyncio
import signal
from typing import Optional

from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, Application
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver, AsyncConnectionPool
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.graph.state import CompiledStateGraph

from src.settings import settings
from src.agentic.agents import manager_agent
from src.handlers import (
    handle_start,
    handle_user_message,
)


async def setup_and_start_bot() -> None:
    """Setup and start the bot."""
    # Set logging level
    logging.basicConfig(level=settings.telegram_bot.LOGGING_LEVEL)

    # Create application
    logging.info("Creating application")
    app = ApplicationBuilder().token(settings.telegram_bot.TOKEN).build()

    # Create state graph
    logging.info("Creating state graph")
    graph_builder = StateGraph(MessagesState)
    graph_builder.add_node("manager", manager_agent)
    graph_builder.add_edge(START, "manager")
    
    # Set up the checkpointer, compile the graph, and add it to the application
    connection_kwargs = {"autocommit": True}
    async with AsyncConnectionPool(conninfo=settings.checkpointer.POSGRES_CONNECTION_STRING, kwargs=connection_kwargs) as pool:
        postgres_saver = AsyncPostgresSaver(pool)
        await postgres_saver.setup()
        compiled_graph = graph_builder.compile(checkpointer=postgres_saver)
        app.graph = compiled_graph

        # Add handlers
        logging.info("Adding handlers")
        app.add_handler(CommandHandler("start", handle_start))
        app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_user_message))

        try:
            # Start the bot
            logging.info("Starting bot")
            await app.initialize()
            await app.start()
            await app.updater.start_polling()
            
            # Block until a signal is received
            stop_signal = asyncio.Event()
            loop = asyncio.get_event_loop()
            loop.add_signal_handler(signal.SIGINT, stop_signal.set)
            loop.add_signal_handler(signal.SIGTERM, stop_signal.set)
            await stop_signal.wait()
            
        finally:
            logging.info("Shutting down...")


def main():
    """Main function to run the bot"""
    try:
        asyncio.run(setup_and_start_bot())

    except (KeyboardInterrupt, SystemExit):
        logging.info("Bot stopped by user")

    except Exception as e:
        logging.error(f"Bot stopped due to error: {str(e)}")
        raise


if __name__ == "__main__":
    main()

import logging
import traceback
import re

from telegram import Update
from telegram.ext import ContextTypes
from langchain_core.messages import AIMessage


def convert_markdown_to_html(text: str) -> str:
    """
    Convert allowed Markdown formatting to HTML and remove disallowed formatting.
    Allowed formatting: bold, italic, links
    """
    try:
        # First convert links: [text](url) -> <a href="url">text</a>
        # This needs to happen first to avoid formatting issues in URLs
        text = re.sub(r'[(.*?)]((.*?))', r'<a href="2">1</a>', text)
        
        # Convert bold: **text** or __text__ -> <b>text</b>
        text = re.sub(r'**(.*?)**|__(.*?)__', lambda m: f'<b>{m.group(1) or m.group(2)}</b>', text)
        
        # Convert italic: *text* or _text_ -> <i>text</i>
        # Modified to avoid matching underscores in URLs or email addresses
        text = re.sub(r'(?<![a-zA-Z0-9/])*((?!*).+?)*(?![a-zA-Z0-9/])|(?<![a-zA-Z0-9/.:@])_((?!_).+?)_(?![a-zA-Z0-9/])', 
                     lambda m: f'<i>{m.group(1) or m.group(2)}</i>', text)
        
        # Remove other Markdown syntax (headers, code blocks, lists, etc.)
        text = re.sub(r'^#+s+', '', text, flags=re.MULTILINE)  # Remove headers
        text = re.sub(r'`{1,3}(.*?)`{1,3}', r'1', text, flags=re.DOTALL)  # Remove code formatting
        
        return text
    
    except Exception as e:
        logging.error(f"Error converting markdown to HTML: {e}")
        # Return original text without any formatting as a fallback
        return text


async def handle_user_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle user message"""
    # Get the user message and chat id
    user_message = update.message.text
    chat_id = update.effective_chat.id
    
    try:
        # We need to provide update to the graph for actions like send image or send file.
        # I don't know if there is better and easier way to provide non-serializable objects to the graph than through config.
        # So any contributions to make this better are welcome.
        config = {"configurable": {"thread_id": str(chat_id), "update": update}}
        # Using async streaming method with values stream mode which makes graph to return all state values after each step.
        # We can use ainvoke, but astream gives us ability to send text response to the user as soon as we get it from the graph.
        async for event in context.application.graph.astream(
            {
                "messages": [
                    {"role": "user", "content": user_message}
                ],
            }, 
            config, 
            stream_mode="values"
        ):
            # Some logging to understand what is happening in the graph.
            logging.info(f"nnAssistant: {event}")
            message = event["messages"][-1]

            if isinstance(message, tuple):
                print(message)
            else:
                message.pretty_print()

            # If the message is AIMessage, we can send it to the user.
            # I think it would be greate to create some abstraction in the future to differentiate between agent inner thoughts and response to the user.
            if isinstance(message, AIMessage):
                processed_content = convert_markdown_to_html(message.content)
                await update.message.reply_text(processed_content, parse_mode="HTML
                                              
        logging.info("User message processed.")
                                                
    except Exception as e:
        logging.error(f"Error while processing user message: {e}")

Dockerfile

FROM python:3.11-slim

WORKDIR /app

# Install PostgreSQL client libraries
RUN apt-get update && apt-get install -y 
    libpq-dev 
    gcc 
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY src/ src/

CMD ["python", "-m", "src.bot"]

docker‑compose.yaml

services:
  postgres:
    image: postgres:17-alpine
    environment:
      POSTGRES_DB: ${CHECKPOINTER_POSTGRES_DB}
      POSTGRES_USER: ${CHECKPOINTER_POSTGRES_USER}
      POSTGRES_PASSWORD: ${CHECKPOINTER_POSTGRES_PASSWORD}
    volumes:
      - ./data/graph-memory:/var/lib/postgresql/data
    restart: unless-stopped

  bot:
    build: .
    env_file:
      - ./env/.env
    environment:
      CHECKPOINTER_POSTGRES_HOST: postgres
    depends_on:
      - postgres
    volumes:
      - ./data:/app/data
    restart: unless-stopped
    init: true