Django：后端管道 – ModelRiver Docs

概述

Django 的“全功能 (Batteries-included)”特性使其处理 ModelRiver 后端管道 Webhooks 变得非常直接。您可以使用 Django views 作为 Webhook 端点，使用 Celery 进行后台处理，并使用 Django REST Framework 进行结构化请求处理。

您将构建的内容：

一个用于接收 ModelRiver 发出的 AI 生成数据的 Webhook 端点
HMAC 签名验证修饰器 (Decorator)
用于异步处理和回调的 Celery 任务
用于数据持久化的 Django ORM 集成

快速开始

安装依赖

Bash

pip install django djangorestframework celery requests python-dotenv

配置

PYTHON

1# settings.py
2import os
3 
4MODELRIVER_API_KEY = os.environ.get("MODELRIVER_API_KEY", "")
5MODELRIVER_WEBHOOK_SECRET = os.environ.get("MODELRIVER_WEBHOOK_SECRET", "")

Webhook 处理程序

View

PYTHON

1# webhooks/views.py
2import hmac
3import hashlib
4import json
5import os
6import requests
7from django.http import JsonResponse
8from django.views.decorators.csrf import csrf_exempt
9from django.views.decorators.http import require_POST
10from django.conf import settings
11 
12 
13def verify_webhook_signature(raw_body, timestamp, signature, secret):
14    """验证 x-modelriver-signature：对 "<timestamp>.<data 的原始 JSON 文本>" 计算 HMAC-SHA256。"""
15    # 从原始请求体中截取 "data" 的原始 JSON 文本（切勿对解析后的 JSON 重新编码）
16    start = raw_body.find(b'"data":')
17    end = raw_body.rfind(b',"timestamp":')
18    if start == -1 or end == -1:
19        return False
20    signed = timestamp.encode("utf-8") + b"." + raw_body[start + len(b'"data":'):end]
21    expected = hmac.new(
22        secret.encode("utf-8"),
23        signed,
24        hashlib.sha256,
25    ).hexdigest()
26    return hmac.compare_digest(expected, signature)
27 
28 
29@csrf_exempt
30@require_POST
31def modelriver_webhook(request):
32    signature = request.headers.get("x-modelriver-signature", "")
33    timestamp = request.headers.get("x-modelriver-timestamp", "")
34    raw_body = request.body
35 
36    # 1. 验证 Webhook 签名
37    if not verify_webhook_signature(raw_body, timestamp, signature, settings.MODELRIVER_WEBHOOK_SECRET):
38        return JsonResponse({"error": "Invalid signature"}, status=401)
39 
40    payload = json.loads(raw_body).get("data", {})  # 事件载荷位于 "data" 之下
41    event_type = payload.get("type")
42    callback_url = payload.get("callback_url")
43 
44    # 2. 处理后端管道的工作流
45    if event_type == "task.ai_generated" and callback_url:
46        from .tasks import process_ai_webhook
47 
48        # 使用 Celery 将任务加入后台队列
49        process_ai_webhook.delay(
50            event=payload.get("event"),
51            ai_response=payload.get("ai_response"),
52            callback_url=callback_url,
53            customer_data=payload.get("customer_data", {}),
54        )
55        return JsonResponse({"received": True}, status=200)
56 
57    # 标准 Webhook
58    return JsonResponse({"received": True}, status=200)

URL 配置

PYTHON

1# webhooks/urls.py
2from django.urls import path
3from . import views
4 
5urlpatterns = [
6    path("webhooks/modelriver/", views.modelriver_webhook, name="modelriver_webhook"),
7]

Celery 后台任务

PYTHON

1# webhooks/tasks.py
2import requests
3from celery import shared_task
4from django.conf import settings
5from datetime import datetime
6 
7 
8@shared_task(bind=True, max_retries=3, default_retry_delay=10)
9def process_ai_webhook(self, event, ai_response, callback_url, customer_data):
10    """处理 AI 响应并回调 ModelRiver。"""
11    try:
12        enriched_data = {**ai_response.get("data", {})}
13 
14        # 3. 基于事件的自定义业务逻辑
15        if event == "content_ready":
16            from myapp.models import Content
17 
18            content = Content.objects.create(
19                title=enriched_data.get("title", ""),
20                body=enriched_data.get("description", ""),
21                category=customer_data.get("category", "general"),
22                source="modelriver",
23            )
24            enriched_data["id"] = str(content.id)
25            enriched_data["slug"] = content.slug
26            enriched_data["saved_at"] = datetime.now().isoformat()
27 
28        if event == "entities_extracted":
29            from myapp.models import Entity
30 
31            for entity in enriched_data.get("entities", []):
32                Entity.objects.create(
33                    name=entity["name"],
34                    entity_type=entity["type"],
35                    source_id=customer_data.get("document_id"),
36                )
37            enriched_data["entities_saved"] = len(enriched_data.get("entities", []))
38 
39        # 4. 回调 ModelRiver
40        response = requests.post(
41            callback_url,
42            json={
43                "data": enriched_data,
44                "task_id": f"django_{event}_{datetime.now().strftime('%Y%m%d%H%M%S')}",
45                "metadata": {
46                    "processed_by": "django",
47                    "processed_at": datetime.now().isoformat(),
48                },
49            },
50            headers={
51                "Authorization": f"Bearer {settings.MODELRIVER_API_KEY}",
52                "Content-Type": "application/json",
53            },
54            timeout=10,
55        )
56        response.raise_for_status()
57        print(f"✅ 已针对事件发送回调: {event}")
58 
59    except requests.exceptions.RequestException as exc:
60        print(f"❌ 回调失败: {exc}")
61        # 发送错误回调
62        try:
63            requests.post(
64                callback_url,
65                json={
66                    "data": {
67                        "processing_status": "failed",
68                        "processing_error": str(exc),
69                    },
70                },
71                headers={
72                    "Authorization": f"Bearer {settings.MODELRIVER_API_KEY}",
73                    "Content-Type": "application/json",
74                },
75                timeout=10,
76            )
77        except Exception:
78            pass
79        raise self.retry(exc=exc)

Django REST Framework

使用 DRF Serializers 的更结构化方法：

PYTHON

1# webhooks/serializers.py
2from rest_framework import serializers
3 
4 
5class AIResponseSerializer(serializers.Serializer):
6    data = serializers.DictField()
7 
8 
9class WebhookPayloadSerializer(serializers.Serializer):
10    """校验信封中 "data" 键下的事件载荷。"""
11    type = serializers.CharField(required=False)
12    event = serializers.CharField(required=False)
13    channel_id = serializers.CharField()
14    ai_response = AIResponseSerializer(required=False)
15    callback_url = serializers.URLField(required=False)
16    callback_required = serializers.BooleanField(required=False)
17    customer_data = serializers.DictField(required=False, default={})
18 
19 
20# webhooks/views.py
21from rest_framework.views import APIView
22from rest_framework.response import Response
23from rest_framework import status
24 
25 
26class ModelRiverWebhookView(APIView):
27    authentication_classes = []  # Webhook 使用签名验证
28    permission_classes = []
29 
30    def post(self, request):
31        signature = request.headers.get("x-modelriver-signature", "")
32        timestamp = request.headers.get("x-modelriver-timestamp", "")
33        raw_body = request.body  # 在 request.data 解析之前读取原始字节
34 
35        if not verify_webhook_signature(raw_body, timestamp, signature, settings.MODELRIVER_WEBHOOK_SECRET):
36            return Response({"error": "Invalid signature"}, status=status.HTTP_401_UNAUTHORIZED)
37 
38        serializer = WebhookPayloadSerializer(data=request.data.get("data", {}))
39        serializer.is_valid(raise_exception=True)
40 
41        data = serializer.validated_data
42        if data.get("type") == "task.ai_generated" and data.get("callback_url"):
43            process_ai_webhook.delay(
44                event=data.get("event"),
45                ai_response=data.get("ai_response"),
46                callback_url=data["callback_url"],
47                customer_data=data.get("customer_data", {}),
48            )
49 
50        return Response({"received": True}, status=status.HTTP_200_OK)

触发异步请求

PYTHON

1# ai/client.py
2import requests
3from django.conf import settings
4 
5 
6def trigger_async_ai(workflow: str, prompt: str, metadata: dict = None) -> dict:
7    """触发后端管道的 AI 请求。"""
8    response = requests.post(
9        "https://api.modelriver.com/v1/ai/async",
10        headers={
11            "Authorization": f"Bearer {settings.MODELRIVER_API_KEY}",
12            "Content-Type": "application/json",
13        },
14        json={
15            "model": workflow,
16            "messages": [{"role": "user", "content": prompt}],
17            "metadata": metadata or {},
18        },
19        timeout=10,
20    )
21    response.raise_for_status()
22    return response.json()

最佳实践

为处理过程使用 Celery：绝不要让繁重的逻辑阻塞 Webhook 响应。
设置任务重试：在 Celery 任务上使用 max_retries 和 default_retry_delay。
首先验证签名：在处理之前，务必先校验 x-modelriver-signature。
使用 Django ORM 事务：为了保证一致性，请将数据库写入操作封装在 transaction.atomic() 中。
使用请求日志进行监控：在可观测性 (Observability) 中追踪后端管道流程。

下一步

FastAPI 后端管道指南：异步 Python 备选方案
Django 集成：标准 ModelRiver + Django 的用法
Webhooks 参考：重试政策和投递监控

Django 的后端管道

概述

快速开始

安装依赖

配置

Webhook 处理程序

View

URL 配置

Celery 后台任务

Django REST Framework

触发异步请求

最佳实践

下一步