Kendi kendine barındırılan Telegram botları

Note

.NET'daki Telegram botları için kendi kendine barındırma yardımcıları yakında geliyor.

Note

Telegram botları için kendi kendine barındırma yardımcıları şu anda Go'da kullanılamıyor.

agent-framework-hosting-telegram Telegram Bot API güncelleştirmelerini Agent Framework çalıştırma değerlerine dönüştürür ve son veya akış çalıştırmalarını Bot API işlemleri olarak işler. Bot istemcisi, yoklama altyapısı, webhook yönlendiricisi, komut kayıt sistemi veya iletim altyapısı sağlamaz.

pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-telegram azure-identity

Bir güncelleştirme yükü sağlayabilen ve yardımcılar tarafından döndürülen işlemleri yürütebilen herhangi bir Telegram istemci kitaplığını kullanın. Örnekte aiogram kullanılır, ancak yardımcılar buna bağlı değildir.

Güncelleştirmeyi işleme

aiogram webhook örneği, Telegram'ın gizli başlığını doğrular, güncellemeyi ilgili işleyiciye yönlendirir ve her özel sohbet veya paylaşılan grup sohbeti için bir ajan oturumunu korumak amacıyla bot kapsamındaki bir oturum kimliği kullanır.

async def handle_update(update: Mapping[str, Any]) -> None:
    """Process one Telegram update through the sample agent."""
    callback_query_id = telegram_callback_query_id(update)
    if callback_query_id is not None:
        await bot.answer_callback_query(callback_query_id=callback_query_id)

    chat_id = telegram_chat_id(update)
    session_id = telegram_session_id(update, bot_id=bot.id)
    if chat_id is None or session_id is None:
        return

    # Background webhook tasks may overlap. Serialize each chat so /new cannot
    # delete a session while an earlier response is still updating it.
    async with session_locks.setdefault(session_id, asyncio.Lock()):
        if (command := telegram_command(update)) is not None and await handle_command(update, command):
            return

        async def resolve_file_url(file_id: str) -> str | None:
            file = await bot.get_file(file_id)
            if file.file_path is None or (file.file_size is not None and file.file_size > MAX_MEDIA_BYTES):
                return None
            destination = BytesIO()
            await bot.download_file(file.file_path, destination=destination)
            data = destination.getvalue()
            if len(data) > MAX_MEDIA_BYTES:
                return None
            encoded = base64.b64encode(data).decode("ascii")
            return f"data:application/octet-stream;base64,{encoded}"

        try:
            run = await telegram_to_run(update, resolve_file_url=resolve_file_url, stream=True)
        except ValueError:
            LOGGER.debug("Ignoring non-actionable Telegram update", exc_info=True)
            return

        await bot.send_chat_action(chat_id=chat_id, action="typing")
        placeholder = await bot.send_message(chat_id=chat_id, text=PLACEHOLDER_TEXT)

        target = await state.get_target()
        # Reuse one AgentSession per Telegram chat. The /new command removes this
        # mapping so get_or_create_session creates a clean session next time.
        session = await state.get_or_create_session(session_id)
        stream = target.run(
            run["messages"],
            stream=True,
            session=session,
            options=run["options"],
        )
        if not isinstance(stream, ResponseStream):
            raise RuntimeError("agent did not return a response stream")

        last_edit_at = 0.0
        async for operation in telegram_from_streaming_run(
            stream,
            chat_id=chat_id,
            message_id=placeholder.message_id,
            initial_text=PLACEHOLDER_TEXT,
        ):
            if operation["method"] == "editMessageText":
                delay = EDIT_INTERVAL_SECONDS - (time.monotonic() - last_edit_at)
                if delay > 0:
                    await asyncio.sleep(delay)
                last_edit_at = time.monotonic()
            await execute_operation(operation)

        # Persist the updated AgentSession back under the stable per-chat key after
        # streaming has finalized and the history provider has recorded the turn.
        await state.set_session(session_id, session)

Yoklama ve web kancası kurulumu, komut işleme, gelen medya ilkesi, akış düzenlemeleri ve üretim dağıtımı kılavuzu için yerel Telegram örneğine bakın.

Important

Güncelleştirmeleri işlemeden önce Telegram web kancası teslimlerini doğrulayın. Bir webhook gizli anahtarı, teslimatın Telegram’dan geldiğini doğrular; ancak Telegram kullanıcısını veya sohbeti uygulama verilerine erişmek için yetkilendirmez. Uygulamanız yetkilendirme politikasını uygulayana kadar sohbet ve kullanıcı kimliklerini güvenilir olmadığını varsayın.

Sonraki Adımlar

A2A Ekle

Daha derine gidin:

  • Last updated on