Chatbot für Klinikterminbuchungen mit LangGraph, SQLite und Streamlit – Schritt für Schritt
Erstellen Sie einen Terminbuchungsassistenten für eine Klinik: SQLite-Tabellen, Servicefunktionen, ein neun Knoten umfassendes LangGraph-Arbeitsflussmodell mit gpt-4o-mini, eine Streamlit-Chat-Oberfläche sowie einen Notebook-Modus.
Quellen: Agentic AI Project: Build a Customer Service Chatbot for a Clinic
Die Leitung einer Klinik erfordert viel mühsame Koordination: Die Zuordnung von Patienten zu Ärzten, das Überwachen, wer zu welcher Zeit frei ist, sowie das Vermeiden doppelter Terminvereinbarungen. Ein konversationaler Assistent kann diesen Teil der Terminverwaltung übernehmen. Dieses Projekt entwickelt einen solchen Assistenten aus vier Komponenten: LangGraph zur Steuerung einer mehrschrittigen Konversation als Zustandsmaschine, OpenAI’s gpt-4o-mini zur Verständnis der eingegebenen Texte, SQLite zum Speichern der Daten sowie Streamlit für das Chat-Fenster.
Das Problem und der Plan
Viele Kliniken verlassen sich weiterhin auf manuelle Prozesse oder eine Mischung aus verschiedenen Tools. Die Folgen sind lange Wartezeiten, Terminkonflikte, uneinheitliche Arbeitsbelastung des Personals sowie eine frustrierende Erfahrung für die Patienten, die bereits Schwierigkeiten haben können, einen verfügbaren Arzt oder die richtige Fachrichtung zu finden. Was eine Klinik benötigt, ist ein automatisiertes System, das über natürliche Konversationen Termine vereinbart und dabei genaue Aufzeichnungen über Ärzte, Patienten und Terminvereinbarungen führt.
Der Chatbot führt den Benutzer durch ein kurzes Skript:
- Er begrüßt den Benutzer und fragt, ob dieser einen Termin wünscht.
- Er listet die verfügbaren medizinischen Fachgebiete auf.
- Er zeigt den Arzt sowie die freien Zeitfenster an.
- Er sammelt den Namen und die Telefonnummer des Patienten ein.
- Er bestätigt die Buchung und speichert sie in der Datenbank.
Der Vorteil besteht darin, dass die Verwaltungsaufgaben reduziert werden, Buchungen schneller erfolgen, es zu keinen Terminkollisionen kommt und der Patientenprozess reibungsloser abläuft. Der fertige Code ist in Ordnern für Daten, Dienste, den Agenten und die Benutzeroberfläche gegliedert. Jeder Teil hat eine spezifische Aufgabe: LangGraph entscheidet, welcher Schritt des Gesprächs als Nächstes erfolgt, das Modell interpretiert freien Text, SQLite merkt sich, wer und wann etwas getan hat, und Streamlit zeigt dem Benutzer alles im Browser an. Durch die Trennung dieser Aufgaben kann man beispielsweise die Benutzeroberfläche ändern, ohne den Rest zu beeinflussen.
Einrichtung
Erstellen Sie eine virtuelle Umgebung. Der Befehl ist auf Mac, Linux und Windows derselbe:
python -m venv .venv
Aktivieren Sie es unter Mac oder Linux:
source .venv/bin/activate
Auf Windows mit PowerShell:
.venv\Scripts\Activate.ps1
In der Windows-Eingabeaufforderung:
.venv\Scripts\activate
Sobald es aktiv ist, beginnt der Terminal-Befehl mit dem Namen der Umgebung, wie folgt:
(.venv) your-folder-name %
Danach folgen die Abhängigkeiten. Die Datei requirements.txt listet Streamlit, python-dotenv, pydantic, pandas sowie die anderen Bibliotheken auf:
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
Installieren Sie sie aus dem Projektverzeichnis:
cd clinic-agent
pip install -r requirements.txt
Zum Schluss erstellen Sie eine .env-Datei im Verzeichnis clinic-agent, in der Ihre OpenAI-Schlüssel gespeichert sind:
OPENAI_API_KEY=your_openai_api_key_here
Teil 1: Die Datenbank (data/db.py)
Ein Buchungssystem basiert auf seiner Datenschicht. Die Klinik verwendet SQLite mit drei Tabellen für Ärzte, Kunden und Buchungen. Das Schema ist relationell und lässt sich während eines Gesprächs leicht abfragen: Eine Buchung verweist auf einen Arzt und einen Kunden, sodass eine einzige Abfrage zeigt, wer wen und wann behandelt. Telefonnummern dienen als praktische Identifikation für Patienten, damit ein wiederkehrender Anrufer nicht auf eine doppelte Datenspur stößt.
Ärzte. Jede Zeile stellt einen Arzt mit einer Spezialisierung und seinen Arbeitszeiten dar, wodurch der Chatbot nach Auswahl der Spezialisierung die richtige Person vorschlagen kann. Die Felder sind doctor_id (eindeutiger Identifikator), doctor_name, speciality (zum Beispiel Dermatologe oder Kinderarzt) sowie office_timing (die Arbeitszeiten des Arztes). Die Demo umfasst fünf Spezialisierungen: Dermatologe, Orthopäde, Allgemeinmediziner, Kinderarzt und HNO-Arzt, jeweils mit erfundenen Namen und Arbeitszeiten. Weitere hinzuzufügen ist einfach.
Kunden. Diese Tabelle enthält die Informationen, die für die Buchung benötigt werden. Jedes Mal, wenn jemand bucht, sucht das System den Kunden anhand der Telefonnummer und erstellt bei Nichtvorhandensein ein neues Datensatz. Die Felder sind customer_id, name und phone.
Buchungen. Die zentrale Tabelle. Jede Zeile verbindet einen Arzt und einen Kunden mit einem bestimmten Datum und Zeitpunkt. Ihre Felder sind booking_id, doctor_id und customer_id (Verweise auf die anderen beiden Tabellen), appointment_date, appointment_time und status, wie z. B. Confirmed oder Cancelled.
Die Skript definiert diese Tabellen und erstellt die Datei 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.")
Führen Sie es aus dem Projektverzeichnis aus:
cd clinic-agent
python data/db.py
Dadurch entsteht clinic.db mit den drei Tabellen. Das erneute Ausführen des Skripts ist eine schnelle Möglichkeit, die Beispieldaten beim Experimentieren zurückzusetzen.
Teil 2: die Service-Schicht
Services enthalten die Geschäftslogik zwischen dem Agenten und der Datenbank. Der LangGraph-Agent fragt nicht direkt Tabellen ab. Er ruft Servicefunktionen auf, um Arztinformationen abzurufen, Zeitfenster zu generieren und Buchungen zu bestätigen. Bei der Gestaltung eines Agenten dienen diese Funktionen als Werkzeuge, wodurch der Agent sich auf den Ablauf des Gesprächs konzentrieren kann. Diese Trennung macht die Logik außerdem unabhängig testbar. Man kann die Generierung von Zeitfenstern und Buchungen überprüfen, ohne jemals das Sprachmodell aufrufen zu müssen – genau das tut auch das nächste kleine Skript.
Dienst für Ärzte (services/doctor_service.py) kümmert sich um Fachgebiete, Ärzte und Zeitfenster. Er gibt die Liste der Fachgebiete aus der Datenbank zurück, holt Ärzte für ein gewähltes Fachgebiet ab, erstellt Zeitfenster aus den Sprechzeiten eines Arztes und wandelt Anzeigezeiten wie 14:00 Uhr in das 24-Stunden-Format um, um sie zu speichern.
# 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}"
Dienstleistungs-Reservierungssystem (services/booking_service.py) kümmert sich um Kunden und Reservierungen. Es findet einen bestehenden Kunden anhand der Telefonnummer oder erstellt einen neuen, prüft die bereits belegten Termine eines Arztes an einem bestimmten Datum, filtert diese heraus, um die freien Termine zu erhalten, generiert eine eindeutige Reservierungs-ID und speichert schließlich die Details des Termins – Arzt, Patient, Datum und Uhrzeit.
# 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
Überprüfung der Dienste
Ein kleiner Script testet beide Dienste:
# 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}")
Führen Sie es aus:
cd clinic-agent
python test_service.py
Die Konsole gibt unter anderem die Liste der Spezialisierungen sowie die Termine für einen Arzt aus:
(.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
Falls das richtig aussieht, die Dienste funktionieren und der interessante Teil beginnen kann. Die Anzeige echter Spezialisierungen sowie tatsächlicher Termine bestätigt, dass Datenbank, Arztlogik und Buchungslogik miteinander übereinstimmen, bevor weitere Konversationslogiken hinzugefügt werden.
Teil 3: der Agent
Das Herzstück des Chatbots ist ein LangGraph-Agent, der eine Buchung durch mehrere Phasen führt.
Buchungsstatus
Das Statusobjekt dient als strukturiertes Gedächtnis für die Konversation. Es protokolliert den aktuellen Schritt, wie zum Beispiel Begrüßung oder Auswahl eines Arztes, merkt sich Auswahlmöglichkeiten wie Spezialisierung, Arzt und Termin sowie Kundendaten und Buchungsinformationen, damit der Agent den Termin abschließen kann.
# 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": []
}
Der LLM-Hilfsprogramm
Eine einzige call_llm-Funktion zentralisiert die Kommunikation mit dem OpenAI-Modell. Sie sendet System- und Benutzereingaben, gibt den generierten Text zurück und sorgt dafür, dass jede Modellaufrufkette konsistent und wiederverwendbar bleibt.
# 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 ""
Knoten
Jeder Knoten stellt einen Schritt im Buchungsworkflow dar. Die Knoten kümmern sich um die von ihnen dargestellte Phase, lesen die Eingaben des Benutzers ein und aktualisieren den Status, wodurch der Workflow modular bleibt. Es gibt insgesamt neun davon:
greeting_nodeöffnet das Chat-Fenster und fragt, ob der Benutzer einen Termin wünscht.select_speciality_nodelistet die angebotenen Spezialisierungen auf.select_doctor_nodezeigt die Ärzte der ausgewählten Spezialisierung an.select_date_nodefragt, welcher Tag dem Benutzer passt.select_slot_nodebietet die freien Zeiten an diesem Tag an.confirm_nodebittet den Benutzer, die Auswahl zu bestätigen.collect_details_nodefragt nach Namen und Telefonnummer.completed_nodebeendet eine erfolgreiche Buchung.
cancelled_node bezieht sich auf einen Benutzer, der zurücktritt.Aufbau des Graphen
Die Funktion build_booking_graph fügt diese Knoten hinzu, definiert die bedingten Übergänge zwischen ihnen und legt somit den Gesamtverlauf der Konversation fest.
# 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()
Ein kurzer Skript rendernt das Diagramm als Bild:
#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()
Führen Sie es aus:
cd clinic-agent
python save_langgraph_flow.py
Es speichert das Diagramm im Ordner „Agents“, was eine praktische Möglichkeit bietet, die Phasen sowie ihre Verbindungen auf einen Blick zu sehen, bevor man sie echten Benutzern anvertraut.
Nachrichten verarbeiten
process_message nimmt Eingaben von der Schnittstelle entgegen, leitet sie durch das Graphen-Modell und gibt den aktualisierten Zustand zusammen mit der Antwort des Assistenten zurück. Es akzeptiert eine Thread-ID sowie einen standardmäßigen Sitzungsnamen, sodass jede Browser-Sitzung oder jedes Notebook-Exekutivprotokoll seinen eigenen gespeicherten Fortschritt im Buchungsprozess erhält.
# 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
Testen des Agents
Ein Testskript steuert den Agenten über die Terminalkonsole:
# 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"])
Führen Sie es aus:
cd clinic-agent
python test/test_agent.py
Es sollte die verfügbaren Optionen ausgeben, was zeigt, dass das Diagramm reagiert. Zu diesem Zeitpunkt funktioniert die gesamte Interaktion über die Terminal, es fehlt nur noch die Präsentation.
Teil 4: Die Streamlit-Schnittstelle
Die Benutzeroberfläche bietet ein Echtzeit-Chat-Fenster, klickbare Optionen sowie Sessions, die durch eindeutige IDs identifiziert werden. In einem solchen Ablauf sind Buttons wichtig, denn das Auswählen einer Spezialisierung oder eines Zeitfensters mit einem Klick ist schneller und weniger fehleranfällig als das Eintippen davon.
# 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()
Teil 5: Der Eingangspunkt
app.py ist der Ausgangspunkt, von dem aus die gesamte Anwendung gestartet wird:
# app.py - Main entry point for the application
from ui.chat_ui import run_chat_ui
if __name__ == "__main__":
run_chat_ui()
Zwei Möglichkeiten, es auszuführen
Als Streamlit-Webanwendung, was der einfachste Weg ist:
streamlit run app.py
Es öffnet sich unter http://localhost:8501.
In einem Jupyter Notebook, das sich für Entwicklung und Experimente eignet. Importieren Sie zuerst die Funktionen:
# 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
)
Bauen Sie das Diagramm und zeigen Sie seine Struktur an:
# 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))
Dann führen Sie eine interaktive Chat-Sitzung innerhalb des Notebooks durch:
# 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)
Der Notebook-Hilfsprogramm führt eine Buchungssitzung in einem benannten Thread aus und kann zurückgesetzt werden, um von vorn zu beginnen. Im Notebook erscheint das Eingabefeld oben, über den Zellen. Das Repository enthält ein clinic-agent.ipynb-Notebook, das den gesamten Ablauf ausführt.
Was der fertige Bot Ihnen bietet
Das Ergebnis ist ein Buchung-Chatbot, der aus LangGraph, OpenAI und SQLite sowie einem Streamlit-Frontend erstellt wurde. Ein durch den Zustand gesteuerter Workflow führt die Nutzer von der Auswahl einer Spezialisierung bis zur Bestätigung eines Zeitfensters, wobei die Schichten (Datenbank, Dienste, Agentenlogik und Benutzeroberfläche) getrennt bleiben, sodass die Anwendung modular und leicht erweiterbar ist. Da die Schichten sauber voneinander getrennt sind, betrifft jede Verbesserung nur eine von ihnen. Zu den natürlichen nächsten Schritten gehören die Integration in Kalender, Benachrichtigungen, mehrsprachige Unterstützung sowie Messaging-Plattformen. Es handelt sich um ein praktisches Beispiel dafür, wie moderne KI-Frameworks und einfache Web-Tools zu einer nützlichen Gesundheitsanwendung kombiniert werden können.