Agent sterowany przez WhatsApp z wyszukiwaniem w sieci

Zbudowałem ten prototyp, aby sprawdzić, jak połączyć WhatsApp, transkrypcję głosu, model językowy i wyszukiwanie w sieci. To eksperyment, nie gotowy system produkcyjny. Prawdziwe wdrożenie wymagałoby dodatkowo silniejszego uwierzytelniania, limitów ruchu, monitoringu i jasnych zasad przechowywania danych.

Jak działa rozwiązanie

Serwer FastAPI odbiera webhook z WhatsApp. Dla wiadomości tekstowej używa treści bezpośrednio. Dla wiadomości głosowej pobiera plik i wysyła go do transkrypcji. Następnie mały klasyfikator decyduje, czy odpowiedź wymaga aktualnych informacji z sieci.

Przepływ wygląda tak:

1. odebranie webhooka,
2. rozpoznanie rodzaju wiadomości,
3. opcjonalna transkrypcja dźwięku,
4. decyzja o wyszukiwaniu,
5. przygotowanie krótkiej odpowiedzi,
6. wysłanie jej przez WhatsApp Cloud API.

Zależności i konfiguracja

1pip install fastapi uvicorn requests python-dotenv openai duckduckgo_search ollama

Sekrety oraz wersję API przechowujemy w .env:

1OPENAI_API_KEY=YOUR_OPENAI_API_KEY
2WHATSAPP_PHONE_NUMBER_ID=YOUR_PHONE_NUMBER_ID
3WHATSAPP_API_TOKEN=YOUR_API_TOKEN
4WEBHOOK_VERIFY_TOKEN=YOUR_VERIFY_TOKEN
5WHATSAPP_GRAPH_VERSION=YOUR_SUPPORTED_GRAPH_API_VERSION

Wersji Graph API nie wpisujemy na stałe w kodzie, ponieważ Meta regularnie ją zmienia.

Podstawowa aplikacja FastAPI

1import os
2import logging
3import requests
4
5from dotenv import load_dotenv
6from fastapi import FastAPI, Request
7from openai import OpenAI
8
9load_dotenv()
10
11app = FastAPI()
12client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
13logger = logging.getLogger(__name__)
14
15phone_number_id = os.getenv("WHATSAPP_PHONE_NUMBER_ID")
16graph_version = os.getenv("WHATSAPP_GRAPH_VERSION")
17whatsapp_token = os.getenv("WHATSAPP_API_TOKEN")
18verify_token = os.getenv("WEBHOOK_VERIFY_TOKEN")
19
20messages_url = (
21    f"https://graph.facebook.com/{graph_version}/"
22    f"{phone_number_id}/messages"
23)

Weryfikacja webhooka

Meta wywołuje endpoint GET /callback, przekazując token i wartość hub.challenge.

1@app.get("/callback")
2async def verify_callback(request: Request):
3    mode = request.query_params.get("hub.mode")
4    token = request.query_params.get("hub.verify_token")
5    challenge = request.query_params.get("hub.challenge")
6
7    if mode == "subscribe" and token == verify_token:
8        return int(challenge)
9
10    return {"error": "verification failed"}

W wersji produkcyjnej odpowiedzi błędne powinny mieć właściwe kody HTTP, a webhook POST powinien również weryfikować podpis żądania.

Pobieranie i transkrypcja dźwięku

WhatsApp przekazuje identyfikator pliku. Najpierw wymieniamy go na adres pobrania, a potem pobieramy zawartość z tym samym tokenem autoryzacyjnym.

1def download_media(media_id: str) -> bytes:
2    headers = {"Authorization": f"Bearer {whatsapp_token}"}
3    media_url = f"https://graph.facebook.com/{graph_version}/{media_id}"
4
5    metadata_response = requests.get(media_url, headers=headers, timeout=20)
6    metadata_response.raise_for_status()
7
8    download_url = metadata_response.json()["url"]
9    media_response = requests.get(download_url, headers=headers, timeout=30)
10    media_response.raise_for_status()
11    return media_response.content

Transkrypcja może korzystać z API audio OpenAI:

1def transcribe_audio(audio_path: str) -> str:
2    with open(audio_path, "rb") as audio_file:
3        transcription = client.audio.transcriptions.create(
4            model="whisper-1",
5            file=audio_file,
6        )
7
8    return transcription.text

Plik tymczasowy należy usuwać również w przypadku błędu. W produkcji trzeba też ustalić maksymalny rozmiar nagrania i czas przechowywania.

Kiedy wyszukiwać w sieci

Klasyfikator powinien zwracać prostą strukturę, np. search_required, search_query oraz answer. Jeśli wynik nie daje się poprawnie odczytać, bezpieczniej wykonać wyszukiwanie lub poprosić użytkownika o doprecyzowanie niż udzielić pewnej, ale nieaktualnej odpowiedzi.

Wyniki wyszukiwania trzeba traktować jak niezaufane dane. Nie powinny mieć możliwości zmieniania instrukcji systemowych ani uruchamiania dowolnych działań.

Wysyłanie odpowiedzi

1def send_message(recipient: str, text: str) -> None:
2    headers = {
3        "Authorization": f"Bearer {whatsapp_token}",
4        "Content-Type": "application/json",
5    }
6    payload = {
7        "messaging_product": "whatsapp",
8        "to": recipient,
9        "type": "text",
10        "text": {"preview_url": False, "body": text},
11    }
12
13    response = requests.post(
14        messages_url,
15        json=payload,
16        headers=headers,
17        timeout=20,
18    )
19    response.raise_for_status()

Co trzeba dodać przed produkcją

Prototyp pokazuje przepływ, ale nie rozwiązuje wszystkich problemów operacyjnych. Przed udostępnieniem użytkownikom należy dodać:

• weryfikację podpisów webhooków,
• kontrolę dostępu i limity żądań,
• obsługę ponowień oraz duplikatów zdarzeń,
• filtrowanie danych wrażliwych w logach,
• monitoring kosztów i błędów,
• politykę retencji wiadomości i plików audio,
• testy dla nieobsługiwanych typów wiadomości.

Najważniejsza lekcja z tego eksperymentu jest prosta: połączenie kilku API zajmuje niewiele kodu, ale niezawodny proces powstaje dopiero po dopracowaniu błędów, bezpieczeństwa i obserwowalności.