Nota
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare ad accedere o modificare le directory.
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare a modificare le directory.
Annotazioni
Gli helper di self-hosting per gli endpoint OpenAI Responses in .NET saranno disponibili a breve.
Annotazioni
Gli strumenti di supporto per l'hosting autonomo degli endpoint Responses di OpenAI non sono attualmente disponibili in Go.
Usa agent-framework-hosting-responses per convertire richieste e risposte nel formato di OpenAI Responses su un endpoint gestito dalla tua applicazione. Il server sceglie il framework Web, la route, l'autenticazione, l'autorizzazione, le opzioni di richiesta e l'archiviazione delle sessioni.
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity
L'esempio FastAPI è un'implementazione. Gli stessi helper funzionano con Django, Flask, Starlette, Funzioni di Azure o un altro framework.
Ospita un endpoint dell'agente
Questo esempio converte la richiesta in valori di esecuzione di Agent Framework, applica un'opzione definita dall'applicazione allowlist e salva in modo permanente la sessione aggiornata con l'ID risposta appena creato.
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 risolve la destinazione e carica o crea una sessione. Salva la sessione dopo l'esecuzione o dopo il completamento di un'esecuzione in streaming, perché l'esecuzione la aggiorna.
Per l'applicazione completa, inclusa la definizione dell'agente e l'elenco consentiti delle opzioni di richiesta, vedere l'esempio locale di Responses.
Comprendere la conversione dell'utilizzo delle risposte
Per le risposte dell'agente e del flusso di lavoro, il pacchetto di hosting mantiene invariato un oggetto OpenAI ResponseUsage nativo valido per l'SDK quando ne è disponibile uno. Non unisce l'utilizzo delle risposte native con Agent Framework UsageDetails.
Quando l'utilizzo nativo non è disponibile, il pacchetto può ricostruire l'utilizzo delle risposte da questi campi di Agent Framework corrispondenti semanticamente:
| Valore di utilizzo | Campo del framework dell'agente |
|---|---|
| Token di input | input_token_count |
| Token di output | output_token_count |
| Token di input letti dalla cache | cache_read_input_token_count |
| Token di input per la scrittura in cache | cache_creation_input_token_count |
| Ragionamento dei token di output | reasoning_output_token_count |
I valori zero espliciti vengono mantenuti. Se total_tokens è assente, ma sono presenti sia il numero di input sia quello di output, il pacchetto lo calcola come somma di input e output.
Se l'utilizzo di Agent Framework disponibile è incompleto o semanticamente incoerente con lo schema Risposte, il pacchetto omette l'utilizzo. Non fa supposizioni, non copia un contatore in un altro contatore e non fa fallire una risposta che altrimenti avrebbe esito positivo. Un conteggio scalare in formato non valido rimane un errore.
Questa ricostruzione è intenzionalmente persa perché l'utilizzo di Agent Framework è indipendente dal provider e l'utilizzo delle risposte OpenAI ha una forma più completa e specifica del provider. I contatori specifici del provider segnalati da un agente ospitato, ad esempio l'utilizzo specifico di Anthropic, potrebbero pertanto non essere visualizzati nella risposta ricevuta dall'applicazione chiamante. Questa conversione non offre interoperabilità tra versioni diverse dell'SDK OpenAI in esecuzione nello stesso processo.
Ospitare un endpoint del flusso di lavoro
WorkflowState gestisce il flusso di lavoro, ma l'applicazione è responsabile dell'archiviazione dei checkpoint e dell'associazione tra un ID di risposta e un checkpoint. In questo esempio viene ripristinato il checkpoint selezionato da un oggetto autorizzato previous_response_id, quindi viene salvato un cursore per la risposta successiva.
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,
)
)
L'archiviazione basata su file dell'esempio è destinata allo sviluppo locale. Usare un'archiviazione durevole quando le repliche possono essere riavviate o sottoposte a scale-out.
Importante
Considerare previous_response_id e conversation come input non attendibile. Autenticare e autorizzare il chiamante prima di usare il valore per caricare o salvare una sessione o un checkpoint. Il campo di richiesta legacy conversation_id è deprecato; usa invece il campo conversation di OpenAI Responses.
Per il formato wire più generale, vedere endpoint compatibili con OpenAI.
Passaggi successivi
Approfondimento: