Bot do rezerwacji w klinice z LangGraph, SQLite i Streamlit – krok po kroku
Stwórz asystenta do rezerwacji wizyt w klinice: tabele SQLite, funkcje służące do obsługi usług, przepływ pracy LangGraph składający się z dziewięciu węzłów z użyciem gpt-4o-mini, interfejs czatowy w Streamlit oraz tryb notatnika.
Źródła: Agentic AI Project: Build a Customer Service Chatbot for a Clinic
Zarządzanie kliniką wymaga wielu żmudnych działań koordynacyjnych: dopasowywania pacjentów do lekarzy, śledzenia, kto jest wolny w danym czasie, oraz unikania podwójnych umówień. Asystent konwersacyjny może przejąć tę część obowiązków związanych z planowaniem wizyt. Ten projekt składa się z czterech elementów: LangGraph do prowadzenia wieloetapowej rozmowy jako maszyny stanów, gpt-4o-mini od OpenAI do rozumienia tego, co ludzie piszą, SQLite do przechowywania danych oraz Streamlit do stworzenia okna czatu.
Problem i plan
Wiele klinik nadal polega na procesach ręcznych lub mieszaninie różnych narzędzi. Skutkami tego są długie czekanie, konflikty w planowaniu, nierównomierny obciążenie personelu oraz frustrujące doświadczenie pacjentów, którzy mają trudności nawet z znalezieniem dostępnego lekarza lub odpowiedniej specjalizacji. Klinika potrzebuje zautomatyzowanego systemu, który umożliwia planowanie wizyt poprzez naturalną rozmowę, jednocześnie utrzymując dokładne zapisy dotyczące lekarzy, pacjentów i umówień.
Bot rozmówczy prowadzi użytkownika przez krótki scenariusz:
- Witaje użytkownika i pyta, czy chce umówić się na wizytę.
- Wyświetla dostępne specjalizacje medyczne.
- Pokazuje lekarza oraz dostępne godziny wizyt.
- Zbiera imię i numer telefonu pacjenta.
- Potwierdza rezerwację i zapisuje ją w bazie danych.
Korzyścią jest mniejsze obciążenie administracyjne, szybsza rezerwacja, brak konfliktujących terminów oraz płynniejszy proces obsługi pacjenta. Gotowy kod jest zorganizowany w folderach dla danych, usług, agenta oraz interfejsu użytkownika. Każda część ma określone zadanie. LangGraph decyduje, który krok rozmowy nastąpi dalej, model interpretuje tekst, SQLite zapamiętuje, kto i kiedy, a Streamlit pokazuje to wszystko użytkownikowi w przeglądarce. Oddzielanie tych zadań umożliwia zmianę jednej części, na przykład interfejsu, bez wpływu na pozostałe elementy.
Ustawienia
Stwórz środowisko wirtualne. Polecenie jest takie samo na Mac, Linux i Windows:
python -m venv .venv
Aktywuj to na Macu lub Linuksu:
source .venv/bin/activate
W systemie Windows z PowerShell:
.venv\Scripts\Activate.ps1
W wierszu poleceń Windowsa:
.venv\Scripts\activate
Gdy zostanie aktywowane, pole wprowadzania komend terminala zaczyna się od nazwy środowiska, w ten sposób:
(.venv) your-folder-name %
Następnie przychodzą zależności. Plik requirements.txt określa wersje Streamlit, python-dotenv, pydantic, pandas oraz innych bibliotek:
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
Zainstaluj je z folderu projektu:
cd clinic-agent
pip install -r requirements.txt
Na koniec utwórz plik .env w katalogu clinic-agent, który będzie zawierał twój klucz OpenAI:
OPENAI_API_KEY=your_openai_api_key_here
Część 1: baza danych (data/db.py)
System rezerwacji opiera się na warstwie danych. Klinika używa SQLite z trzema tabelami przeznaczonymi dla lekarzy, pacjentów oraz rezerwacji. Schemat jest relacyjny i ułatwia wyszukiwanie informacji podczas rozmowy: rezerwacja odnosi się do jednego lekarza i jednego pacjenta, więc jedno zapytanie pokazuje, kto przychodzi do kogo i kiedy. Numery telefonów pełnią rolę praktycznego identyfikatora pacjenta, aby osoba dzwoniąca ponownie nie trafiła na duplikat wpisu.
Lekarze. Każdy wiersz przedstawia lekarza z specjalizacją i godzinami pracy, co umożliwia botowi rozmówczemu zaproponowanie odpowiedniej osoby po wyborze specjalizacji. Pola to doctor_id (unikalny identyfikator), doctor_name, speciality (np. dermatolog lub pediatra) oraz office_timing (godziny pracy lekarza). W demonstracji uwzględniono pięć specjalizacji: dermatolog, ortopeda, lekarz rodzinny, pediatra i specjalista ORL, z wymyślonymi nazwami i godzinami pracy dla każdej z nich. Dodanie kolejnych specjalizacji jest proste.
Klienci. Ta tabela zawiera informacje niezbędne do dokonywania rezerwacji. Za każdym razem, gdy ktoś dokonuje rezerwacji, system wyszukuje klienta pod numerem telefonu i tworzy nowy rekord, jeśli taki jeszcze nie istnieje. Pola to customer_id, name i phone.
Zamówienia. Tabela centralna. Każdy wiersz łączy lekarza i klienta z konkretną datą oraz godziną. Jej pola to booking_id, doctor_id i customer_id (odniesienia do pozostałych dwóch tabel), appointment_date, appointment_time oraz status, np. Confirmed lub Cancelled.
Skrypt definiuje te tabele i tworzy plik 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.")
Zrób to z folderu projektu:
cd clinic-agent
python data/db.py
To tworzy plik clinic.db zawierający trzy tabele. Ponowne uruchomienie skryptu to szybki sposób na sformatowanie danych przykładowych podczas eksperymentowania.
Część 2: warstwa usług
Usługi przechowują logikę biznesową pomiędzy agentem a bazą danych. Agent LangGraph nie wysyła zapytań bezpośrednio do tabel – zamiast tego wywołuje funkcje usługowe w celu pobierania informacji o lekarzach, generowania godzin dostępnych oraz potwierdzania rezerwacji. W takim projekcie agenta te funkcje pełnią rolę narzędzi, umożliwiając agentowi skupienie się na przebiegu rozmowy. Taki podział sprawia również, że logikę można testować oddzielnie – można sprawdzić proces generowania godzin i rezerwacji bez konieczności wywoływania modelu językowego, co dokładnie robi następny krótki skrypt.
Służba lekarska (services/doctor_service.py) zajmuje się specjalnościami, lekarzami oraz godzinami przyjęć. Zwraca listę specjalności z bazy danych, pobiera lekarzy do wybranej specjalności, tworzy godziny przyjęć na podstawie godzin pracy lekarza oraz konwertuje godziny wyświetlania, takie jak 14:00, na format 24-godzinny w celu przechowywania.
# 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}"
Służba rezerwacji (services/booking_service.py) zarządza klientami i rezerwacjami. Znajduje istniejącego klienta pod numerem telefonu lub tworzy nowego, sprawdza dostępne terminy u lekarza w danym dniu, wyklucza te już zajęte, aby pozostały wolne terminy, generuje unikalny identyfikator rezerwacji oraz zapisuje ostateczne dane umowy: lekarza, pacjenta, datę i godzinę.
# 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
Sprawdzanie usług
Niewielki skrypt testuje obie usługi:
# 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}")
Zrób to:
cd clinic-agent
python test_service.py
Terminal wyświetla między innymi listę specjalizacji oraz dostępne terminy dla lekarza:
(.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
Jeśli wszystko wygląda poprawnie, usługi działają i można rozpocząć ciekawszą część. Widok rzeczywistych specjalizacji oraz dostępnych terminów potwierdza, że baza danych, logika lekarza oraz logika rezerwacji są ze sobą spójne, zanim zostanie dodana logika rozmowy.
Część 3: agent
Sercem chatbotu jest agent LangGraph, który przeprowadza rezerwację przez kilka etapów.
Stan rezerwacji
Obiekt stanu to strukturyzowana pamięć rozmowy. Zapisuje aktualny etap, np. powitanie lub wybór lekarza, pamięta dokonane wybory takie jak specjalizacja, lekarz i termin, a także przechowuje dane klienta i rezerwacji, aby agent mógł zakończyć umówienie wizyty.
# 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": []
}
Pomocnik LLM
Jedna funkcja call_llm centralizuje komunikację z modelem OpenAI. Wysyła komendy systemowe i użytkownika, zwraca wygenerowany tekst oraz zapewnia, że każde wezwanie modelu jest spójne i nadaje się do ponownego użycia.
# 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 ""
Węzły
Każdy węzeł stanowi jeden krok w procesie rezerwacji. Węzły obsługują odpowiedni etap, odczytują dane wprowadzone przez użytkownika i aktualizują stan, dzięki czemu proces pozostaje modułowy. Istnieje ich dziewięć:
greeting_nodeotwiera czat i pyta, czy użytkownik chce umówić się na wizytę.select_speciality_nodeprzedstawia dostępne specjalizacje.select_doctor_nodepokazuje lekarzy w wybranej specjalizacji.select_date_nodepyta, który dzień pasuje użytkownikowi.select_slot_nodeproponuje wolne terminy w tym dniu.confirm_nodeprosi użytkownika o zatwierdzenie wyboru.collect_details_nodeżąda imienia i numeru telefonu.completed_nodezamyka proces po udanej rezerwacji.
cancelled_node służy do obsługi użytkownika, który rezygnuje.Budowanie grafu
Funkcja build_booking_graph dodaje te węzły, określa warunkowe przejścia między nimi i w ten sposób ustala ogólny tok rozmowy.
# 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()
Krótki skrypt renderuje wykres jako obrazek:
#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()
Zrób to:
cd clinic-agent
python save_langgraph_flow.py
Zapisuje diagram do folderu agents, co jest wygodnym sposobem na szybkie zobaczenie etapów i ich powiązań, zanim przekażemy je rzeczywistym użytkownikom.
Przetwarzanie wiadomości
process_message odbiera dane z interfejsu, przetwarza je za pomocą grafu i zwraca zaktualizowany stan wraz z odpowiedzią asystenta. Przyjmuje identyfikator wątku oraz nazwę sesji domyślną, dzięki czemu każda sesja przeglądarki lub uruchomienie notatnika ma swój własny zapisany postęp w procesie rezerwacji.
# 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
Testowanie agenta
Skrypt testowy steruje agentem z terminala:
# 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"])
Zrób to:
cd clinic-agent
python test/test_agent.py
Należy, aby wyświetlono dostępne opcje, co pokazuje, że grafik reaguje. W tym momencie cała interakcja działa z terminala, brakuje już tylko prezentacji.
Część 4: interfejs Streamlit
Interfejs użytkownika oferuje okno czatu w czasie rzeczywistym, przyciski opcji, które można kliknąć, oraz sesje identyfikowane unikalnymi ID. Przyciski są ważne w takim procesie, ponieważ wybór specjalizacji lub godziny jednym kliknięciem jest szybszy i mniej podatny na błędy niż wpisywanie ich ręcznie.
# 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()
Część 5: punkt wejścia
app.py jest punktem wyjścia, który uruchamia całe aplikacje:
# app.py - Main entry point for the application
from ui.chat_ui import run_chat_ui
if __name__ == "__main__":
run_chat_ui()
Dwa sposoby uruchomienia
Jako aplikacja webowa Streamlit, co jest najprostszą metodą:
streamlit run app.py
Otwiera się pod adresem http://localhost:8501.
W notatniku Jupyter, który nadaje się do rozwoju i eksperymentów. Najpierw należy sprowadzić funkcje:
# 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
)
Stwórz graf i wyświetl jego strukturę:
# 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))
Następnie uruchom interaktywną sesję czatu wewnątrz notatnika:
# 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)
Pomocnik w notatniku uruchamia sesję rezerwacji na określonym wątku i można go sformatować, aby rozpocząć od nowa. W notatniku pole wprowadzania danych znajduje się u góry, nad komórkami. Repozytorium zawiera notatnik clinic-agent.ipynb, który uruchamia cały proces.
Co daje Ci gotowy bot
Rezultatem jest chatbot do rezerwacji stworzony na bazie LangGraph, OpenAI i SQLite z interfejsem użytkownika w Streamlit. Proces oparty na stanach prowadzi użytkowników od wyboru specjalizacji do potwierdzenia terminu, a poszczególne warstwy (baza danych, usługi, logika agenta i interfejs) pozostają oddzielone, dzięki czemu aplikacja jest modułowa i łatwa do rozbudowy. Ponieważ warstwy są wyraźnie oddzielone, każda poprawka dotyczy tylko jednej z nich. Naturalnymi dalszymi krokami są integracja z kalendarzem, powiadomienia, obsługa wielojęzyczna oraz platformy do komunikacji. Jest to praktyczny przykład połączenia nowoczesnych frameworki AI z prostymi narzędziami internetowymi w użyteczną aplikację medyczną.
Literatura pokrewna
- Przepływy pracy LangGraph Multi-Agent: porównanie roju a nadzorcy — Podziel przepływ pracy LangGraph na podagenty z funkcją checkpointing, a następnie stwórz asystenta finansowego działającego zarówno jako rój, jak i nadzorca, aby zobaczyć, w którym momencie każdy z tych modeli przestaje być skuteczny.
- Stwórz ShopBot: agent obsługi klienta w LangGraph z pamięcią, narzędziami i możliwością zatwierdzania zwrotów — Buduj agenta obsługi klientów w module LangGraph krok po kroku: stan, narzędzia, węzły, routowanie, aktualne podsumowania oraz mechanizm przerwania czekający na zatwierdzenie zwrotu.