OpenAI Yanıtlar uç noktalarını kendiniz barındırın

Note

.NET için OpenAI Responses uç noktalarına yönelik kendi sunucunda barındırma yardımcıları yakında kullanıma sunulacak.

Note

OpenAI Responses uç noktaları için kendi sunucunda barındırma yardımcıları şu anda Go için kullanılamamaktadır.

Uygulamanıza ait bir uç noktada, OpenAI Responses biçimindeki istekleri ve yanıtları dönüştürmek için agent-framework-hosting-responses kullanın. Sunucunuz web çerçevesini, yönlendirmeyi, kimlik doğrulamayı, yetkilendirmeyi, istek seçeneklerini ve oturum depolama alanını seçer.

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

FastAPI örneği bir uygulamadır. Aynı yardımcılar Django, Flask, Starlette, Azure İşlevleri veya başka bir çerçeveyle çalışır.

Ajan uç noktasını barındırın

Bu örnek isteği Agent Framework çalıştırma değerlerine dönüştürür, uygulama tanımlı seçenek izin listesi uygular ve güncelleştirilmiş oturumu yeni oluşturulan yanıt kimliği altında kalıcı hale döndürür.

app = FastAPI()
state = AgentState(create_agent)

ALLOWED_REQUEST_OPTIONS = frozenset({"max_tokens", "reasoning"})


@app.post("/responses", response_model=None)
async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse | StreamingResponse:  # noqa: B008
    """Handle one OpenAI Responses-shaped request."""
    try:
        run = responses_to_run(body)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    session_id, is_conversation_id = responses_session_id(body)
    conversation_id = session_id if is_conversation_id else None
    response_id = create_response_id()

    # App-specific policy: allow only the request options this route is willing
    # to honor. This denies tools, tool_choice, deployment/persistence fields,
    # and all other caller-supplied options by default. Your app decides which
    # options are allowed, altered, or denied.
    options = {key: value for key, value in run["options"].items() if key in ALLOWED_REQUEST_OPTIONS}
    options["reasoning"] = {"effort": "medium", "summary": "auto"}
    options_for_run = cast(Any, options)

    target = await state.get_target()
    lookup_id = session_id or response_id
    # An unknown `conversation_id` becomes a new session here. Production apps
    # can choose to require a separate "create conversation" API instead.
    session = await state.get_or_create_session(lookup_id)
    if run["stream"]:
        stream = target.run(
            run["messages"],
            stream=True,
            session=session,
            options=options_for_run,
        )
        if not isinstance(stream, ResponseStream):
            raise HTTPException(status_code=500, detail="agent did not return a response stream")

        async def stream_events() -> AsyncIterator[str]:
            async for event in responses_from_streaming_run(
                stream,
                response_id=response_id,
                conversation_id=conversation_id,
            ):
                yield event
            # `agent.run(..., stream=True)` updates the session while the stream
            # is consumed/finalized. Persist the selected continuation only
            # after finalization.
            if conversation_id is not None:
                # A stable conversation id is a mutable head. Apps must ensure
                # only one caller advances it at a time; AgentState does not
                # serialize concurrent runs for the same id.
                await state.set_session(conversation_id, session)
            else:
                await state.set_session(response_id, session)

        return StreamingResponse(
            stream_events(),
            media_type="text/event-stream",
        )

    result = await target.run(
        run["messages"],
        session=session,
        options=options_for_run,
    )
    # `agent.run(...)` updates the session. Persist the selected continuation
    # only after the run completes.
    if conversation_id is not None:
        # Preserve sequential conversation continuity. Production apps must
        # provide their own per-conversation single-writer coordination.

AgentState hedefi çözümler ve bir oturum yükler veya oluşturur. Çalıştırma tamamlandıktan sonra veya akış çalıştırması tamamlandıktan sonra oturumu kaydedin çünkü çalıştırma bunu güncelleştirir.

Aracı tanımı ve request-option allowlist dahil olmak üzere uygulamanın tamamı için yerel Yanıtlar örneğine bakın.

İş akışı uç noktasını barındır

WorkflowState iş akışını çözümler, ancak uygulamanız denetim noktası depolamasına ve yanıt kimliğinden denetim noktasına eşlemeye sahip olur. Bu örnek, yetkili bir previous_response_id tarafından seçilen kontrol noktasını geri yükler, ardından sonraki yanıt için bir imleç kaydeder.

app = FastAPI()
state = WorkflowState(workflow_builder, cache_target=False)


@app.post("/responses", response_model=None)
async def responses(body: dict[str, Any] = Body(...)) -> JSONResponse:  # noqa: B008
    """Handle one OpenAI Responses-shaped request for the workflow."""
    try:
        run = responses_to_run(body)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    # This sample demonstrates only Responses `previous_response_id`
    # continuation, so reject `conversation_id` instead of treating it as a
    # checkpoint cursor.
    previous_response_id, is_conversation_id = responses_session_id(body)
    if is_conversation_id:
        raise HTTPException(
            status_code=400,
            detail="This server supports previous_response_id continuation only; conversation_id is not implemented.",
        )
    response_id = create_response_id()

    target = await state.get_target()
    if previous_response_id and (checkpoint_cursor := checkpoint_cursor_store.get(previous_response_id)) is not None:
        # Restore first. Workflow.run does not allow `message` and
        # `checkpoint_id` in the same call.
        await target.run(
            checkpoint_id=checkpoint_cursor["checkpoint_id"],
            checkpoint_storage=checkpoint_storage_for(checkpoint_cursor["storage_id"]),
        )

    storage_id = response_id
    checkpoint_storage = checkpoint_storage_for(storage_id)
    result = await target.run(
        message=workflow_prompt_from_messages(run["messages"]),
        checkpoint_storage=checkpoint_storage,
    )

    latest = await checkpoint_storage.get_latest(workflow_name=target.name)
    if latest is not None:
        # Responses `previous_response_id` can point to any response id. Store
        # the current response id as the cursor for this workflow continuation.
        cursor = CheckpointCursor(checkpoint_id=latest.checkpoint_id, storage_id=storage_id)
        checkpoint_cursor_store.set_many({response_id: cursor})

    return JSONResponse(
        responses_from_run(
            response_from_workflow_result(result),
            response_id=response_id,
        )
    )

Örnek uygulamanın dosya tabanlı depolaması yerel geliştirme içindir. Replikalar yeniden başlatılabildiğinde veya yatay olarak ölçeklenebildiğinde dayanıklı depolama kullanın.

Important

previous_response_id ve conversation_id öğelerini güvenilmeyen girdi olarak kabul edin. Bir oturumu veya denetim noktasını yüklemek ya da kaydetmek için bu iki kimlikten herhangi birini kullanmadan önce çağıranın kimliğini doğrulayın ve çağıranı yetkilendirin.

Daha geniş kablo biçimi için bkz. OpenAI uyumlu uç noktalar.

Sonraki Adımlar

Telegram Ekle

Daha derine gidin:

  • Last updated on