Remarque
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de vous connecter ou de modifier des répertoires.
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de modifier des répertoires.
Note
Les outils d’auto-hébergement pour les points de terminaison de l’API Responses d’OpenAI dans .NET seront bientôt disponibles.
Note
Les outils d’auto-hébergement pour les endpoints Responses d’OpenAI ne sont pas disponibles actuellement pour Go.
Utilisez agent-framework-hosting-responses pour convertir des requêtes et des réponses au format OpenAI Responses sur un point de terminaison appartenant à votre application. Votre serveur choisit l’infrastructure web, l’itinéraire, l’authentification, l’autorisation, les options de demande et le stockage de session.
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity
L’exemple FastAPI est une implémentation. Les mêmes utilitaires fonctionnent avec Django, Flask, Starlette, Azure Functions ou un autre cadre applicatif.
Héberger un point de terminaison d’agent
Cet exemple convertit la requête en valeurs d’exécution d’Agent Framework, applique une liste d’autorisation d’options définie par l’application et enregistre la session mise à jour sous l’ID de réponse nouvellement créé.
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 résout la cible et charge ou crée une session. Enregistrez la session après l’exécution, ou une fois qu’une exécution de diffusion en continu s’est terminée, car l’exécution la met à jour.
Pour l’application complète, y compris la définition de l’agent et la liste d’autorisation des options de requête, consultez l’exemple local de Responses.
Comprendre la conversion de l’utilisation des réponses
Pour les réponses de l’agent et du flux de travail, le package d’hébergement conserve inchangé un objet OpenAI natif valide pour le SDK ResponseUsage lorsqu’un tel objet est disponible. Elle ne fusionne pas l’utilisation des réponses natives avec Agent Framework UsageDetails.
Lorsque l’utilisation native n’est pas disponible, le package peut reconstruire l’utilisation des réponses à partir de ces champs d’infrastructure agent correspondant sémantiquement :
| Valeur d’utilisation | Champ du cadre d’agent |
|---|---|
| Jetons d’entrée | input_token_count |
| Jetons de sortie | output_token_count |
| Jetons d'entrée lus depuis le cache | cache_read_input_token_count |
| Jetons d’entrée d’écriture en cache | cache_creation_input_token_count |
| Tokens de sortie du raisonnement | reasoning_output_token_count |
Les valeurs zéro explicites sont conservées. Si total_tokens est absent alors que les nombres d’entrées et de sorties sont présents, le paquet le calcule comme la somme des entrées et des sorties.
Si l’utilisation de l’Infrastructure de l’agent disponible est incomplète ou sémantiquement incohérente avec le schéma Réponses, le package omet l’utilisation. Il ne devine pas, ne copie pas un compteur dans un autre et n’invalide pas une réponse par ailleurs réussie. Un nombre scalaire mal formé reste une erreur.
Cette reconstruction est volontairement avec perte, car l’utilisation d’Agent Framework est indépendante du fournisseur et l’utilisation d’OpenAI Responses présente une structure plus riche et spécifique au fournisseur. Les compteurs spécifiques au fournisseur rapportés par un agent hébergé, comme les données d’utilisation propres à Anthropic, peuvent par conséquent ne pas apparaître dans la réponse reçue par l’application appelante. Cette conversion ne fournit pas d’interopérabilité entre différentes versions du Kit de développement logiciel (SDK) OpenAI en cours d’exécution dans le même processus.
Héberger un point de terminaison de flux de travail
WorkflowState résout le flux de travail, mais votre application possède un stockage de points de contrôle et le mappage d’un ID de réponse à un point de contrôle. Cet exemple restaure le point de contrôle sélectionné par un previous_response_id autorisé, puis enregistre un curseur pour la réponse suivante.
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,
)
)
Le stockage sur fichiers de cet exemple est destiné au développement local. Utilisez un stockage durable lorsque les réplicas peuvent redémarrer ou effectuer un scale-out.
Important
Traiter previous_response_id et conversation comme une entrée non approuvée. Authentifiez et autorisez l’appelant avant d’utiliser la valeur pour charger ou enregistrer une session ou un point de contrôle. Le champ de requête hérité conversation_id est déconseillé ; utilisez plutôt le champ Réponses conversation OpenAI.
Pour obtenir le format de câble plus large, consultez les points de terminaison compatibles OpenAI.
Étapes suivantes
Aller plus loin :