Chainlit for AI Engineering: WebSockets, Multi-Modal & Step Tracing

import chainlit as cl
import asyncio

@cl.step(type="tool", name="Database Ledger Query")
async def execute_ledger_query(account_id: str):
  """Sub-step visualized in the Chainlit UI trace drawer."""
  await asyncio.sleep(0.3) # Simulate database query latency
  return f"Ledger data for {account_id}: $1,450,000.00 USD available balance."

@cl.on_chat_start
async def start():
  cl.user_session.set("session_id", "sess_90812")
  await cl.Message(content="Welcome to the Esaholic Enterprise Agent. How can I assist with your financial telemetry today?").send()

@cl.on_message
async def main(message: cl.Message):
  # Step 1: Execute tool query with visual tracing
  ledger_result = await execute_ledger_query("ACC-4491")
  
  # Step 2: Initialize streaming response message
  msg = cl.Message(content="")
  await msg.send()

  # Step 3: Stream tokens to client interface
  sample_text = f"Based on the query result ({ledger_result}), your account is fully collateralized and healthy."
  for token in sample_text.split(" "):
      await msg.stream_token(token + " ")
      await asyncio.sleep(0.03)

  await msg.update()