Чат-бот запісу ў клініку на LangGraph, SQLite і Streamlit крок за крокам
Будуем асістэнта запісу на прыём у клініку: табліцы SQLite, сэрвісныя функцыі, працэс LangGraph з дзевяці вузлоў на gpt-4o-mini, чат-інтэрфейс на Streamlit і рэжым нататніка.
Крыніцы: Agentic AI Project: Build a Customer Service Chatbot for a Clinic
Адміністрацыя клінікай трэбуе багато трыдзевога коордынацыі: падбіранне пацяўтакоў да лекароў, стэжылля за тым, хто і калі ўвольны, а таксама запобежэнне подвойных броніраў. Памочны асістэнт, які ведае дыялог, можа перадзяліць на сябе частку, якая стосуецца планавання апэксыў. У этым праекце такі асістэнт ствараецца з чатырох элементаў: LangGraph для керавання багаташагонавым дыялогам як машынай станаў, gpt-4o-mini ад OpenAI для розумення таго, што пішу людзі, SQLite для зберагання дадзэнняў, і Streamlit для вікна чату.
Проблема і план
Багатыя клінікі яшчэ выкарыстоўваюць ручныя працэсы або сумешанне разных інструментаў. Якія ў тым наследкі — дзейныя чаканні, суперсечання у планаванні, неравны тыраж працы персоналу і разчароўваючы досвід пацяўтакоў, якім трэба ўсілкія старанні, каб знайсці доступны лекар або правы спецыяліст. Клінікам патрэбна аўтаматызаваная сістэма, якая броніруе апэксы ў меры дыялогу, пры тым точна фіксуючы інфармацыю пра лекароў, пацяўтакоў і броніраў.
Чат-бот напамінак ведае корыстучага па складным сценарыю:
- Ён прывітаеся з корыстучым і запытае, чы хоча ён забронюваць прыемку.
- Ён паказвае доступныя медычныя спецыялізаціі.
- Ён паказвае лекара і часы, калі ён доступны.
- Ён збірае імя паціўента і номер тэлфона.
- Ён падтверджае броніраванне і зберагае яго ў базе дадзеных.
Рэзультатам являецца меньшая адміністрацыйная навантажэння, шырэйша броніраванне, адсутнасць суперактываў прыемак і болей павольны працэс для паціўніка. Готовы код арганізаваны ў папкі для дадзеных, служб, агента і інтэрфейса. Кожны элемент мае чыста вялікую задачу. LangGraph выбірае, які наступны крок у размове, мадэль адначытае вольны тэкст, SQLite запамячае, хто і калі, а Streamlit паказвае все гэта корыстучым у браузеры. Раздзелэнне гэтых задач дазволяе змяніць адзін элемент, напрыклад інтэрфейс, не тычась да рэшты.
Настройка
Створыце вярчынны абяканак. Каманда аднаковая на Mac, Linux і Windows:
python -m venv .venv
Актывацыя на Mac або Linux:
source .venv/bin/activate
У Windows з PowerShell:
.venv\Scripts\Activate.ps1
У прамптазе каманд Windows:
.venv\Scripts\activate
Калі ён актыўны, запыт тэрміналу пачынаецца з назвы сяродавішча, так:
(.venv) your-folder-name %
Далей йду залежнасці. Файл requirements.txt указвае Streamlit, python-dotenv, pydantic, pandas і іншыя бібліятэкі:
streamlit>=1.50.0
python-dotenv==1.0.0
pydantic==2.12.5
pandas==2.3.3
python-dateutil==2.8.2
langgraph>=1.0.7
openai>=2.16.0
pygraphviz==1.14
З’явіце іх з папкі праекта:
cd clinic-agent
pip install -r requirements.txt
На завершанне створыце файл .env у каталозе clinic-agent, які будзе мячыць ваш ключ OpenAI:
OPENAI_API_KEY=your_openai_api_key_here
Частка 1: база дадзеных (data/db.py)
Сістэма броніравання стоіць на свайом шарэ дадзеных. Клініка выкорыстоўвае SQLite з трыма тэбламі для лекароў, кліянтав і броніраванняў. Схема ёсць рэляцыйная і лёгкая для запытанняў пад час размовы: броніраванне адносіцца да аднаго лекара і аднаго кліянта, таму адна жоўтва паведамляе, хто прыходзіць да каго і калі. Нумеры тэлефонаў служаюць практычным ідэнтыфікаторам пацяўпінца, калі ўжо аднойчы звоніўшы не павінен стаць об’ектам дублікатнага запісу.
Лекарзы. Кожны рад — гэта лекарз з якойсь спецыялізацыяй і часамі працы, што дазволяе чат-боту запропанаваць правільнага спеціяліста пасля выбору спецыялізацыі. Поле doctor_id — унікальны ідэнтыфікатор, doctor_name — імя лекарза, speciality — спецыялізацыя (напрыклад, дэрматалог чы ўздоўжнік), а office_timing — часы працы лекарза. У дэмантэ выкананы пяць спецыялізацыяў: дэрматалог, ортапед, загальны лекарз, ўздоўжнік і спеціяліст па вушах, носе і горле; для кожнай з іх заданы вымышлены імены і часы працы. Дадаць новых лекарзоў ёсць проста.
Кліянты. У гэтай таблыцы зберагаецца усё, што неабходна для броніравання. Кожны раз, калі хтось броніруе, сістэма шукае кліента па номеру тэлефона і стварае новы запіс, якога няма. Поле customer_id, name і phone — гэта іменнік кліянта, імя і номер тэлефона.
Броніравання. Гэта центральная таблыца. Кожны рад з’яўляеся ў зв’язку лекара і кліента з конкрэтным датам і часам. Яе поля — booking_id, doctor_id і customer_id (пасыланні на два іншых таблыцы), appointment_date, appointment_time і status, напрыклад, Confirmed або Cancelled.
Скрапт выявляе гэтыя таблыцы і стварае файл clinic.db:
# data/db.py - Database initialization and operations
import sqlite3
import os
from datetime import datetime, timedelta
DB_PATH = os.path.join(os.path.dirname(__file__), "clinic.db")
def get_connection():
"""Get a database connection."""
return sqlite3.connect(DB_PATH)
def init_db():
"""Initialize the database with tables and sample data."""
conn = get_connection()
cursor = conn.cursor()
# Doctors table
cursor.execute("""
CREATE TABLE IF NOT EXISTS doctors (
doctor_id TEXT PRIMARY KEY,
doctor_name TEXT NOT NULL,
speciality TEXT NOT NULL,
office_timing TEXT NOT NULL
)
""")
# Customers table
cursor.execute("""
CREATE TABLE IF NOT EXISTS customers (
customer_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
phone TEXT NOT NULL
)
""")
# Bookings table
cursor.execute("""
CREATE TABLE IF NOT EXISTS bookings (
booking_id TEXT PRIMARY KEY,
doctor_id TEXT NOT NULL,
customer_id TEXT NOT NULL,
appointment_date TEXT NOT NULL,
appointment_time TEXT NOT NULL,
status TEXT NOT NULL,
FOREIGN KEY (doctor_id) REFERENCES doctors (doctor_id),
FOREIGN KEY (customer_id) REFERENCES customers (customer_id)
)
""")
# Insert sample doctors
doctors = [
("D1", "Dr. Anil Sharma", "General Physician", "10:00-14:00"),
("D2", "Dr. Neha Verma", "Dermatologist", "11:00-16:00"),
("D3", "Dr. Rohit Mehta", "Orthopedic", "09:00-13:00"),
("D4", "Dr. Kavita Rao", "Pediatrician", "10:00-15:00"),
("D5", "Dr. Sanjay Iyer", "ENT Specialist", "12:00-17:00"),
]
for doctor in doctors:
cursor.execute(
"INSERT OR IGNORE INTO doctors (doctor_id, doctor_name, speciality, office_timing) VALUES (?, ?, ?, ?)",
doctor
)
conn.commit()
conn.close()
if __name__ == "__main__":
print("Initializing database...")
init_db()
print("Database initialized successfully.")
Выканаце яго зярэйковыя падак спявах:
cd clinic-agent
python data/db.py
Гэта стварае clinic.db з трыма тэбламі. Паўтарны запуск скрыпты ёсць прыемны спосаб перазначэння прыкладных дадзеных пад час експерыментаў.
Частка 2: шар служб
Службы зберагаюць логіку бізнесу межы агента і базай дадзеных. Агент LangGraph не запытвае тэблы безпасяродна. Ён вызывае функціі служб, каб атрымаць інфармацыю пра лекароў, стварыць часовыя інтэрвалы і паўтарыць броніраванні. У дзялеўцы агента гэтыя функцыі выступаюць як інструменты, ўжо таму агент можа сфокусавацца на ходзе размовы. Такая супадзел таксама дазволяе тэставаць логіку окрэма. Вы можэце пераканацца ў стварэнні часовых інтэрвалоў і броніраванні, не вызываючы мовны мадэль, што самэ гэта і робіць наступны маленькі скрыпт.
Сэрвіс доктара (services/doctor_service.py) керуе спецыяльнасцямі, доктарамі і часовымі інтэрваламі. Ён вяртае спіс спецыяльнасцей з базы дадзеных, запошчае доктароў для выбранай спецыяльнасці, стварае часовыя інтэрвалы на адпаведнасць гадзінам працы доктара і перакладае часы адображэння, такія як 2:00 PM, у 24-гадзінны фармат для зберагання.
# services/doctor_service.py - Doctor operations
from data.db import get_all_doctors, get_doctor_by_speciality, get_doctor_by_id
def get_specialities_list():
"""Get list of all specialities."""
doctors = get_all_doctors()
# Return unique specialities
return list(dict.fromkeys([doc[2] for doc in doctors]))
def get_doctor_info(speciality):
"""Get doctor information by speciality."""
doctor = get_doctor_by_speciality(speciality)
if doctor:
return {
"doctor_id": doctor[0],
"doctor_name": doctor[1],
"speciality": doctor[2],
"office_timing": doctor[3]
}
return None
def generate_time_slots(office_timing):
"""Generate hourly time slots from office timing string.
Args:
office_timing: String like "11:00-16:00"
Returns:
List of time slots like ["11:00 AM", "12:00 PM", ...]
"""
start_time, end_time = office_timing.split("-")
start_hour = int(start_time.split(":")[0])
end_hour = int(end_time.split(":")[0])
slots = []
for hour in range(start_hour, end_hour):
if hour < 12:
suffix = "AM"
display_hour = hour if hour > 0 else 12
elif hour == 12:
suffix = "PM"
display_hour = 12
else:
suffix = "PM"
display_hour = hour - 12
slots.append(f"{display_hour}:00 {suffix}")
return slots
def parse_time_slot(slot_str):
"""Parse time slot string to 24-hour format.
Args:
slot_str: String like "1:00 PM"
Returns:
String like "13:00"
"""
time_part, suffix = slot_str.split(" ")
hour, minute = time_part.split(":")
hour = int(hour)
if suffix == "PM" and hour != 12:
hour += 12
elif suffix == "AM" and hour == 12:
hour = 0
return f"{hour:02d}:{minute}"
Сэрвіс бронювання (services/booking_service.py) керуе кліентамі і бронюваннямі. Ён знаходзіць існуючага кліента па номерэ тэлефона або стварае новага, шукае ужо зайнятыя часовыя прыводы доктара на пэўную дату, фільтруе іх, ўсталяючы вольныя прыводы, генеруе унікальны ідэнтыфікатор бронювання і зберагае пакончныя даныя прызначэння: доктар, паціўент, дата і час.
# services/booking_service.py - Booking operations
import uuid
from datetime import datetime
from data.db import (
create_customer,
create_booking,
get_customer_by_phone,
get_bookings_by_doctor_and_date,
get_booking_by_id
)
from services.doctor_service import parse_time_slot
def get_or_create_customer(name, phone):
"""Get existing customer or create new one."""
customer = get_customer_by_phone(phone)
if customer:
return customer[0] # Return customer_id
customer_id = f"CUST-{uuid.uuid4().hex[:6].upper()}"
create_customer(customer_id, name, phone)
return customer_id
def get_available_slots(doctor_id, office_timing):
"""Get available time slots for a doctor for today.
Args:
doctor_id: Doctor ID
office_timing: Office timing string like "11:00-16:00"
Returns:
List of available time slots
"""
from services.doctor_service import generate_time_slots
today = datetime.now().strftime("%Y-%m-%d")
all_slots = generate_time_slots(office_timing)
# Get booked slots
booked_times = get_bookings_by_doctor_and_date(doctor_id, today)
# Filter out booked slots
available = []
for slot in all_slots:
slot_24h = parse_time_slot(slot)
if slot_24h not in booked_times:
available.append(slot)
return available
def confirm_booking(doctor_id, customer_name, customer_phone, time_slot, appointment_date=None):
"""Confirm a booking.
Args:
doctor_id: Doctor ID
customer_name: Customer name
customer_phone: Customer phone
time_slot: Time slot like "1:00 PM"
appointment_date: Optional date in YYYY-MM-DD format. Defaults to today.
Returns:
Booking ID
"""
# Get or create customer
customer_id = get_or_create_customer(customer_name, customer_phone)
# Generate booking ID
booking_id = f"BKG-{uuid.uuid4().hex[:6].upper()}"
# Format appointment time
if not appointment_date:
appointment_date = datetime.now().strftime("%Y-%m-%d")
appointment_time = parse_time_slot(time_slot)
# Create booking
create_booking(booking_id, doctor_id, customer_id, appointment_date, appointment_time)
return booking_id
Перакантраванне служб
Маленькі скрыпты працююць з обумовленымі службамі:
# test/test_service.py - to test the services created
from pathlib import Path
import sys
# Allow running this file directly: `python test/test_service.py`.
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from services.doctor_service import get_specialities_list, generate_time_slots
from services.booking_service import confirm_booking
# Get all specialities
specialities = get_specialities_list()
print("Available specialities:", specialities)
# Generate time slots for a doctor (11:00 AM - 4:00 PM)
slots = generate_time_slots("11:00-16:00")
print("Available slots:", slots)
# Confirm a booking
booking_id = confirm_booking(
doctor_id="D1",
customer_name="John Doe",
customer_phone="9876543210",
time_slot="2:00 PM"
)
print(f"Booking confirmed: {booking_id}")
Заўяжыце яго:
cd clinic-agent
python test_service.py
Тэрмінал выдрукавае, сярод іншаго, спіс спецыяльнасцей і часовых прыводзов для лекара:
(.venv) (base) my-mac clinic-agent % python test_service.py
Available specialities: ['General Physician', 'Dermatologist', 'Orthopedic', 'Pediatrician', 'ENT Specialist']
Available slots: ['11:00 AM', '12:00 PM', '1:00 PM', '2:00 PM', '3:00 PM']
Booking confirmed: BKG-C4F60A
Калі все выглядае правільна, сервісы працуюць, і можа пачыцца цікавая частка. Адзінство рэальных спецыялітэтаў і рэальных часоў запісу ў выходным дадзенні паўтарае, што база дадзенаў, логіка выбору лекара і логіка бронювання ўсе суспяльна згодныя, прычаму яшчэ не была дадана жадная логіка ведчыцтва размовы.
Частка 3: агент
Серцам чат-бота ёсць агент LangGraph, які прыводзіць бронюванне па калькольных этапах.
Стан бронювання
Об’ект стану — гэта структураваная памяць для размовы. Ён фіксуе тэкушчую стадыю, такую як прыветственне або выбор лекара, запам’ятовуе выбраныя параметры, такія як спецыялітэт, лекар і час запісу, а таксама зберагае дакладныя відомасці кліента і бронювання, каб агент могаў завершыць запіс.
# agents/booking_agent.py - LangGraph agent implementation
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, END
from openai import OpenAI
import os
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class BookingState(TypedDict):
"""State for the booking conversation."""
messages: List[dict] # Chat history
stage: str # greeting, select_speciality, select_doctor, etc.
selected_speciality: Optional[str] # Chosen medical specialty
selected_doctor: Optional[dict] # Selected doctor details
selected_date: Optional[str] # Appointment date
selected_slot: Optional[str] # Time slot
customer_name: Optional[str] # Customer name
customer_phone: Optional[str] # Customer phone
booking_id: Optional[str] # Confirmation ID
available_options: List[str] # UI options
def create_initial_state():
"""Create initial state for the conversation."""
return {
"messages": [],
"stage": "greeting",
"selected_speciality": None,
"selected_doctor": None,
"selected_date": None,
"selected_slot": None,
"customer_name": None,
"customer_phone": None,
"booking_id": None,
"available_options": []
}
Памочнік LLM
Функцыя call_llm аб’еднвае кантакты з модэлем OpenAI. Яна адправляе запыткі сістэмы і корыстніка, вяртае створаны тэкст і дбае пра тое, каб кожны вызов модэля быў аднаковым і можна было яго перыварыць.
# agents/booking_agent.py - LangGraph agent implementation
def call_llm(
system_prompt: str,
user_prompt: str,
*,
model: str = "gpt-4o-mini",
temperature: float = 0,
max_tokens: int = 50,
) -> str:
"""
Centralized helper for all LLM calls.
Returns the assistant's response
"""
try:
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
temperature=temperature,
max_tokens=max_tokens,
)
return response
except Exception as e:
print(f"LLM call error: {e}")
return ""
Вузлы
Кожны вузел — гэта адміністратыўны крок у процэсе броніравання. Вузлы керуюць тым этапам, які ўособляюць, чытаюць даныя, якія вводзі корыстувачы, і адкорэгуюць статус, што дапамагае падтрымваць модульную структуру процэса. Іх ёсць дзевяць:
greeting_nodeпачынае чат і запытае, чы хочаць корыстувач узначыць час прыемы.select_speciality_nodeпаказвае наявныя спецыялізацыі.select_doctor_nodeпрадстаўляе лекароў у выбранай спецыялізацыі.select_date_nodeзапытае, які дзень падходзіць корыстувачу.select_slot_nodeпаказвае вільныя часы таго дня.confirm_nodeпрасіць корыстувача пастаравацца свой выбор.collect_details_nodeзапытае імя і номер тэлфона.completed_nodeзавершае успешнае броніравання.
cancelled_node выкорыстоўваецца для абработкі сітуацыі, калі корыстнік відмовляецца.Стварэнне графа
Функцыя build_booking_graph дадае гэтыя вузлы, задае умовныя перэходы між ямі і такім чынам візначае загальны маршрут размовы.
# agents/booking_agent.py - LangGraph agent implementation
from langgraph.checkpoint.memory import MemorySaver
def build_booking_graph():
"""Build the LangGraph workflow."""
workflow = StateGraph(BookingState)
# Add all nodes
workflow.add_node("greeting", greeting_node)
workflow.add_node("select_speciality", select_speciality_node)
workflow.add_node("select_doctor", select_doctor_node)
workflow.add_node("select_date", select_date_node)
workflow.add_node("select_slot", select_slot_node)
workflow.add_node("confirm", confirm_node)
workflow.add_node("collect_details", collect_details_node)
workflow.add_node("completed", completed_node)
workflow.add_node("cancelled", cancelled_node)
# Set entry point
workflow.set_entry_point("greeting")
# Add conditional edges based on routing
workflow.add_conditional_edges(
"greeting",
llm_router,
{
"greeting": "greeting",
"select_speciality": "select_speciality",
"cancelled": "cancelled"
}
)
# Similar conditional edges for other nodes...
# Final edges to END
workflow.add_edge("completed", END)
workflow.add_edge("cancelled", END)
# Compile with checkpointer for session management
return workflow.compile(checkpointer=MemorySaver())
# Create the compiled graph
booking_graph = build_booking_graph()
Корты сценары выканаюць графік у вачынку:
#agents/save_langgraph_flow.py
"""Save the clinic booking LangGraph flow to png format in this folder."""
from pathlib import Path
import sys
# Ensure imports work whether the script is run from project root or this folder.
AGENTS_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = AGENTS_DIR.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from agents.booking_agent import booking_graph # noqa: E402
def save_graph_files() -> None:
"""Export graph as PNG."""
graph = booking_graph.get_graph()
png_path = AGENTS_DIR / "langgraph_flow.png"
png_data = graph.draw_mermaid_png()
png_path.write_bytes(png_data)
print(f"Saved PNG flow to: {png_path}")
if __name__ == "__main__":
save_graph_files()
Заўяжыце яго:
cd clinic-agent
python save_langgraph_flow.py
Ён зберагае дыяграму ў папку agents, што ўтрымлівае зручны спосаб разглядзець этапы і ўзаеці межы ўсіх іх за мгновенне, перш чым даваць іх у выкарыстоўванне рэальным корыстнікам.
Апрацоўка паведамленняў
process_message прымае даны з інтэрфейса, праходзіць іх чэрзь граф, і вяртае апошні стан разам з адказам асістэнта. Ён прымае ID задання і значэнне падзеўкі за замовчаннем, так што кожна сесія браузера чы аперацыя з нотбукам мае свой сахаваны прыглед працы ў працэсе броніравання.
# agents/booking_agent.py - LangGraph agent implementation
def process_message(state: BookingState, user_message: str, thread_id: str = "default_session") -> BookingState:
"""Process a user message through the booking graph."""
config = {"configurable": {"thread_id": thread_id}}
# Check if the graph is currently interrupted
current_state = booking_graph.get_state(config)
if current_state.tasks and current_state.tasks[0].interrupts:
# Resume the graph with the user's message
result = booking_graph.invoke(Command(resume=user_message), config=config)
else:
# No interrupt, so start/continue normally
# Add user message to state (unless it's an initial trigger)
if user_message.lower() != "hi" or state["messages"]:
# Avoid duplicate user messages if already added
if not state["messages"] or state["messages"][-1].get("content") != user_message:
state["messages"].append({
"role": "user",
"content": user_message
})
# Run the graph
result = booking_graph.invoke(state, config=config)
# Update available_options and ensure message is in history
snapshot = booking_graph.get_state(config)
if snapshot.tasks and snapshot.tasks[0].interrupts:
interrupt_value = snapshot.tasks[0].interrupts[0].value
# Handle both dict and string interrupt values
msg_content = ""
options = []
if isinstance(interrupt_value, dict):
msg_content = interrupt_value.get("content", "")
options = interrupt_value.get("available_options", [])
else:
msg_content = str(interrupt_value)
# Ensure the interrupt message is in the chat history
if msg_content:
# Check if it was already added by the node
last_msg_content = result["messages"][-1].get("content", "") if result["messages"] else ""
if last_msg_content != msg_content:
result["messages"].append({
"role": "assistant",
"content": msg_content,
"options": options
})
else:
# If already added, just update it with options if missing
result["messages"][-1]["options"] = options
result["available_options"] = options
else:
# If not interrupted, use whatever set in state, or default to empty
if "available_options" not in result:
result["available_options"] = []
return result
Тэставанне агента
Скрыпт тэста керуе агентам з тэрміналу:
# test/test_agent.py - test agent
# Initialize state
from pathlib import Path
import sys
# Allow running this file directly: `python test/test_agent.py`.
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from agents.booking_agent import create_initial_state, process_message
state = create_initial_state()
# Process messages
state = process_message(state, "Hi", thread_id="session_1")
print(state["messages"][-1]["content"])
state = process_message(state, "I want to book", thread_id="session_1")
print(state["available_options"])
Заўяжыце яго:
cd clinic-agent
python test/test_agent.py
Ягоўныя маюць выдаваць доступныя опцыі, што паказвае, што графік рэагуе. У гэты момент весь працэс працюе з тэрміналу, і не хватает толькі прыказа.
Частка 4: інтэрфейс Streamlit
Інтэрфейс корыстніка адкрывае вікно чату у рэальным часе, кнопкі з опцыямі, якія можна нажаць, а таксама сесіі, ідэнтыфікаваныя унікальнымі ID. Кнопкі маюць важлівую ролю ў такім прадзе, таму што выбір спецыялізаціі або часовага прыемку за адны нажатак ёсць швэйцарскі і менш схільны да памылак, чым ўводзіць іх вручную.
# ui/chat_ui.py - Streamlit chatbot interface
"""Streamlit UI for the clinic booking chatbot."""
import streamlit as st
from agents.booking_agent import create_initial_state, process_message
from data.db import init_db
def initialize_session():
"""Initialize session state."""
if "state" not in st.session_state:
st.session_state.state = create_initial_state()
if "initialized" not in st.session_state:
st.session_state.initialized = False
if "session_id" not in st.session_state:
import uuid
st.session_state.session_id = str(uuid.uuid4())
def display_chat_history():
"""Display the chat history with persistent options and styling."""
messages = st.session_state.state.get("messages", [])
for i, message in enumerate(messages):
if message["role"] == "assistant":
with st.chat_message("assistant"):
st.markdown(message["content"])
# Show options if they exist
options = message.get("options", [])
if options:
# If this is the last message in history, show as clickable buttons
if i == len(messages) - 1 and st.session_state.state["stage"] not in ["completed", "cancelled"]:
st.markdown("---")
# Create columns for buttons
cols = st.columns(min(len(options), 3))
for idx, option in enumerate(options):
col_idx = idx % 3
with cols[col_idx]:
if st.button(option, key=f"btn_{i}_{idx}", use_container_width=True):
handle_user_input(option)
else:
# For older messages, show options as pills/text to keep history
options_str = " ".join([f"`{opt}`" for opt in options])
st.markdown(f"**Available options:** {options_str}")
else:
with st.chat_message("user"):
st.markdown(message["content"])
def handle_user_input(user_input: str):
"""Handle user input and process through agent."""
# Process the message
st.session_state.state = process_message(
st.session_state.state,
user_input,
thread_id=st.session_state.session_id
)
# Rerun to update UI
st.rerun()
def run_chat_ui():
"""Run the chat UI."""
# Page config
st.set_page_config(
page_title="CarePlus Clinic - Book Appointment",
page_icon="🏥",
layout="centered"
)
# Custom CSS for distinction between messages
st.markdown("""
<style>
[data-testid="stChatMessageUser"] {
flex-direction: row-reverse;
text-align: right;
background-color: #e0f2f1;
border-radius: 15px 15px 0px 15px;
}
[data-testid="stChatMessageAssistant"] {
background-color: #f5f5f5;
border-radius: 15px 15px 15px 0px;
}
</style>
""", unsafe_allow_html=True)
# Initialize database
init_db()
# Initialize session
initialize_session()
# Header
st.title("🏥 CarePlus Clinic")
st.markdown("*Book your doctor appointment easily*")
st.markdown("---")
# Send initial greeting if not initialized
if not st.session_state.initialized:
st.session_state.state = process_message(
st.session_state.state,
"Hi",
thread_id=st.session_state.session_id
)
st.session_state.initialized = True
st.rerun()
# Display chat history
display_chat_history()
# Chat input (only show if not completed)
if st.session_state.state["stage"] not in ["completed", "cancelled"]:
if prompt := st.chat_input("Type your message here..."):
handle_user_input(prompt)
else:
# Show restart button after completion
st.markdown("---")
if st.button("🔄 Start New Booking", use_container_width=True):
st.session_state.state = create_initial_state()
st.session_state.initialized = False
st.rerun()
Частка 5: тэпіс падхоўду
app.py — гэта пачаткова точка, якая запускае весь прыемлік:
# app.py - Main entry point for the application
from ui.chat_ui import run_chat_ui
if __name__ == "__main__":
run_chat_ui()
Два спосабы ўработы з яным
Як веб-дзеянне Streamlit, што ўсё простае:
streamlit run app.py
Ён ачыяецца па адрэсе http://localhost:8501.
У Jupyter notebook, які падходзіць для разработкі і эксперыментаў. Спачатку імпортуйце функцыі:
# clinic-agent.ipynb
from services.doctor_service import get_specialities_list, get_doctor_info, generate_time_slots
from services.booking_service import confirm_booking
from agents.booking_agent import (
BookingState,
create_initial_state,
build_booking_graph,
process_message
)
Створыце граф і адказваеце яго структуру:
# clinic-agent.ipynb
# Initialize the booking graph
booking_graph = build_booking_graph()
## Visualize the booking graph structure
from IPython.display import Image, display
png_bytes = booking_graph.get_graph().draw_mermaid_png()
display(Image(png_bytes))
Потым запусціце інтэрактыўную сесію чату ўнутранік нотбуку:
# clinic-agent.ipynb
from langgraph.types import Command
def run_booking_session(graph, thread_id="notebook_session", reset=False):
config = {"configurable": {"thread_id": thread_id}}
# 1. Start or Reset logic
current_state = graph.get_state(config)
if reset or not current_state.values:
print(f"--- {'🔄 Resetting' if reset else '🆕 Initializing'} Session ---")
# Using invoke() here kicks off the 'greeting' node immediately
graph.invoke(create_initial_state(), config=config)
print("---⚕⚕ Starting CarePlus Booking Session ---")
last_displayed_message_idx = -1 # Track which messages have been displayed
while True:
state = graph.get_state(config)
# Display any new assistant messages that haven't been shown yet
# (This handles guardrail/off-topic responses)
if state.values and state.values.get('messages'):
messages = state.values['messages']
for idx in range(last_displayed_message_idx + 1, len(messages)):
msg = messages[idx]
if msg.get("role") == "assistant":
print(f"\n[AI]: {msg['content']}")
last_displayed_message_idx = len(messages) - 1
# 2. Check for Interrupts
if state.tasks and state.tasks[0].interrupts:
interrupt_info = state.tasks[0].interrupts[0].value
# --- FIX: Safely handle both String and Dict interrupts ---
if isinstance(interrupt_info, dict):
ai_message = interrupt_info.get('content', 'No message content')
options = interrupt_info.get('available_options', [])
else:
ai_message = interrupt_info
options = []
print(f"\n[AI]: {ai_message}")
if options:
print(f"Options: {', '.join(options)}")
# -------------------------------------------------------
user_input = input("\n[YOU]: ")
print(f"[YOU]: {user_input}")
# Resume the graph with the user's input
graph.invoke(Command(resume=user_input), config=config)
# 3. Check if the graph has finished
elif not state.next:
# Before ending, check if there's a final assistant message to print
if state.values and state.values.get('messages') and state.values['messages'][-1]["role"] == "assistant":
if last_displayed_message_idx < len(state.values['messages']) - 1:
print(f"\n[AI]: {state.values['messages'][-1]['content']}")
print("\n--- ⚑⚑ Session Ended ---")
break
# 4. If nodes are pending but no interrupt, let them run (the gas pedal)
else:
graph.invoke(None, config=config)
# IMPORTANT: Set reset=True only when you want to wipe the history.
# Set it to False to actually continue the conversation!
run_booking_session(booking_graph, reset=True)
Дапаможнік для зошыта запускае сесію броніравання на адзначанай нітку і можа быць скасаваны, ўбачымае пачатак працы. У зошыце поле для вводу выклікаецца ў верхней часты, над клеткамі. У рэпазітарые включаны зошыт clinic-agent.ipynb, які запускае весь працэс.
Што дае вам готовы бот
Рэзультатам являецца чат-бот для броніравання, створанный на базе LangGraph, OpenAI і SQLite з фронтэндам у фармаце Streamlit. Рабочы працэс, керованы станам, памагае ўжоўцам выбраць спецыялізацію і падтвердзіць час зустрэчы, а шары (база дадзеных, сервісы, логіка агента і UI) застаюцца окрэмымі, таму прыклад аплікацыі ёсць модульным і лёгкім у розшырэнні. Услявольна, калі шары чыста аддзелены, кожная паследоўная змянена тычыцца толькі адного з іх. Серэд наступных крокаў — інтэграцыя з календарам, паведамлення, падтрымка калькі мов і платформы для перасылкі паведамленняў. Цэе практычны прыклад адузеўнення сучасных фрэймворкаў AI і простых веб-інструментаў у корыстную аплікацыю для сферы охорны здароў’я.