備註
針對 .NET 中 OpenAI 回應端點的自架輔助工具即將推出。
備註
目前尚未提供適用於 Go 的 OpenAI Responses 端點自託管輔助工具。
使用 agent-framework-hosting-responses 在您的應用程式所擁有的端點上轉換 OpenAI Responses 格式的要求與回應。 你的伺服器負責選擇網頁架構、路由、認證、授權、請求選項和會話儲存。
pip install --pre agent-framework agent-framework-foundry agent-framework-hosting agent-framework-hosting-responses azure-identity
FastAPI 範例就是一個實作。 同樣的助手也能使用 Django、Flask、Starlette、Azure Functions 或其他框架。
主機代理端點
此範例將請求轉換為代理框架執行值，套用應用程式定義的選項允許清單，並將更新後的工作階段持續存在於新建立的回應 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合併。
當原生使用不可用時，套件可從以下語意相符的代理框架欄位重建回應使用情況：
| 使用價值 | 代理框架欄位 |
|---|---|
| 輸入標記 | input_token_count |
| 輸出標記 | output_token_count |
| 從快取讀取的輸入詞元 | cache_read_input_token_count |
| 快取寫入輸入標記 | cache_creation_input_token_count |
| 推理輸出標記 | reasoning_output_token_count |
明確的零值會被保留。 若 total_tokens 在同時有輸入與輸出計數時不存在，封裝會將其推導為輸入加輸出。
若可用的代理框架使用資料不完整或語意上與回應結構不一致，套件會省略使用。 它不會猜測、不會把一個計數器複製到另一個計數器，也不會讓原本會成功的回應變成失敗。 格式錯誤的純量計數仍然是錯誤。
這種重建刻意採用有損方式，因為 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 相容端點。
下一步
深入探討：