Remarque
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de vous connecter ou de modifier des répertoires.
L’accès à cette page nécessite une autorisation. Vous pouvez essayer de modifier des répertoires.
agent-framework-chatkit convertit les éléments de thread OpenAI ChatKit en messages Agent Framework et convertit les mises à jour de l’agent en continu en événements ChatKit. Utilisez-le lorsque vous souhaitez une interface ChatKit avec un backend Python basé sur Agent Framework.
L’intégration fournit les éléments suivants :
-
ThreadItemConverterpour convertir les éléments et pièces jointes du thread ChatKit. -
stream_agent_response()pour convertir les mises à jour de l’agent en continu en événements ChatKit. -
simple_to_agent_input()pour le chemin de conversion de message par défaut.
Prerequisites
- Python 3.10 ou version ultérieure.
- Un framework web back-end tel que FastAPI.
- Node.js pour le front-end ChatKit.
- Clé de domaine ChatKit pour un domaine frontal de production.
Installer le package
pip install agent-framework-chatkit --pre
Créer un serveur ChatKit
Sous-classe ChatKitServer, créez l’agent Agent Framework et configurez un convertisseur pour les éléments de thread et les pièces jointes.
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,
)
Convertir et diffuser des réponses
Chargez l’historique des threads, convertissez-le en messages Agent Framework, exécutez l’agent en mode streaming et produisez des événements 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
Appelez agent.run(..., stream=True) d’abord, puis transmettez ce flux de réponse et l’ID de thread ChatKit à stream_agent_response(...). Ne transmettez pas l’agent lui-même au convertisseur.
ThreadItemConverter
StructuredInputItem conserve l’état et représente chaque question en tant que réponse, ignorée ou sans réponse dans un message utilisateur. Remplacez la méthode asynchrone structured_input_to_input() pour personnaliser la mise en forme ou la rédaction des réponses. Retournez un Message, une liste de messages ou None pour omettre l’élément.
Le convertisseur conserve également les valeurs finalisées GeneratedImageItem comme contexte pour le modèle lors des tours suivants. La conversion par défaut crée un utilisateur Message avec une préface de texte court et l’URI d’image généré. Les URI de données conservent leur type de média incorporé, les URL externes utilisent image/*et les éléments non terminés sans image sont ignorés.
Remplacez la méthode asynchrone generated_image_to_input() pour personnaliser le contexte d’image généré, par exemple la résolution d’une URL d’image privée. Retournez un Message, une liste de messages ou None pour omettre l’élément.
L’exemple complet illustre également les threads soutenus par SQLite, les chargements de fichiers, le stockage de pièces jointes, les actions et les widgets interactifs.
Avertissement
Le front-end ChatKit est chargé à partir du CDN d’OpenAI et effectue des requêtes sortantes vers des domaines OpenAI. Il ne peut actuellement pas être auto-hébergé et n’est pas adapté aux environnements à air gapé.