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
전체 샘플에서는 SQLite 지원 스레드, 파일 업로드, 첨부 파일 스토리지, 작업 및 대화형 위젯도 보여 줍니다.
Warning
ChatKit 프런트 엔드는 OpenAI의 CDN에서 로드되고 OpenAI 도메인에 대한 아웃바운드 요청을 수행합니다. 현재 자체 호스팅될 수 없으며 공기가 틈새가 있는 환경에는 적합하지 않습니다.