Nota
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare ad accedere o modificare le directory.
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare a modificare le directory.
agent-framework-chatkit converte gli elementi del thread OpenAI ChatKit in messaggi di Agent Framework e converte gli aggiornamenti degli agenti trasmessi in eventi ChatKit. Usalo quando vuoi un frontend ChatKit con un backend Python di Agent Framework.
L'integrazione fornisce:
-
ThreadItemConverterper convertire gli elementi del thread e gli allegati di ChatKit. -
stream_agent_response()per convertire gli aggiornamenti degli agenti trasmessi in eventi ChatKit. -
simple_to_agent_input()per il percorso di conversione dei messaggi predefinito.
Prerequisites
- Python 3.11 o versione successiva.
- Un framework web backend, ad esempio FastAPI.
- Node.js per il front-end di ChatKit.
- Un progetto Microsoft Foundry con un modello distribuito.
- Autenticazione di interfaccia della riga di comando di Azure (
az login). - Una chiave di dominio ChatKit per un dominio front-end di produzione.
Installare il pacchetto
pip install agent-framework-chatkit agent-framework-foundry --pre
Configurare il modello Foundry
L'esempio crea l'agente con FoundryChatClient e AzureCliCredential.
Impostare l'endpoint del progetto e la distribuzione del modello prima di avviare il back-end:
export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL="<model-deployment>"
Creare un server ChatKit
Sottoclasse ChatKitServer, creare l'agente di Agent Framework e configurare un convertitore per gli elementi e gli allegati del thread.
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,
)
logger.info("WeatherChatKitServer initialized")
Convertire e trasmettere le risposte
Caricare la cronologia dei thread, convertirla in messaggi di Agent Framework, eseguire l'agente in modalità di streaming e restituire gli eventi di ChatKit.
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.
"""
if input_user_message is None:
logger.debug("Received None user message, skipping")
return
logger.info(f"Processing message for thread: {thread.id}")
try:
# Collect weather data from get_weather and track the city selector flag for this request
weather_results: list[WeatherData] = []
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. get_weather appends to weather_results.
agent_stream = self.weather_agent.run(
agent_messages,
stream=True,
function_invocation_kwargs={"weather_results": weather_results},
)
# Create an intercepting stream that detects the city selector marker while passing through updates
async def intercept_stream() -> AsyncIterator[AgentResponseUpdate]:
nonlocal show_city_selector
async for update in agent_stream:
# Check for the city selector marker in function results
for content in update.contents:
if content.type == "function_result" and content.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
# If weather data was collected during the tool call, create a widget
if weather_results:
weather_data = weather_results[-1]
logger.info(f"Creating weather widget for location: {weather_data.location}")
# Create weather widget
widget = render_weather_widget(weather_data)
copy_text = weather_widget_copy_text(weather_data)
# Stream the widget
async for widget_event in stream_widget(thread_id=thread.id, widget=widget, copy_text=copy_text):
Chiamare agent.run(..., stream=True) prima, quindi passare il flusso di risposta e l'ID del thread ChatKit a stream_agent_response(...). Non passare l'agente al convertitore.
ThreadItemConverter mantiene StructuredInputItem lo stato e rappresenta ogni domanda come risposta, ignorata o senza risposta in un messaggio utente. Eseguire l'override del metodo asincrono structured_input_to_input() per personalizzare la formattazione o oscurare le risposte. Restituire un Message, un elenco di messaggi oppure None per omettere l'elemento.
Il convertitore preserva anche i valori completati GeneratedImageItem come contesto del modello nei turni successivi. La conversione predefinita crea un utente Message con una prefazione di testo breve e l'URI dell'immagine generata. Gli URI dati mantengono il tipo di supporto incorporato, gli URL esterni usano image/*e gli elementi non finiti senza un'immagine vengono ignorati.
Eseguire l'override del metodo asincrono generated_image_to_input() per personalizzare il contesto di immagine generato, ad esempio la risoluzione di un URL di immagine privata. Restituire un Message, un elenco di messaggi oppure None per omettere l'elemento.
L'esempio completo illustra anche thread supportati da SQLite, caricamenti di file, archiviazione allegati, azioni e widget interattivi.
Avvertimento
Il front-end di ChatKit viene caricato dalla rete CDN di OpenAI ed effettua richieste in uscita ai domini OpenAI. Attualmente non può essere self-hosted e non è adatto ad ambienti isolati.