Saját üzemeltetésű OpenAI-válaszok végpontjai

Note

A .NET OpenAI Responses-végpontjaihoz készült önálló üzemeltetési segédeszközök hamarosan érkeznek.

Note

Az OpenAI Responses végpontjaihoz készült, saját üzemeltetéshez használható segédprogramok jelenleg nem állnak rendelkezésre Go nyelvhez.

Használja a(z) agent-framework-hosting-responses elemet az OpenAI Responses formátumú kérések és válaszok konvertálására egy, az alkalmazása tulajdonában lévő végponton. A kiszolgáló kiválasztja a webes keretrendszert, az útvonalat, a hitelesítést, az engedélyezést, a kérési beállításokat és a munkamenet-tárolót.

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

A FastAPI-minta egy implementáció. Ugyanezek a segítők együttműködnek a Django, Flask, Starlette, Azure Functions vagy más keretrendszerrel.

Ügynökvégpont üzemeltetése

Ez a minta átalakítja a kérést ügynök-keretrendszer-futtatási értékekké, alkalmazás által definiált engedélyezési listát alkalmaz, és megőrzi a frissített munkamenetet az újonnan létrehozott válaszazonosító alatt.

app = FastAPI()
state = AgentState(
    create_agent,
    session_store=FileSessionStore(SESSIONS_DIR / "snapshots"),
)

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.

AgentState azonosítja a célt, és betölti vagy létrehozza a munkamenetet. Mentse a munkamenetet a futtatás után vagy egy streamelési futtatás befejezése után, mert a futtatás frissíti.

A teljes alkalmazásért, beleértve az ügynök definícióját és a kérésbeállítások engedélyezési listáját, lásd a helyi Responses-mintát.

Munkafolyamat-végpont üzemeltetése

WorkflowState lezárja a munkafolyamatot, de az alkalmazás felel az ellenőrzőpontok tárolásáért, valamint a válaszazonosító és az ellenőrzőpont közötti megfeleltetésért. Ez a minta visszaállítja az egy jogosult previous_response_id által kiválasztott ellenőrzőpontot, majd elment egy kurzort a következő válaszhoz.

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,
        )
    )

A minta fájlalapú tárhelye helyi fejlesztésre használható. Használjon tartós tárterületet, ha a replikák újraindulhatnak vagy felskálázhatók.

Important

Tekintse a(z) previous_response_id és conversation_id elemet nem megbízható bemenetnek. Hitelesítse a hívót, és ellenőrizze a jogosultságait, mielőtt bármelyik azonosítót a munkamenet vagy az ellenőrzőpont betöltésére vagy mentésére használná.

A szélesebb körű vezetékformátumért lásd az OpenAI-kompatibilis végpontokat.

Következő lépések

Telegram hozzáadása

Mélyedjen el:

  • Last updated on