注释
用于 .NET 的 OpenAI Responses 端点自托管辅助工具即将推出。
注释
目前，Go 语言不支持 OpenAI Responses 端点的自托管辅助程序。
使用 agent-framework-hosting-responses 在由您的应用程序托管的端点将请求和响应转换为 OpenAI Responses 格式。 服务器选择 Web 框架、路由、身份验证、授权、请求选项和会话存储。
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity
FastAPI 示例是其中一种实现。 同样的辅助函数也适用于 Django、Flask、Starlette、Azure Functions 或其他框架。
托管代理端点
此示例将请求转换为 Agent Framework 运行值，应用应用程序定义的选项允许列表，并将更新的会话保留在新创建的响应 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 解析目标并加载或创建会话。 在运行结束后或流式运行完成后保存会话，因为运行会更新该会话。
有关完整的应用程序，包括代理定义和请求选项允许列表，请参阅 本地响应示例。
了解响应使用量换算
对于智能体和工作流响应，托管包会在存在可用对象时，原样保留符合 SDK 要求的原生 OpenAI ResponseUsage 对象。 它不会将本机响应使用情况与代理框架 UsageDetails合并。
当无法获得原生用量数据时，该包可以根据以下在语义上对应的 Agent Framework 字段重建 Responses 的用量信息：
| 使用价值 | 代理框架字段 |
|---|---|
| 输入标记 | input_token_count |
| 输出标记 | output_token_count |
| 缓存读取输入令牌 | cache_read_input_token_count |
| 缓存写入输入令牌 | cache_creation_input_token_count |
| 推理输出词元 | reasoning_output_token_count |
将保留显式零值。 如果 total_tokens 缺失，而输入计数和输出计数均存在，则该包会将其推导为输入与输出之和。
如果可用的 Agent Framework 用法不完整或语义上与响应架构不一致，则包会省略使用情况。 它不会进行猜测，不会将一个计数器的值复制到另一个计数器，也不会让原本成功的响应失败。 格式不正确的标量计数仍为错误。
这种重建是有意存在信息损失的，因为 Agent Framework 的用法与提供方无关，而 OpenAI Responses 的用法具有更丰富且依赖具体提供方的结构。 由托管代理报告的特定提供方计数器（例如 Anthropic 的使用情况）因此可能不会出现在调用方应用程序收到的响应中。 此转换不提供同一进程中运行的不同版本的 OpenAI SDK 之间的互操作性。
托管工作流终结点
WorkflowState 负责处理该工作流，但检查点存储以及响应 ID 到检查点的映射关系由您的应用程序负责。 此示例恢复由经授权的 previous_response_id 选择的检查点，然后为下一个响应保存游标。
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,
)
)
该示例的基于文件的存储仅用于本地开发。 当副本可以重启或横向扩展时，请使用持久存储。
重要
将 previous_response_id 和 conversation 视为不受信任的输入。 在使用值加载或保存会话或检查点之前，对调用方进行身份验证和授权。 旧 conversation_id 请求字段已弃用;请改用 OpenAI 响应 conversation 字段。
有关更通用的线格式，请参阅 OpenAI 兼容端点。
后续步骤
更深入：