Nota:
El acceso a esta página requiere autorización. Puede intentar iniciar sesión o cambiar directorios.
El acceso a esta página requiere autorización. Puede intentar cambiar los directorios.
Note
Próximamente estarán disponibles las herramientas de autohospedaje para los puntos de conexión de OpenAI Responses en .NET.
Note
Las herramientas de autohospedaje para los puntos de conexión de OpenAI Responses no están disponibles actualmente para Go.
Utiliza agent-framework-hosting-responses para convertir solicitudes y respuestas con el formato OpenAI Responses en un punto de conexión propiedad de tu aplicación. El servidor elige el marco web, la ruta, la autenticación, la autorización, las opciones de solicitud y el almacenamiento de sesión.
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity
El ejemplo de FastAPI es una implementación. Los mismos asistentes trabajan con Django, Flask, Starlette, Azure Functions u otro marco.
Alojar un punto de conexión de un agente
Este ejemplo convierte la solicitud en valores de ejecución del Agent Framework, aplica una lista de opciones permitidas definida por la aplicación y guarda la sesión actualizada con el identificador de respuesta recién creado.
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 resuelve el destino y carga o crea una sesión. Guarde la sesión después de la ejecución o después de que finalice una ejecución de streaming, ya que la ejecución la actualiza.
Para ver la aplicación completa, incluida la definición del agente y la lista de opciones de solicitud permitidas, consulte el ejemplo local de Responses.
Comprender la conversión del uso de respuestas
En el caso de las respuestas de agente y flujo de trabajo, el paquete de hospedaje conserva un objeto OpenAI ResponseUsage nativo válido del SDK sin cambios cuando hay uno disponible. No combina el uso nativo de Responses con Agent Framework UsageDetails.
Cuando el uso nativo no está disponible, el paquete puede reconstruir el uso de respuestas a partir de estos campos del marco del agente que coincidan semánticamente:
| Valor de uso | Campo de Agent Framework |
|---|---|
| Tokens de entrada | input_token_count |
| Tokens de salida | output_token_count |
| Tokens de entrada de lectura en caché | cache_read_input_token_count |
| Tokens de entrada para escritura en caché | cache_creation_input_token_count |
| Tokens de salida del razonamiento | reasoning_output_token_count |
Se conservan los valores cero explícitos. Si total_tokens no está presente, pero sí lo están los recuentos de entrada y de salida, el paquete lo calcula como la suma de entrada y salida.
Si el uso de Agent Framework disponible está incompleto o semánticamente incoherente con el esquema de respuestas, el paquete omite el uso. No realiza conjeturas, no copia un contador en otro ni rechaza una respuesta que, de otro modo, sería correcta. Un recuento escalar con formato incorrecto sigue siendo un error.
Esta reconstrucción presenta pérdidas intencionadas, ya que el uso de Agent Framework es independiente del proveedor, mientras que el uso de OpenAI Responses tiene una estructura más rica y específica del proveedor. Por lo tanto, es posible que los contadores específicos del proveedor notificados por un agente hospedado, como el uso específico de Anthropic, no aparezcan en la respuesta recibida por la aplicación que realiza la llamada. Esta conversión no proporciona interoperabilidad entre distintas versiones del SDK de OpenAI que se ejecutan en el mismo proceso.
Hospedaje de un punto de conexión de flujo de trabajo
WorkflowState resuelve el flujo de trabajo, pero su aplicación es responsable del almacenamiento de los puntos de control y de la correspondencia entre un identificador de respuesta y un punto de control. En este ejemplo se restaura el punto de control seleccionado por un autorizado previous_response_idy, a continuación, se guarda un cursor para la siguiente respuesta.
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,
)
)
El almacenamiento basado en archivos del ejemplo está destinado al desarrollo local. Utiliza almacenamiento duradero cuando las réplicas puedan reiniciarse o escalarse horizontalmente.
Important
Trate previous_response_id y conversation como entrada que no es de confianza. Autentique y autorice al autor de la llamada antes de usar cualquier valor para cargar o guardar una sesión o un punto de control. El campo de solicitud heredado conversation_id está en desuso; use el campo Respuestas conversation de OpenAI en su lugar.
Para obtener el formato de conexión más amplio, consulte Puntos de conexión compatibles con OpenAI.
Pasos siguientes
Vaya más profundamente: