释放无限潜能：使用 Gpt-4o 打造革命性的 Whatsapp Ai 代理！

Created with DALL-E

如何使用Meta API构建自己的LLM驱动的WhatsApp聊天机器人

在人工智能和商业管理领域，人工智能代理与广泛使用的通信工具的集成是一个颠覆性的变化。想象一下，拥有一个熟悉的聊天界面，可以进行实时数据请求、更新和任务自动化，这一切都通过与您业务的管理或个人助理人工智能的WhatsApp直接互动实现。

在我们关于创建人工智能驱动的商业管理者系列的第三部分中，我将带您了解如何将您的人工智能代理连接到WhatsApp，以增强其能力和覆盖范围。目标是实现一个能够与您所有相关数据库表进行交互，甚至能够独立创建表和所有必要工具的人工智能助手。作为主要展示，我专注于跟踪费用、发票等商业用例。然而，您可以轻松地调整相同的逻辑来创建一个个人助理，帮助您跟踪任务、项目和想法。

这是我系列的第三部分。在我们开始之前，对于所有等待的人，我为漫长的延迟表示歉意。在过去的几个月里，我忙于开始新的人工智能软件工程工作，并适应新的工作与生活平衡。我已经准备了一些未来部分的内容，我们将探索代理工作流中的重大变化，以及包含几个附加功能的更复杂的工作流。在前两篇文章中使用的一些变通方法在当时是为了可靠的工具调用而必要的，但由于更高效的模型如GPT-4o和GPT-4o-mini，现在已不再需要。如果您是工具调用和代理工作流开发的新手，我仍然建议您从前两部分开始。我发现了解如何从头开始构建某些东西是有用的，然后再依赖于像LangChain或更具体的LangGraph这样的框架来实现深度可定制的代理工作流（我将在不久的将来介绍）。

现在，我们必须退后一步，首先关注基础设施。我认为在大多数项目中，特别是在人工智能软件项目中，最初创建一个可工作的端到端产品是良好的实践，以免在功能膨胀中迷失。我经常发现自己过度思考初始设计选择，并在脑海中开发出过于复杂的产品。为了解决这个问题，专注于在几天的开发时间内构建一个可工作的端到端产品确实有助于建立一个明确的基础。之后，您将知道哪些功能需要优先考虑，并能够收集初步反馈。这启动了增量开发过程，这始终是我在承诺一个项目时的目标。

之前部分的回顾

我们在本系列的早期章节中为我们的人工智能驱动的业务经理奠定了基础：

• 第 1 部分：目标是开发一个原型代理工作流，可以与工具对象交互，并减少底层语言模型（LLM）生成的工具参数中的幻觉。
• 第 2 部分：我们专注于定义基本功能，组织项目仓库，并使用 SQLModel 构建数据库模式。我们还添加了用于数据添加和查询的 SQL 工具，并更新了 Tool 类以适应 SQLModel 对象。此外，我们展示了一个 TaskAgent 类，用于自动化代理启动，并设置了一个 OpenAIAgent 以实现上下文感知的工具使用。

本文的范围

如往常一样，让我们开始定义本文的范围：

1. 使用MetaAPI将人工智能代理与WhatsApp集成 设置和配置MetaAPI以实现WhatsApp Business集成。确保人工智能代理能够通过WhatsApp发送和接收消息。
2. 设置Webhook并使用Ngrok在本地运行 创建一个FastAPI应用程序来处理来自WhatsApp的Webhook事件。使用ngrok将本地FastAPI服务器暴露到互联网。配置MetaAPI将Webhook事件发送到ngrok URL。

由于我们正在向可部署的服务器迈进，我们还需要调整我们的项目架构。我们本质上是在实施一个FastAPI服务器，因此，我首选的存储库结构是领域驱动设计（DDD），或者更倾向于DDD。（您可以在这里查看存储库结构）

1. 设置 WhatsApp 云API

首先，您需要熟悉 Meta 提供的云API。您可以使用像 Twilio 这样的 SaaS 产品实现相同的结果，它们提供更用户友好的集成。然而，由于最近的数据泄露以及出于成本效率的考虑，我更倾向于使用 Meta 提供的根API。

1.1 前提条件

1. 在开始之前，您需要按照以下步骤注册一个Meta开发者账户：如何开设Meta开发者账户。在注册过程中，您需要使用手机号码进行验证。请注意，这将不是您最终WhatsApp客户端的手机号码。相反，您将获得一个由平台分配的测试手机号码，之后可以更改为其他手机号码。
2. 注册后，前往您的仪表板并创建一个应用程序。
3. 此外，您还需要一个与您的真实账户关联的Meta商业账户（MBA），或者您可以创建一个新的账户以链接到您的MBA。您也可以跳过此步骤，因为在下一步中系统会自动提示您链接或创建MBA。

1.2 将WhatsApp产品添加到您的应用程序

在您创建了一个Meta开发者账户中的应用程序后，系统会要求您向其添加产品。在这里，您需要选择WhatsApp并按照设置流程进行。如果您尚未完成，请创建一个Meta商业账户。一旦完成，您将拥有一个测试的WhatsApp商业账户和一个测试电话号码。

1.3 添加接收号码

1. 在应用仪表板的左侧菜单中，导航到 WhatsApp > API Setup。在 发送和接收消息 下，选择 To 字段并选择 管理电话号码列表。在这里，您可以添加一个允许从您的测试电话号码发送和接收消息的电话号码。理想情况下，这应该是您自己的电话号码，因为您想测试您的应用程序。在将此 WhatsApp API 账户链接到真实号码之前，您最多只能添加 5 个接收号码。
2. 在 WhatsApp > API Setup 中，您现在可以通过在 from 字段中填写您的测试电话号码，在 to 字段中填写您的接收号码（您自己的电话号码）来发送测试消息。
3. 生成访问令牌。这是您的 WHATSAPP_API_TOKEN，我们将在第 6 步中需要它。

Screenshot ngrok (作者提供的图片)

我们已成功设置所需的云API。在下一步中，我们将创建一个 Webhook，以便与我们的人工智能助手应用程序进行通信。

为此，我们需要在我们的后端应用程序中创建并提供一个端点。这意味着我们的 Python 后端必须通过 URL 可访问。此 URL 将充当 Webhook 端点，供人工智能助手调用以发送和接收数据。

2. 创建一个 FastAPI 端点

为了被 Webhook 接受，我们的根端点必须验证一个特定的 GET请求，该请求将在添加我们的 URL 时由 webhook 发送。webhook 将发送三个查询参数：

• hub.mode
• hub.challenge
• hub.verify.token

验证令牌是在云API中创建 webhook 时定义的。您的后端应该验证该令牌是否与您定义的匹配，并将 hub.challenge 对象作为响应返回。确保使用以下命令安装 FastAPI 和 Uvicorn：

pip install fastapi uvicorn

2.1 创建 main.py

创建一个名为 main.py 的文件，内容如下：

from fastapi import FastAPI, Query, HTTPException

VERIFICATION_TOKEN = "abcdefg12345"

app = FastAPI()

@app.get("/")
def verify_whatsapp(
    hub_mode: str = Query("subscribe", description="Webhook 的模式", alias="hub.mode"),
    hub_challenge: int = Query(..., description="用于验证 webhook 的挑战", alias="hub.challenge"),
    hub_verify_token: str = Query(..., description="验证令牌", alias="hub.verify_token"),
):
    if hub_mode == "subscribe" and hub_verify_token == VERIFICATION_TOKEN:
        return hub_challenge
    raise HTTPException(status_code=403, detail="无效的验证令牌")

@app.get("/health")
def health():
    return {"status": "healthy"}

@app.get("/readiness")
def readiness():
    return {"status": "ready"}

在第三行中，您可以定义一个 VERIFICATION_TOKEN，该令牌稍后将由 webhook 用于验证后端是否在您的控制之下。在这种情况下，我们将其定义为 "abcdefg12345"，但您可以定义自己的自定义令牌。

2.2 运行应用程序

使用 Uvicorn 运行应用程序：

uvicorn main:app --reload

2.3 本地提供您的API

您的后端现在在 http://localhost:8000 和/或 http://127.0.0.1:8000 上运行。

我们现在提供以下端点：

• 验证WhatsApp Webhook: http://127.0.0.1:8000/?hub.mode=subscribe&hub.challenge=1234&hub.verify_token=abcdefg12345
• 健康端点: http://127.0.0.1:8000/health
• 准备就绪端点: http://127.0.0.1:8000/readiness

您可以使用健康端点检查您的应用程序是否正在运行。在浏览器中打开 http://127.0.0.1:8000/health，您应该看到: {"status": "healthy"}

3. 使用 Ngrok 运行代理服务器

由于我们的服务器在本地运行，WhatsApp Webhook 无法调用端点进行验证。我们需要的是一个可以被 webhook 使用的公共 URL。有两个选择：将应用程序部署到云服务器或创建一个代理服务器隧道。由于我们仍在开发过程中，我们将使用第二个选项。

1. 前往 ngrok 注册 并创建一个免费账户。

2. 在本地安装 ngrok。根据您的系统，您可以使用 Brew、Chocolatey，或简单地下载并安装。请参见：设置与安装。

3. 安装完成后，在终端中使用以下命令添加您的身份验证代码。将 $YOUR-AUTHENTICATION_TOKEN 替换为您的 ngrok 身份验证令牌，可以在 ngrok 仪表板的“您的身份验证令牌”下找到。

4. 通过在终端中运行以下命令，开始从本地主机的 8000 端口转发流量：

   ngrok config add-authtoken $YOUR-AUTHENTICATION_TOKEN
   ngrok http http://localhost:8000
   

   Forwarding https://.ngrok.io -> http://localhost:8000

您的本地服务器现在可以通过 ngrok 提供的公共 URL 访问。您应该看到如下内容：

Forwarding https://.ngrok.io -> http://localhost:8000

使用 ngrok 提供的 HTTPS URL 进行 webhook 配置。

4. 实现 Webhook

现在让我们回到 Meta 的云API，以实现所需的 webhook。

1. 导航到 Meta for Developers 并选择之前创建的应用。
2. 在左侧菜单中转到 WhatsApp > Configuration。
3. 在 Webhook 部分将你的 ngrok HTTPS 转发 URL 粘贴到 Callback URL 字段，并在 Verification Token 字段中输入在 main.py 中定义的 VERIFICATION_TOKEN。
4. 点击确认并保存按钮，等待 webhook 验证你的后端。
5. 在 Webhook Fields 部分，在 Subscribed Fields 下启用 messages 切换。

就这样！你现在应该能够在你的 Python 后端服务器中接收 WhatsApp 消息。

4.1 理解 Webhook 和对话

Webhook 是 HTTP 回调，允许程序在特定事件发生时（例如新消息或状态变化）接收实时更新。Webhook 通过将包含事件数据的 HTTP 请求发送到预配置的 URL（在我们的案例中是 ngrok 代理服务器 URL），使系统集成和自动化成为可能。

要理解 Meta 宇宙中 Webhook 的逻辑和定价，了解一些关于对话的基本原则是有帮助的。

在 WhatsApp API 上，“对话”开始于：

1. 用户发送消息：这会打开一个 24 小时的窗口，在此期间，您可以回复包括文本、图像或其他媒体的消息 而无需额外费用。
2. 企业发起联系：如果最近没有收到用户消息（没有打开的 24 小时窗口），您的人工智能助手必须使用 预先批准的模板消息 来开始对话。您可以添加自定义模板，但需要得到 Meta 的批准。

只要用户持续回复，24 小时窗口将在每条新消息时重置。这使得可以在没有额外费用的情况下进行持续交互。一次对话的费用大约为 0.00–0.08 美元。具体定价基于您的对话类型（营销、实用、服务）和您的位置。仅供参考：目前服务对话似乎是免费的。您可以在此处找到具体定价：Whatsapp 定价

5. 构建接收消息端点

现在我们能够在后端接收消息。由于我们已经订阅了消息对象，每当有消息发送到您的测试号码时，Webhook 将会向您在上一步中定义的回调 URL 创建一个 POST请求。接下来我们需要做的是在我们的 FastAPI 应用程序中构建一个用于 POST请求的端点。

让我们首先定义需求：

• 返回 200 HTTP 状态码： 这对于通知 CloudAPI 消息已成功接收至关重要。未能做到这一点将导致 CloudAPI 在长达 7 天内重试发送消息。
• 提取电话号码和消息： 传入请求的有效载荷包含包括电话号码和消息的数据，我们需要在后端处理这些数据。
• 过滤传入对象： 由于 CloudAPI 可能会为同一消息发送多个事件（例如已发送、已接收和已阅读），后端需要确保只处理一条消息实例。
• 处理多种消息类型： 后端可以处理不同类型的消息，例如文本、语音消息和图像。为了不扩展文章的范围，我们将仅为图像奠定基础，但不实现到底。
• 使用 LLM-代理工作流处理： 提取的信息使用我们在本系列的前面部分开发的 LLM-代理工作流进行处理。您也可以使用其他代理实现，例如 Langchain 或 Langgraph。

5.1 定义模型和模式

我们将从一个 webhook 接收有效载荷。您可以在 Meta 的文档中找到示例有效载荷：示例有效载荷

我更喜欢使用 Pydantic 编写代码，以为我的 Python 代码添加类型安全。此外，类型注释和 Pydantic 是 FastAPI 应用程序的最佳匹配。因此，让我们首先定义在我们的端点中使用的模型：

from typing import List, Optional  
from pydantic import BaseModel, Field

class Profile(BaseModel):  
    name: str

class Contact(BaseModel):  
    profile: Profile  
    wa_id: str

class Text(BaseModel):  
    body: str

class Image(BaseModel):  
    mime_type: str  
    sha256: str  
    id: str

class Audio(BaseModel):  
    mime_type: str  
    sha256: str  
    id: str  
    voice: bool

class Message(BaseModel):  
    from_: str = Field(..., alias="from")  
    id: str  
    timestamp: str  
    text: Text | None = None  
    image: Image | None = None  
    audio: Audio | None = None  
    type: str

class Metadata(BaseModel):  
    display_phone_number: str  
    phone_number_id: str

class Value(BaseModel):  
    messaging_product: str  
    metadata: Metadata  
    contacts: List[Contact] | None = None  
    messages: List[Message] | None = None

class Change(BaseModel):  
    value: Value  
    field: str  
    statuses: List[dict] | None = None

class Entry(BaseModel):  
    id: str  
    changes: List[Change]

class Payload(BaseModel):  
    object: str  
    entry: List[Entry]

class User(BaseModel):  
    id: int  
    first_name: str  
    last_name: str  
    phone: str
    role: str

class UserMessage(BaseModel):  
    user: User  
    message: str | None = None  
    image: Image | None = None  
    audio: Audio | None = None

5.2 解析传入消息

接下来，我们将创建一些辅助函数，以便在 FastAPI 中使用依赖注入：

from app.domain import message_service

def parse_message(payload: Payload) -> Message | None:  
    if not payload.entry[0].changes[0].value.messages:  
        return None  
    return payload.entry[0].changes[0].value.messages[0]

def get_current_user(message: Annotated[Message, Depends(parse_message)]) -> User | None:  
    if not message:  
        return None  
    return message_service.authenticate_user_by_phone_number(message.from_)

def parse_audio_file(message: Annotated[Message, Depends(parse_message)]) -> Audio | None:  
    if message and message.type == "audio":  
        return message.audio  
    return None

def parse_image_file(message: Annotated[Message, Depends(parse_message)]) -> Image | None:  
    if message and message.type == "image":  
        return message.image  
    return None

def message_extractor(  
        message: Annotated[Message, Depends(parse_message)],  
        audio: Annotated[Audio, Depends(parse_audio_file)],  
):  
    if audio:  
        return message_service.transcribe_audio(audio)  
    if message and message.text:  
        return message.text.body  
    return None

• 解析有效载荷： parse_message 函数提取传入有效载荷中的第一条消息（如果存在）。如果未找到消息，该函数返回 None，以确保仅处理有效消息。
• 用户身份验证： get_current_user 函数使用 parse_message 依赖注入来提取消息，然后根据与消息关联的电话号码验证用户。在这里，我们确保只有经过身份验证的用户才能发送消息。
• 音频和图像解析： 这些函数从消息中提取音频或图像文件，如果消息类型分别为“audio”或“image”。这使得应用程序能够处理不同类型的媒体。
• 消息提取： message_extractor 函数尝试从消息中提取文本或将音频转录为文本。这确保无论消息类型如何，内容都可以被处理。

在这里，我们从我们的领域层导入了一个模块。整个脚本 message_service 是我们放置此实现的所有领域特定代码的地方，例如 authenticate_user_by_phone_number 和 transcribe_audio。

5.3 实现 POST 端点

import threading  
from typing_extensions import Annotated  
from fastapi import APIRouter, Query, HTTPException, Depends  
from app.domain import message_service  
from app.schema import Payload, Message, Audio, Image, User

@app.post("/", status_code=200)  
def receive_whatsapp(  
        user: Annotated[User, Depends(get_current_user)],  
        user_message: Annotated[str, Depends(message_extractor)],  
        image: Annotated[Image, Depends(parse_image_file)],  
):  
    if not user and not user_message and not image:  
        return {"status": "ok"}  
    if not user:  
        raise HTTPException(status_code=401, detail="Unauthorized")  
    if image:  
        return print("Image received")  
    if user_message:  
        thread = threading.Thread(
            target=message_service.respond_and_send_message, 
            args=(user_message, user)
        )  
        thread.daemon = True  
        thread.start()  
    return {"status": "ok"}

• POST 端点实现： 该端点处理传入的 POST请求。它检查用户、消息或图像是否有效。如果都无效，它仅返回一个状态消息给云API。如果用户未通过身份验证，它将引发一个 HTTPException，状态码为 401。
• 处理图像和消息： 如果接收到图像，我们将进行简单的标准输出打印，作为未来图像处理的占位符。如果接收到文本消息，则使用单独的线程异步处理，以避免阻塞主应用程序线程。调用 message_service.respond_and_send_message 函数根据 LLM-代理工作流处理消息。

使用线程池处理 Webhook 的解释： WhatsApp 将重新发送 webhook，直到收到 200 响应，因此使用线程池确保消息处理不会阻塞 webhook 响应。

6 消息服务

在我们之前定义端点的展示层中，我们需要定义一些 message_service 函数。具体来说，我们需要一个实现，用于处理和转录音频有效载荷，身份验证用户，最后调用我们的代理并发送响应。我们将把所有这些功能放在 domain/message_service.py 中。在生产环境中，随着应用程序的增长，我建议进一步将它们拆分，例如 transcription_service.py、message_service.py 和 authentication_service.py。

在本节的多个函数中，我们将向 Meta API 发送请求 https://graph.facebook.com/...。在所有这些请求中，我们需要包含带有 WHATSAPP_API_KEY 的授权头，这个密钥是在 步骤 1.3 中创建的，作为 Bearer 令牌。我通常将 API 密钥和令牌存储在 .env 文件中，并使用 Python 的 dotenv 库访问它们。我们还使用 OpenAI 客户端和你的 OPENAI_API_KEY，这也可以存储在 .env 文件中。

但为了简单起见，让我们将它们放置并初始化在 message_service.py 脚本的顶部，如下所示：

import os  
import json  
import requests  
from typing import BinaryIO

WHATSAPP_API_KEY = "YOUR_ACCESS_TOKEN"
llm = OpenAI(api_key="YOUR_OPENAI_API_KEY")

将“YOUR_ACCESS_TOKEN”替换为你在步骤 1.3 中创建的实际访问令牌。

6.1 处理和转录音频文件

处理来自WhatsApp Webhook的语音记录并不像看起来那么简单。首先，重要的是要知道，传入的Webhook仅告诉我们数据类型和对象ID。因此，它不包含二进制音频文件。我们首先必须使用Meta的Graph API下载音频文件。要下载我们收到的音频，我们需要进行两个顺序请求。第一个是带有object_id的GET请求，以获取下载URL。这个下载URL是我们第二个GET请求的目标。

def download_file_from_facebook(file_id: str, file_type: str, mime_type: str) -> str | None:
    url = f"https://graph.facebook.com/v19.0/{file_id}"  
    headers = {"Authorization": f"Bearer {WHATSAPP_API_KEY}"}  
    response = requests.get(url, headers=headers)
    if response.status_code == 200:  
        download_url = response.json().get('url')

        response = requests.get(download_url, headers=headers)  
        if response.status_code == 200:
            file_extension = mime_type.split('/')[-1].split(';')[0]
            file_path = f"{file_id}.{file_extension}"  
            with open(file_path, 'wb') as file:  
                file.write(response.content)  
            if file_type == "image" or file_type == "audio":  
                return file_path  
        raise ValueError(f"下载文件失败。状态码: {response.status_code}")  
    raise ValueError(f"获取下载URL失败。状态码: {response.status_code}")

在这里，我们基本上获取下载URL，并使用对象ID和文件扩展名将文件下载到文件系统，作为其file_path。如果出现问题，我们会引发一个ValueError，指示错误发生的位置。

接下来，我们简单地定义一个函数，接受音频二进制并使用Whisper进行转录：

def transcribe_audio_file(audio_file: BinaryIO) -> str:  
    if not audio_file:  
        return "未提供音频文件"  
    try:  
        transcription = llm.audio.transcriptions.create(  
            file=audio_file,  
            model="whisper-1",  
            response_format="text"  
        )  
        return transcription  
    except Exception as e:  
        raise ValueError("转录音频时出错") from e

最后，让我们将下载和转录函数结合在一起：

def transcribe_audio(audio: Audio) -> str:  
    file_path = download_file_from_facebook(audio.id, "audio", audio.mime_type)  
    with open(file_path, 'rb') as audio_binary:  
        transcription = transcribe_audio_file(audio_binary)  
    try:  
        os.remove(file_path)  
    except Exception as e:  
        print(f"删除文件失败: {e}")  
    return transcription

6.2 身份验证用户

在使用Meta提供的测试号码时，我们必须预先定义我们的聊天机器人可以发送消息的号码。我不太确定，也没有测试过是否任何号码都可以向我们的聊天机器人发送消息。不过，无论如何，一旦我们切换到自定义号码，我们不希望任何人能够执行我们的代理聊天机器人。因此，我们需要一种方法来验证用户。我们有几种选择来做到这一点。首先，我们必须考虑在哪里存储用户信息。我们可以使用例如PostgreSQL这样的数据库或像Firestore这样的非关系数据库。我们可以在文件系统中以JSON文件或.env文件的形式预定义我们的用户。对于本教程，我将采用最简单的方法，将用户硬编码在我们的身份验证函数中的列表中。

列表条目的结构是步骤5.1中定义的User模型。因此，一个用户由ID、名字、姓氏和电话号码组成。我们尚未在我们的代理工作流中实现角色系统。但是在大多数具有不同用户的用例中，例如小型企业助手的示例案例，不同的用户将拥有不同的权限和访问范围。现在，我们只是传递"default"作为占位符角色。

def authenticate_user_by_phone_number(phone_number: str) -> User | None:  
    allowed_users = [  
        {"id": 1, "phone": "+1234567890", "first_name": "John", "last_name": "Doe", "role": "default"},  
        {"id": 2, "phone": "+0987654321", "first_name": "Jane", "last_name": "Smith", "role": "default"}  
    ]  
    for user in allowed_users:  
        if user["phone"] == phone_number:  
            return User(**user)  
    return None

因此，只需验证电话号码是否在我们的allowed_users列表中，如果在，则返回用户。否则，我们返回None。如果你查看我们在步骤5.3中的端点，你会看到如果用户为None，我们会引发错误，以防止进一步处理未经授权的用户消息。

6.3 发送消息

现在，在我们实际调用代理之前，最后一个助手函数是 send_whatsapp_message。我在这个函数中包含了两种模式，因为一些与Meta特定WhatsApp API逻辑相关的原因。

基本上，您不允许向用户发送自定义消息作为对话的开场白。这意味着如果用户首先开始对话并向聊天机器人发送消息，您可以用单独的文本消息进行回复。否则，如果您希望聊天机器人启动对话，则只能使用经过批准的模板，例如“Hello World”模板。

还值得一提的是，当我们谈论Meta逻辑时，启动对话后会开启一个24小时的对话窗口，在此窗口内您可以向该用户发送消息。这个对话窗口是收费的，而不是单独的消息。根据对话的类型（例如营销、支持等），情况会变得更加复杂。

您也可以自己定义一个模板并让其获得Meta的批准。在这一点上我还没有这样做，因此为了测试我们是否可以从后端向用户发送消息，我使用“Hello World”模板。如果您添加了一些自定义的已批准模板，您也可以使用此函数将它们发送给用户。

回到代码。要发送消息，我们发起一个POST请求并定义一个有效载荷，该有效载荷要么包含文本主体，要么包含模板：

def send_whatsapp_message(to, message, template=True):  
    url = f"https://graph.facebook.com/v18.0/289534840903017/messages"  
    headers = {  
        "Authorization": f"Bearer " + WHATSAPP_API_KEY,  
        "Content-Type": "application/json"  
    }  
    if not template:  
        data = {  
            "messaging_product": "whatsapp",  
            "preview_url": False,  
            "recipient_type": "individual",  
            "to": to,  
            "type": "text",  
            "text": {  
                "body": message  
            }  
        }  
    else:  
        data = {  
            "messaging_product": "whatsapp",  
            "to": to,  
            "type": "template",  
            "template": {  
                "name": "hello_world",  
                "language": {  
                    "code": "en_US"  
                }  
            }  
        }

    response = requests.post(url, headers=headers, data=json.dumps(data))  
    return response.json()

6.4 调用我们的代理

最后，我们可以整合我们之前示例中的代理。在这个阶段，您还可以整合自定义代理、Langchain AgentExecutor、Langgraph AgentWorkflow 等。

因此，我们每次接收到消息时调用的主要函数是 respond_and_send_message，它接收 user_message 字符串并将其作为输入对象传递给我们的代理工作流。

import json  
import requests
from app.domain.agents.routing_agent import RoutingAgent  
from app.schema import User

def respond_and_send_message(user_message: str, user: User):  
    agent = RoutingAgent()  
    response = agent.run(user_message, user.id)  
    send_whatsapp_message(user.phone, response, template=False)

在调用我们的代理后，我们得到一个响应消息，想要通过 send_whatsapp_message 函数发送回用户。

现在，您应该能够向测试号码发送消息，并通过代理执行器获得答案。备注：在使用 WhatsApp 测试号码时，您必须在您的 Meta API 应用中注册允许向您的机器人发送消息的电话号码。

通过遵循本指南，您已迈出了创建一个强大的 LLM 驱动的聊天机器人的重要一步，该机器人可以与 WhatsApp 无缝协作。这不仅仅是关于实时设置自动化业务通信；这也是为未来更高级的人工智能驱动工作流奠定基础。

我们做了什么：

• WhatsApp 云API 集成： 我们成功启动了 Meta 的 WhatsApp 云API，包括构建用于实时消息处理的 Webhook。
• 使用 FastAPI 的后端开发： 设置端点以验证 Webhook，处理传入消息，并异步处理响应。
• 多媒体与身份验证支持： 启用了音频转录和用户身份验证等功能，以实现更个性化的交互。图像处理尚未完全实现，但我们已经增加了实现该功能的可能性。

接下来是什么：

在接下来的部分中，我将把实现迁移到 LangGraph。我将为代理添加一些更多的功能，比如自主创建数据库表和工具，这将使代理更加灵活。我也欢迎关于添加哪些功能的反馈和想法！

总结：

将WhatsApp的可达性和可用性与LLMs结合起来，对企业和个人使用案例来说是一个重大胜利。无论您是想要一个个人助手还是一个完整的商业工具，本指南为您提供了到达目标的路径。继续尝试、改进和突破界限——这只是您可以构建的开始。

祝您编程愉快！🚀

您可以在这里找到完整的代码：Github Repo