agent-framework-chatkit 는 OpenAI ChatKit 스레드 항목을 에이전트 프레임워크 메시지로 변환하고 스트리밍된 에이전트 업데이트를 ChatKit 이벤트로 다시 변환합니다. 에이전트 프레임워크 Python 백 엔드가 있는 ChatKit 프런트 엔드를 원하는 경우 사용합니다.
통합은 다음을 제공합니다.
-
ThreadItemConverter는 ChatKit 스레드 항목 및 첨부 파일을 변환하는 데 사용됩니다. -
stream_agent_response()스트리밍된 에이전트 업데이트를 ChatKit 이벤트로 변환하기 위한 것입니다. -
simple_to_agent_input()기본 메시지 변환 경로의 경우
필수 조건
- Python 3.10 이상.
- FastAPI와 같은 백 엔드 웹 프레임워크입니다.
- ChatKit 프런트 엔드에 대한 Node.js.
- 프로덕션 프런트 엔드 도메인에 대한 ChatKit 도메인 키입니다.
패키지 설치
pip install agent-framework-chatkit --pre
ChatKit 서버 만들기
하위 클래스 ChatKitServer, Agent Framework 에이전트를 만들고 스레드 항목 및 첨부 파일에 대한 변환기를 구성합니다.
class WeatherChatKitServer(ChatKitServer[dict[str, Any]]):
"""ChatKit server implementation using Agent Framework.
This server integrates Agent Framework agents with ChatKit's server protocol,
providing weather information with interactive widgets and time queries through Azure OpenAI.
"""
def __init__(self, data_store: SQLiteStore, attachment_store: FileBasedAttachmentStore):
super().__init__(data_store, attachment_store)
logger.info("Initializing WeatherChatKitServer")
# Create Agent Framework agent with Azure OpenAI
# For authentication, run `az login` command in terminal
try:
self.weather_agent = Agent(
client=FoundryChatClient(credential=AzureCliCredential()),
instructions=(
"You are a helpful weather assistant with image analysis capabilities. "
"You can provide weather information for any location, tell the current time, "
"and analyze images that users upload. Be friendly and informative in your responses.\n\n"
"If a user asks to see a list of cities or wants to choose from available cities, "
"use the show_city_selector tool to display an interactive city selector.\n\n"
"When users upload images, you will automatically receive them and can analyze their content. "
"Describe what you see in detail and be helpful in answering questions about the images."
),
tools=[get_weather, get_time, show_city_selector],
)
logger.info("Weather agent initialized successfully with Azure OpenAI")
except Exception as e:
logger.error(f"Failed to initialize weather agent: {e}")
raise
# Create ThreadItemConverter with attachment data fetcher
self.converter = ThreadItemConverter(
attachment_data_fetcher=self._fetch_attachment_data,
)
응답 변환 및 스트리밍
스레드 기록을 로드하고, 에이전트 프레임워크 메시지로 변환하고, 스트리밍 모드에서 에이전트를 실행하고, ChatKit 이벤트를 생성합니다.
async def respond(
self,
thread: ThreadMetadata,
input_user_message: UserMessageItem | None,
context: dict[str, Any],
) -> AsyncIterator[ThreadStreamEvent]:
"""Handle incoming user messages and generate responses.
This method converts ChatKit messages to Agent Framework format using ThreadItemConverter,
runs the agent, converts the response back to ChatKit events using stream_agent_response,
and creates interactive weather widgets when weather data is queried.
"""
from agent_framework import FunctionResultContent
if input_user_message is None:
logger.debug("Received None user message, skipping")
return
logger.info(f"Processing message for thread: {thread.id}")
try:
# Track weather data and city selector flag for this request
weather_data: WeatherData | None = None
show_city_selector = False
# Load full thread history from the store
thread_items_page = await self.store.load_thread_items(
thread_id=thread.id,
after=None,
limit=1000,
order="asc",
context=context,
)
thread_items = thread_items_page.data
# Convert ALL thread items to Agent Framework ChatMessages using ThreadItemConverter
# This ensures the agent has the full conversation context
agent_messages = await self.converter.to_agent_input(thread_items)
if not agent_messages:
logger.warning("No messages after conversion")
return
logger.info(f"Running agent with {len(agent_messages)} message(s)")
# Run the Agent Framework agent with streaming
agent_stream = self.weather_agent.run(agent_messages, stream=True)
# Create an intercepting stream that extracts function results while passing through updates
async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
nonlocal weather_data, show_city_selector
async for update in agent_stream:
# Check for function results in the update
if update.contents:
for content in update.contents:
if isinstance(content, FunctionResultContent):
result = content.result
# Check if it's a WeatherResponse (string subclass with weather_data attribute)
if isinstance(result, str) and hasattr(result, "weather_data"):
extracted_data = getattr(result, "weather_data", None)
if isinstance(extracted_data, WeatherData):
weather_data = extracted_data
logger.info(f"Weather data extracted: {weather_data.location}")
# Check if it's the city selector marker
elif isinstance(result, str) and result == "__SHOW_CITY_SELECTOR__":
show_city_selector = True
logger.info("City selector flag detected")
yield update
# Stream updates as ChatKit events with interception
async for event in stream_agent_response(
intercept_stream(),
thread_id=thread.id,
):
yield event
ThreadItemConverter 는 StructuredInputItem 상태를 유지하고 각 질문을 사용자 메시지에서 답변, 건너뛰기 또는 답변이 없는 것으로 나타냅니다. 서식 지정을 사용자 지정하거나 답변을 편집할 수 있도록 비동기 structured_input_to_input() 메서드를 재정의합니다.
Message 하나, 메시지 목록 또는 항목을 생략하려면 None을 반환합니다.
또한 변환기는 나중에 턴에 모델 컨텍스트로 완료된 GeneratedImageItem 값을 유지합니다. 기본 변환은 짧은 텍스트 서문과 생성된 이미지 URI가 있는 사용자를 Message 만듭니다. 데이터 URI는 내장된 미디어 유형을 유지하고, 외부 URL에는 image/*를 사용하며, 이미지가 없는 미완성 항목은 건너뜁니다.
비공개 이미지 URL 확인과 같이 생성된 이미지 컨텍스트를 사용자 지정하려면 비동기 generated_image_to_input() 메서드를 재정의합니다.
Message 하나, 메시지 목록 또는 항목을 생략하려면 None을 반환합니다.
전체 샘플에서는 SQLite 지원 스레드, 파일 업로드, 첨부 파일 스토리지, 작업 및 대화형 위젯도 보여 줍니다.
Warning
ChatKit 프런트 엔드는 OpenAI의 CDN에서 로드되고 OpenAI 도메인에 대한 아웃바운드 요청을 수행합니다. 현재 자체 호스팅될 수 없으며 공기가 틈새가 있는 환경에는 적합하지 않습니다.