메모
.NET OpenAI 응답 엔드포인트에 대한 자체 호스팅 도우미가 곧 제공될 예정입니다.
메모
OpenAI 응답 엔드포인트에 대한 자체 호스팅 도우미는 현재 Go에서 사용할 수 없습니다.
애플리케이션이 소유한 엔드포인트에서 OpenAI Responses 형식의 요청과 응답으로 변환하려면 agent-framework-hosting-responses를 사용하세요. 서버는 웹 프레임워크, 경로, 인증, 권한 부여, 요청 옵션 및 세션 스토리지를 선택합니다.
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 `conversation_id` 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 는 대상을 확인하고 세션을 로드하거나 만듭니다. 실행 후 또는 스트리밍 실행이 끝난 후 세션을 저장하세요. 실행이 세션을 업데이트하기 때문입니다.
에이전트 정의 및 요청 옵션 허용 목록을 포함한 전체 애플리케이션은 로컬 응답 샘플을 참조하세요.
워크플로 엔드포인트 호스트
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_id` 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_id 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,
)
)
샘플의 파일 기반 저장소는 로컬 개발용입니다. 복제본을 다시 시작하거나 스케일 아웃할 수 있는 경우 지속성 스토리지를 사용합니다.
Important
previous_response_id 및 conversation_id을 신뢰할 수 없는 입력으로 처리하십시오. ID를 사용하여 세션 또는 검사점을 로드하거나 저장하기 전에 호출자를 인증하고 권한을 부여합니다.
더 광범위한 와이어 형식은 OpenAI 호환 엔드포인트를 참조하세요.
다음 단계
더 자세히 살펴보기: