Genkit Python 实战：用 Django 视图托管 Flow，实现 SSE 流式响应与请求上下文注入

Genkit Python 实战用 Django 视图托管 Flow实现 SSE 流式响应与请求上下文注入【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit本文基于 Genkit 仓库中py/samples/django-hello官方示例讲解如何把一个 Genkit Flow 直接挂到 Django 的urlpatterns上对外提供 JSON API 与 SSEServer-Sent Events流式接口。读完本文你将掌握示例的完整运行步骤、每个 Django 项目文件的职责ASGI 入口、URL 路由、精简配置以及genkit_django_handler装饰器底层的 HTTP 线协议——包括请求体包装格式、流式分片报文与错误响应的具体结构。示例定位与前置条件py/samples/django-hello的定位可以用示例代码文件头一句话概括见 recipes/views.py“A flow as a Django view. Streaming and request context included.”——它同时演示了两件事流式输出和从 HTTP 请求提取上下文。示例的依赖声明在 pyproject.toml 中运行环境要求 Python 3.10[project] name django-hello version 0.2.0 requires-python 3.10 dependencies [ django4.2, uvicorn, genkit, genkit-django, genkit-google-genai, pydantic, ]其中三个关键依赖各司其职genkit核心框架提供Genkit、ActionRunContext、ModelResponse等类型genkit-djangoDjango 适配插件提供genkit_django_handler装饰器源码位于 py/packages/genkit-django/src/genkit_django/handler.pygenkit-google-genaiGoogle AI 模型插件示例用它调用gemini-flash-latest。另外genkit-django插件要求Django 4.1 的 ASGI 服务因为返回的视图是async defWSGI 无法承载这也是uvicorn出现在依赖列表里的原因。快速启动三步跑通官方 README 给出的完整启动流程如下。先配置 Google AI 的 API Key再用uv同步依赖并启动 ASGI 服务export GEMINI_API_KEYyour-api-key uv sync uv run uvicorn myproject.asgi:application --port 8080myproject.asgi:application指向 Django 的 ASGI 应用对象定义在 myproject/asgi.pyimport os from django.core.asgi import get_asgi_application os.environ.setdefault(DJANGO_SETTINGS_MODULE, myproject.settings) application get_asgi_application()该文件注释明确写道“Django 4.1 async views need this, not WSGI.”即异步视图必须走 ASGI 入口。服务启动后用一条curl命令验证流式接口curl -N -X POST http://localhost:8080/chat \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -H Authorization: beginner-demo \ -d {data:{name:Mittens}}逐字段拆解这条请求请求要素作用curl -N关闭 curl 输出缓冲让 SSE 分片实时打印到终端POST .../chat插件视图仅支持 POST其他方法返回 405Accept: text/event-stream触发 SSE 流式模式也可以改用?streamtrue查询参数Authorization: beginner-demo被auth_context捕获作为用户名注入 Flow 上下文-d {data:{name:Mittens}}请求体必须是{data: ...}包装name缺省时 Flow 输入模型的默认值Mittens生效注意请求体不是直接传{name: Mittens}而是多了一层data包装——这是 Genkit HTTP 协议的统一约定下文线协议一节会给出源码级依据。核心代码一个 Flow 同时承担视图角色示例最精华的文件是 recipes/views.py完整代码如下from django.http import HttpRequest from genkit_django import genkit_django_handler from genkit_google_genai import GoogleAI from pydantic import BaseModel from genkit import ActionRunContext, Genkit, ModelResponse from genkit.plugin_api import RequestData ai Genkit( plugins[GoogleAI()], modelGoogleAI.gemini_model(gemini-flash-latest), ) class SayHiInput(BaseModel): name: str Mittens async def auth_context(request: RequestData[HttpRequest]) - dict[str, object]: # The caller is identified from the request, not the JSON body. return {username: request.request.META.get(HTTP_AUTHORIZATION) or guest} genkit_django_handler(ai, context_providerauth_context) ai.flow() async def say_hi(input: SayHiInput, ctx: ActionRunContext) - ModelResponse: username ctx.context.get(username, guest) stream ai.generate_stream( promptftell a medium sized joke about {input.name} for user {username}, contextctx.context, ) async for chunk in stream.stream: if chunk.text: ctx.send_chunk(chunk.text) return await stream.response这段代码包含四个值得展开的技术点1. Flow 输入用 Pydantic 模型声明。SayHiInput是一个BaseModel字段name带默认值Mittens。请求体data内的 JSON 会按此模型校验缺字段时回落到默认值。2.context_provider把调用者身份从请求头注入 Flow。auth_context接收一个RequestData[HttpRequest]插件对 Django 请求的包装类型示例特意选择从Authorization头Django 中对应META[HTTP_AUTHORIZATION]而非 JSON 正文识别调用者——正文数据不可信、可被随意构造身份标识必须来自请求元数据。该 provider 可以是同步或异步函数源码中会await其协程结果返回的 dict 最终出现在ctx.context里Flow 内通过ctx.context.get(username, guest)读取。3. 装饰器叠加顺序有强制约束。必须是genkit_django_handler在外、ai.flow()在内。从 handler.py 可以看到装饰器内部会检查isinstance(flow, Action)不满足时抛出GenkitError(statusINVALID_ARGUMENT, messagemust apply genkit_django_handler on a flow)。4. 流式输出的双通道写法。ai.generate_stream返回一个流对象代码在循环中对每个文本分片调用ctx.send_chunk(chunk.text)向外推送 SSE 分片最后await stream.response拿到完整ModelResponse作为 Flow 的最终返回值。前者决定客户端“边说边看”的体验后者决定接口的最终结果。Django 侧装配URL、ASGI 与精简配置路由注册在 myproject/urls.py 中Flow 被“当作”一个普通视图挂到urlpatternsfrom django.urls import path from recipes.views import say_hi urlpatterns [ path(chat, say_hi), ]这里能看到装饰器链的实际效果say_hi经过genkit_django_handler包装后类型已经是一个接受HttpRequest的异步视图函数可以直接被path()接收。项目配置在 myproject/settings.py 中被刻意裁剪到最小——因为“本进程只需要 ASGI 和 urlpatterns”文件头注释原话INSTALLED_APPS [ django.contrib.contenttypes, django.contrib.auth, recipes, ] MIDDLEWARE [ django.middleware.common.CommonMiddleware, ] ROOT_URLCONF myproject.urls ASGI_APPLICATION myproject.asgi.application DATABASES {}要点DATABASES {}表示完全不配置数据库无 ORM 依赖MIDDLEWARE只保留CommonMiddleware且没有启用 CSRF 中间件——这与插件自身对视图施加csrf_exempt相配合因为这是一个 JSON API 而非表单应用。SECRET_KEY从环境变量DJANGO_SECRET_KEY读取缺省为dev生产环境务必覆盖。项目保留了标准的 manage.py 管理入口DJANGO_SETTINGS_MODULE同样指向myproject.settings可用于migrate、check等管理命令但本示例的日常运行完全走 uvicorn。底层线协议genkit_django_handler 到底做了什么genkit-django插件与genkit-flask、genkit-fastapi保持同一套 HTTP 协议见 插件 README装饰器把任意框架的请求转换为统一的“JSON 信封 可选 SSE 流”。结合 handler.py 源码协议行为可以精确描述为请求侧全部由handler内部校验仅接受 POST其他方法返回405INVALID_ARGUMENT错误体请求体必须是合法 JSON否则400请求体必须是包含data键的对象{data: flow_input}缺失时报错信息为 “Action request must be wrapped in {data: ...} object”同样返回400若提供了context_provider其返回值dict会作为context参数传入flow.run/flow.stream。流式模式Accept: text/event-stream或?streamtrue触发data: {message:chunk}\n\n # 每个分片一条 SSE 消息 data: {result:response}\n\n # 流结束时的最终结果 data: {error:error}\n\n # 异常时改为错误消息对应源码中StreamingHttpResponse(event_stream(), content_typetext/event-stream)的生成器先逐分片yield fdata: {json.dumps({message: _to_dict(chunk)}, ...)}\n\n待await stream_response.response完成后再yield一条{result: ...}JSON 序列化使用紧凑分隔符(,, :)以减少传输体积。非流式模式直接await flow.run(...)成功返回JsonResponse({result: flow_output})Flow 执行期间抛出的异常统一以500HttpErrorWireFormatJSON 返回且会先尝试解开GenkitError.cause以保留最原始的错误信息_unwrap_cause函数。响应值的序列化细节Flow 的返回值可以是 Pydantic 模型、模型列表或含模型的 dict而 Django 的JsonResponse不原生支持BaseModel序列化。handler 内部有一个_to_dict递归函数会遍历 list/tuple/dict 并把遇到的每个BaseModel转为model_dump()结果再交给json.dumps——这就是为什么say_hi可以安全地返回ModelResponse这样的复杂对象。小结这个示例教会你什么django-hello虽然只有 6 个 Python 文件却给出了在 Django 中集成 Genkit 的完整最小路径装配路径Genkit(plugins[GoogleAI()])→ai.flow()定义 Flow →genkit_django_handler(ai, context_provider...)包装为视图 →urlpatterns挂载 → ASGI 入口由 uvicorn 承载协议契约客户端必须发送{data: ...}包装体Accept: text/event-stream决定 SSE 与非流式两种响应形态上下文注入范式身份等可信信息走context_provider从请求头/请求对象提取经ctx.context流入 Flow而不是依赖请求正文可替换性把GoogleAI换成Anthropic、OpenAI等其他插件后视图、路由、启动命令均无需改动——这是把 Flow 与 HTTP 传输层解耦由genkit-django统一处理带来的直接好处。如果你需要更完整的协议字段说明可继续参考 py/packages/genkit-django/README.md 的 Wire protocol 与 Context provider 两节以及 py/packages/genkit-django/tests/django_handler_test.py 中的行为验证用例。【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考