ChatKit 고급 통합

ChatKit 고급 통합

완전한 제어가 필요할 때—커스텀 인증, 데이터 상주, 온프레미스 배포, 맞춤 에이전트 오케스트레이션—ChatKit을 자체 인프라에서 실행할 수 있어요. OpenAI의 고급 자체 호스팅 옵션을 사용해 나만의 서버와 커스터마이즈된 ChatKit을 쓸 수 있어요.

Agent Builder 호스팅 ChatKit 워크플로우는 전환 기간 중이에요. 새 ChatKit 앱은 ChatKit SDK와 Agents SDK로 자체 서버 측 에이전트 구현 위에 구축하세요. ChatKit 전환 지침을 참고하세요.

출처: 문서

본문

자체 인프라에서 ChatKit 실행하기

높은 수준에서 보면 고급 ChatKit 통합은 나만의 ChatKit 서버를 만들고 위젯을 추가해 채팅 표면을 구축하는 과정이에요. OpenAI API와 ChatKit 서버를 사용해 OpenAI 모델로 구동되는 커스텀 채팅을 만들어요.

ChatKit 서버 설정하기

들어오는 요청 처리, 툴 실행, 결과를 클라이언트로 스트리밍하는 방법은 GitHub의 서버 가이드를 따르세요. 아래 스니펫이 주요 구성 요소를 보여줘요.

1. 서버 패키지 설치하기

pip install openai-chatkit

2. 서버 클래스 구현하기

ChatKitServer가 대화를 이끌어요. 사용자 메시지나 클라이언트 툴 출력이 도착할 때마다 이벤트를 스트리밍하도록 respond를 오버라이드해요. stream_agent_response 같은 헬퍼가 서버를 Agents SDK에 연결해요.

class MyChatKitServer(ChatKitServer[RequestContext]):
    async def respond(
        self,
        thread: ThreadMetadata,
        input: UserMessageItem | ClientToolCallOutputItem | None,
        context: RequestContext,
    ) -> AsyncIterator[Event]:
        items_page = await self.store.load_thread_items(
            thread.id,
            after=None,
            limit=20,
            order="desc",
            context=context,
        )
        input_items = await simple_to_agent_input(list(reversed(items_page.data)))
        agent_context = AgentContext(
            thread=thread,
            store=self.store,
            request_context=context,
        )
        result = Runner.run_streamed(
            assistant_agent,
            input_items,
            context=agent_context,
        )
        async for event in stream_agent_response(agent_context, result):
            yield event

3. 엔드포인트 노출하기

선택한 프레임워크로 HTTP 요청을 서버 인스턴스에 전달해요. 예를 들어 FastAPI라면:

from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse

app = FastAPI()
data_store = MemoryStore()
server = MyChatKitServer(data_store)


@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
    result = await server.process(await request.body(), {})
    if isinstance(result, StreamingResult):
        return StreamingResponse(result, media_type="text/event-stream")
    return Response(content=result.json, media_type="application/json")

4. 데이터 저장소 계약 세우기

스레드, 메시지, 파일을 선호하는 데이터베이스로 영속화하려면 chatkit.store.Store를 구현해요. 로컬 개발에는 인메모리 Store 구현을 쓸 수 있어요. 프로덕션에는 영구 저장소를 쓰고, 라이브러리 업데이트가 마이그레이션 없이 스키마를 진화시킬 수 있도록 모델을 JSON blob으로 저장하는 걸 고려해 보세요.

5. 파일 저장소 계약 제공하기

업로드를 지원한다면 FileStore 구현을 제공하세요. ChatKit은 직접 업로드(클라이언트가 파일을 엔드포인트로 POST) 또는 2단계 업로드(클라이언트가 서명된 URL을 요청한 뒤 클라우드 저장소로 업로드)와 함께 동작해요. 인라인 썸네일을 지원하도록 미리보기를 노출하고, 스레드가 삭제될 때 삭제를 처리하세요.

6. 서버에서 클라이언트 툴 트리거하기

클라이언트 툴은 클라이언트 옵션과 에이전트 양쪽에 등록해야 해요. ctx.context.client_tool_call을 사용해 Agents SDK 툴에서 호출을 큐잉해요.

@function_tool(description_override="Add an item to the user's todo list.")
async def add_to_todo_list(ctx: RunContextWrapper[AgentContext], item: str) -> None:
    ctx.context.client_tool_call = ClientToolCall(
        name="add_to_todo_list",
        arguments={"item": item},
    )


assistant_agent = Agent[AgentContext](
    model="gpt-6-astra",
    name="Assistant",
    instructions="You are a helpful assistant",
    tools=[add_to_todo_list],
    tool_use_behavior=StopAtTools(stop_at_tool_names=[add_to_todo_list.name]),
)

7. 스레드 메타데이터와 상태 사용하기

thread.metadata를 사용해 이전 Responses API 실행 ID나 커스텀 라벨 같은 서버 측 상태를 저장해요. 메타데이터는 클라이언트에 노출되지 않지만 모든 respond 호출에서 사용 가능해요.

8. 툴 상태 업데이트 받기

오래 실행되는 툴은 ProgressUpdateEvent로 UI에 진행 상황을 스트리밍할 수 있어요. ChatKit은 다음 어시스턴트 메시지나 위젯 출력으로 진행 이벤트를 대체해요.

9. 서버 컨텍스트 사용하기

server.process(body, context)에 커스텀 컨텍스트 객체를 전달해 권한을 강제하거나 store와 file store 구현을 통해 사용자 정체성을 전파할 수 있어요.

인라인 인터랙티브 위젯 추가하기

위젯은 에이전트가 채팅 표면 안에 풍부한 UI를 표시할 수 있게 해요. 카드, 폼, 텍스트 블록, 목록 등 레이아웃에 사용해요. stream_widget 헬퍼는 위젯을 즉시 렌더링하거나 도착하는 대로 업데이트를 스트리밍할 수 있어요.

async def respond(
    self,
    thread: ThreadMetadata,
    input: UserMessageItem | ClientToolCallOutputItem | None,
    context: RequestContext,
) -> AsyncIterator[Event]:
    widget = Card(
        children=[
            Text(
                id="description",
                value="Generated summary",
            )
        ]
    )
    async for event in stream_widget(
        thread,
        widget,
        generate_id=lambda item_type: self.store.generate_item_id(
            item_type, thread, context
        ),
    ):
        yield event

ChatKit은 카드, 목록, 폼, 텍스트, 버튼 등 폭넓은 위젯 노드 집합을 제공해요. 모든 컴포넌트, props, 스트리밍 지침은 GitHub의 위젯 가이드를 참고하세요.

위젯을 인터랙티브 UI로 탐색하고 만들려면 Widget Builder를 보세요.

액션 사용하기

액션은 사용자 메시지를 보내지 않고도 ChatKit UI가 작업을 트리거하게 해요. 지원하는 위젯 노드(버튼, select, 그 외 제어)에 ActionConfig를 붙이면 새 스레드 항목을 스트리밍하거나 위젯을 제자리에서 업데이트할 수 있어요. 위젯이 Form 안에 있으면 ChatKit이 수집한 폼 값을 액션 payload에 포함해요.

서버에서는 ChatKitServer에 action 메서드를 구현해 payload를 처리하고 선택적으로 추가 이벤트를 스트리밍해요. handler="client"를 설정하고 JavaScript에서 응답한 뒤 후속 작업을 서버로 전달하는 식으로 클라이언트에서 액션을 처리할 수도 있어요.

액션 체이닝, 강한 타입 payload 생성, 클라이언트·서버 핸들러 조정 같은 패턴은 GitHub의 액션 가이드를 참고하세요.

리소스

통합을 완성하는 데 다음 리소스와 reference를 사용하세요.

디자인 리소스

• OpenAI Sans Variable 다운로드.
• 파일을 복제해 제품에 맞게 컴포넌트를 커스터마이즈하세요.

Events reference

ChatKit은 Web Component에서 CustomEvent 인스턴스를 내보내요. 수명 주기 이벤트를 듣고 event.detail에서 payload 데이터를 읽어요.

chatkit.addEventListener("chatkit.error", (event) => {
  console.error(event.detail.error);
});

chatkit.addEventListener("chatkit.response.start", () => {
  console.log("Response started");
});

chatkit.addEventListener("chatkit.response.end", () => {
  console.log("Response ended");
});

chatkit.addEventListener("chatkit.thread.change", (event) => {
  console.log("Active thread:", event.detail.threadId);
});

chatkit.addEventListener("chatkit.log", (event) => {
  console.log(event.detail.name, event.detail.data);
});

Options reference

더 알아보기 (Learn more)

• ChatKit 시작하기
• ChatKit 테마
• Widget Builder