Hinweis
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, sich anzumelden oder das Verzeichnis zu wechseln.
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, das Verzeichnis zu wechseln.
Note
Hilfen zum Self-Hosting für OpenAI Responses-Endpunkte in .NET kommen bald.
Note
Hilfsprogramme für Self-Hosting von OpenAI-Responses-Endpunkten sind derzeit für Go nicht verfügbar.
Verwenden Sie agent-framework-hosting-responses, um Anfragen und Antworten im Format von OpenAI Responses über einen Endpunkt zu konvertieren, der Ihrer Anwendung gehört. Ihr Server wählt das Webframework, die Route, Authentifizierung, Autorisierung, Anforderungsoptionen und Sitzungsspeicher aus.
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity
Das FastAPI-Beispiel ist eine Implementierung. Die gleichen Helfer arbeiten mit Django, Flask, Starlette, Azure Functions oder einem anderen Framework.
Hosten eines Agent-Endpunkts
Dieses Beispiel konvertiert die Anfrage in Ausführungswerte des Agent Framework, wendet eine anwendungsdefinierte Zulassungsliste für Optionen an und speichert die aktualisierte Sitzung unter der neu erstellten Antwort-ID.
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 id supplied through `conversation` 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 ermittelt das Ziel und lädt oder erstellt eine Sitzung. Speichern Sie die Sitzung nach der Ausführung oder nach Abschluss einer Streamingausführung, da die Ausführung sie aktualisiert.
Die vollständige Anwendung, einschließlich der Agentendefinition und der Positivliste für Anforderungsoptionen, finden Sie im Beispiel „Local Responses“.
Konvertierung der Antwortnutzung verstehen
Für Agent- und Workflowantworten behält das Hostingpaket ein SDK-gültiges natives OpenAI-Objekt ResponseUsage unverändert bei, wenn ein Objekt verfügbar ist. Es führt die native Nutzung von Responses nicht mit Agent Framework UsageDetails zusammen.
Wenn die systemeigene Verwendung nicht verfügbar ist, kann das Paket die Verwendung von Antworten aus diesen semantisch übereinstimmenden Agent Framework-Feldern rekonstruieren:
| Verwendungswert | Feld für Agent Framework |
|---|---|
| Eingabetoken | input_token_count |
| Ausgabetoken | output_token_count |
| Eingabetoken für das Lesen aus dem Cache | cache_read_input_token_count |
| Eingabetoken für Cacheschreibvorgänge | cache_creation_input_token_count |
| Ausgabetoken für die Schlussfolgerung | reasoning_output_token_count |
Explizite Nullwerte bleiben erhalten. Wenn total_tokens fehlt, während sowohl die Anzahl der Eingaben als auch die der Ausgaben vorhanden sind, berechnet das Paket den Wert als Summe aus Eingaben und Ausgaben.
Wenn die verfügbare Agent Framework-Verwendung unvollständig oder semantisch mit dem Antwortschema inkonsistent ist, wird die Verwendung des Pakets weggelassen. Es wird nicht erraten, einen Zähler in einen anderen kopieren oder eine andernfalls erfolgreiche Antwort fehlschlagen. Eine fehlerhaft formatierte Skalaranzahl bleibt ein Fehler.
Diese Rekonstruktion ist bewusst verlustbehaftet, da die Nutzung des Agent Frameworks anbieterneutral ist und die Nutzung von OpenAI Responses eine umfangreichere, anbieterspezifische Struktur aufweist. Anbieterspezifische Zähler, die von einem gehosteten Agenten gemeldet werden, wie etwa Anthropic-spezifische Nutzung, könnten daher in der von der aufrufenden Anwendung empfangenen Antwort möglicherweise nicht erscheinen. Diese Konvertierung bietet keine Interoperabilität zwischen verschiedenen Versionen des OpenAI SDK, das im selben Prozess ausgeführt wird.
Hosten eines Workflowendpunkts
WorkflowState löst den Workflow auf, ihre Anwendung besitzt jedoch den Prüfpunktspeicher und die Zuordnung von einer Antwort-ID zu einem Prüfpunkt. Dieses Beispiel stellt den von einem autorisierten previous_response_id ausgewählten Checkpoint wieder her und speichert anschließend einen Cursor für die nächste Antwort.
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` 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 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,
)
)
Die dateibasierte Speicherung des Beispiels ist für die lokale Entwicklung vorgesehen. Verwenden Sie persistenten Speicher, wenn Replikate neu gestartet oder horizontal skaliert werden können.
Important
Behandeln previous_response_id und conversation als nicht vertrauenswürdige Eingabe. Authentifizieren und autorisieren Sie den Anrufer, bevor Sie entweder einen Wert zum Laden oder Speichern einer Sitzung oder eines Prüfpunkts verwenden. Das Feld für veraltete Anfragen conversation_id ist veraltet; verwenden Sie stattdessen das Feld OpenAI Responses conversation.
Informationen zum breiteren Drahtformat finden Sie unter OpenAI-kompatible Endpunkte.
Nächste Schritte
Gehen Sie tiefer: