使用 LangGraph、SQLite 和 Streamlit 构建诊所预约聊天机器人，分步指南
为诊所开发一个预约系统助手：包含SQLite数据库表、服务函数、基于gpt-4o-mini的九节点LangGraph工作流、Streamlit聊天界面以及笔记本模式。
来源: Agentic AI Project: Build a Customer Service Chatbot for a Clinic
运营诊所需要大量繁琐的协调工作：为患者匹配医生、追踪医生的空闲时间，以及避免重复预约。对话式助手可以承担其中的预约功能。这个项目由四个部分组成：用于以状态机形式驱动多步骤对话的LangGraph、用于理解用户输入内容的OpenAI gpt-4o-mini、用于存储数据的SQLite，以及用于实现聊天界面的Streamlit。
问题与解决方案
许多诊所仍然依赖手工操作或零散的工具组合。其结果是患者需要长时间等待、出现预约冲突、员工工作负荷不均，同时患者也会因难以找到合适的医生或专业而感到沮丧。诊所需要的是一种能够通过自然对话完成预约，并同时准确记录医生、患者及预约信息的自动化系统。
聊天机器人会引导用户完成一段简短的流程：
- 它首先向用户问好，并询问是否需要预约。
- 接着列出可选择的医疗科目。
- 显示医生信息及可用的时间段。
- 收集患者的姓名和电话号码。
- 确认预约信息并将其保存在数据库中。
这样做的好处是减少了行政工作量，加快了预约速度，避免了时间冲突，同时也让患者的体验更加顺畅。完整的代码被划分到数据、服务、智能体以及用户界面这几个文件夹中，每个部分都有明确的职责。LangGraph负责决定对话的下一步走向，模型负责解析自由文本，SQLite用于记录患者信息及预约时间，而Streamlit则负责在浏览器中向用户展示所有内容。将这些功能分开，就能在修改某一部分（比如界面）时不会影响到其他部分。
设置
创建一个虚拟环境。在 Mac、Linux 和 Windows 上命令相同：
python -m venv .venv
在 Mac 或 Linux 上激活它：
source .venv/bin/activate
在安装了 PowerShell 的 Windows 系统上：
.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
最后，在clinic-agent目录中创建一个.env文件，将您的OpenAI密钥存储在其中：
OPENAI_API_KEY=your_openai_api_key_here
第一部分：数据库（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，如“已确认”或“已取消”等。
该脚本会定义这些表格并创建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 文件。在实验过程中，重新运行该脚本是快速重置示例数据的一种方法。
第二部分：服务层
服务层负责处理代理与数据库之间的业务逻辑。LangGraph 代理不会直接查询表格，而是通过调用服务函数来获取医生信息、生成时间槽并确认预订。在代理设计中，这些函数充当工具角色，使代理能够专注于对话流程。这种分离结构还使得相关逻辑可以独立测试——你无需调用语言模型即可检查时间槽生成和预订功能，而这正是下一个小型脚本所实现的。
医生服务模块（services/doctor_service.py）负责处理专业领域、医生信息以及时间安排。它从数据库中获取专业列表，为选定的专业查找对应的医生，根据医生的办公时间生成可用时间段，并将诸如“下午2:00”这样的显示时间转换为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
如果这些内容看起来正确，服务也能正常运行，那么有趣的部分就可以开始了。在添加任何对话逻辑之前，通过查看实际的专业领域和具体时间槽，可以确认数据库、医生匹配逻辑以及预约逻辑之间是一致的。
第三部分：智能体
聊天机器人的核心是一个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 笔记本中，这种环境更适合开发和实验。首先需要导入相关函数：
# 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作为前端界面。该系统通过状态驱动的工作流引导用户从选择服务项目到确认预约时间，各层（数据库、服务模块、智能体逻辑及用户界面）相互独立，因此具备模块化结构且易于扩展。由于各层划分清晰，任何改进都只需针对其中某一层进行。后续可自然扩展的功能包括日历集成、通知系统、多语言支持以及消息平台功能。这是个将现代人工智能框架与简单网页工具结合为实用医疗应用的实际案例。