Практичні нотатки: Проєкт з штучного інтелекту типу агент: створіть чат-бота для обслуговування клієнтів для
Покрокова інструкція з практичних нотаток: Проєкт агентського ШІ: створення чат-бота для обслуговування клієнтів для: контрактів, перевірок та слотів для коду для команд, які використовують цю схему.
Наведені нижче примітки описують практичний підхід до реалізації проекту «Agentic AI Project: Створення чат-бота для обслуговування клієнтів у клініці». Основна увага приділяється контрактам, перевіркам та місцям для вставки коду, а не мотиваційним аспектам. Під час роботи на етапі огляду спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та наслідки часткової невдачі. Такий перелік допоможе зберегти чесність пізніших змін у коді. Документуйте як успішний, так і аварійний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої доробки.
Вступ
Етап введення працює найкраще, якщо його розглядати як вимірювану поверхню. Зафіксуйте один ідеальний приклад роботи, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед складними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій. Зберігайте стан графа у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, і ускладнюють продовження роботи після перерв.
Опис проблеми
Етап формулювання проблеми найкраще функціонує, якщо його розглядати як вимірювану поверхню. Зафіксуйте один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Зберігайте стан графа у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, що ускладнює продовження роботи після перерв.
Рішення
Етап «Рішення» працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-середовища до спільних. Зберігайте стан графа у простому та типованому вигляді. Вкладені блоки приховують інформацію про те, який вузол заповнив яке поле, і ускладнюють продовження роботи після перерв. Етап «Рішення» працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Документуйте як успішний, так і відновлювальний сценарії роботи разом. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Налаштування
На етапі налаштування необхідно визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Краще використовувати невеликі, перевірювані одиниці коду замість об’ємних скриптів. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану структуру процесу. Необхідно встановити людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення елементів під час компіляції не є гарантією повноти бізнес-функціоналу.
Mac / Linux / Windows
На етапі Mac, Linux та Windows необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Призначте назви артефактів, визначте критерії успіху та не допускайте мовчазного часткового завершення. Запровадьте людське схвалення для операцій, які вимагають фінансових витрат чи змінюють дані у продакшені. Компіляційна налаштування не є гарантією повності бізнес-процесу.
python -m venv .venv
source .venv/bin/activate
.venv\Scripts\Activate.ps1
.venv\Scripts\activate
(.venv) your-folder-name %
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
OPENAI_API_KEY=your_openai_api_key_here
Частина 1: Налаштування бази даних (data/db.py)
На етапі налаштування бази даних частини 1 необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Встановлюйте людське схвалення для кроків, які витрачають гроші чи змінюють дані в продакшені. Підключення під час компіляції не є гарантією повноти функціоналу продукту. На етапі налаштування бази даних частини 1 необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Документуйте як оптимальний, так і резервний сценарії виконання. Повторні спроби, людське схвалення та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
1. Таблиця лікарів
Під час роботи над етапом 1 «Таблиця лікарів» спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям коду перед об’ємними скриптами. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність операцій. Робіть перевірки після дорогих кроків. Система повторного запуску не повинна знову оплачувати один і той самий виклик LLM, коли оператор намагається виконати наступний етап.
2. Таблиця клієнтів
Під час роботи над етапом „Таблиця клієнтів 2“ спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Призначте назви елементам, визначте критерії успіху та не допускайте безпроблемного часткового виконання завдань. Робіть перевірки після дорогих кроків. Система повернення до виконання не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор намагається виконати наступний етап.
3. Таблиця бронювань
Під час роботи над третім етапом «Таблиця бронювань» спочатку запишіть умови використання: необхідні дані вхіду, сигнал про успіх та наслідки часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із результатами функціоналу. Відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Робіть контрольні пункти після дорогих кроків. Система повторного запуску не повинна знову стягувати плату за один і той самий виклик ШІ, коли оператор перепробовує пізніший етап. Під час роботи над третім етапом «Таблиця бронювань» спочатку запишіть умови використання: необхідні дані вхіду, сигнал про успіх та наслідки часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Одночасно задокументуйте оптимальний та резервний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
4. Створення бази даних
Етап 4 «Створення бази даних» найкраще функціонує, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний зразок запису, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій. Зберігайте стан графа у простій та типованій формі. Вкладені блоки приховують інформацію про те, який вузол записав яке поле, і ускладнюють продовження роботи після перерв.
# 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
Частина 2: Шар послуг (інструменти для агента)
Етап шару обслуговування Частина 2 працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування перед розширенням обсягу робіт. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового виконання завдань. Запропонуйте інструменти з вузькими схемами та чіткими позначеннями побічних ефектів. У користувачів має бути можливість дізнатися, які виклики змінюють стан системи, перш ніж вони автоматично схвалять їх.
Служба лікаря (services/doctor_service.py)
Сервіс Doctor Service на етапі doctorservice працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу функціоналу. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-середовища до спільних. Зберігайте стан графа у випрямленому та типованому форматі. Вкладені блоки приховують інформацію про те, який вузол заповнив яке поле, та ускладнюють відновлення після перерв. Сервіс Doctor Service на етапі doctorservice працює найкраще, коли його розглядають як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу функціоналу. Документуйте як успішний, так і відновлювальний шляхи роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
# 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)
Для етапу bookingservice у сервісі бронювань необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Краще використовувати невеликі, тестовані одиниці коду замість об’ємних скриптів. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану структуру процесу. Встановлюйте людське схвалення для тих кроків, які пов’язані з витратами грошей чи зміною даних у продакшені. Підключення на етапі компіляції не є гарантією повноти бізнес-функціоналу.
# 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: Шар агентів
На етапі 3-ї частини – шару агентів необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Встановлюйте людське схвалення для тих кроків, які спричиняють витрати чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повності функціоналу продукту. На етапі 3-ї частини – шару агентів необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Документуйте як оптимальний, так і альтернативний шлях виконання. Повторні спроби, людське схвалення та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Статус бронювання:
Під час роботи над етапом статусу бронювання спочатку запишіть умови контракту: необхідні дані, сигнал про успіх та наслідки часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям коду перед об’ємними скриптами. Якщо якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на складну послідовність операцій. Робіть перевірки після дорогих кроків. Функція відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор намагається знову виконати пізніший етап.
# 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
Під час роботи над етапом допоміжної функції LLM спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Дайте назви створюваним елементам, визначте критерії успіху та не допускайте безповідомного часткового виконання. Зберігайте у кеші стабільні інструкції системи та схеми інструментів. Повторна передача ідентичних даних є поширеною причиною витрат ресурсів.
# 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 ""
Вузли агента
Під час роботи над етапом Agent Nodes спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із результатами функціонування. Відображення витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-версії до спільних середовищ. Робіть контрольні пункти після дорогих кроків. Функція відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор перезапускає пізніший етап. Під час роботи над етапом Agent Nodes спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Одночасно задокументуйте оптимальний та резервний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Створення графу
Етап створення графа працює найкраще, якщо його розглядати як вимірювану поверхню. Зафіксуйте один ідеальний приклад роботи, один випадок збою та примітку щодо скасування змін перед розширенням обсягу роботи. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність дій. Зберігайте стан графа у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, і ускладнюють продовження роботи після перерв.
# 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/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: UI Streamlit
Для етапу інтерфейсу Streamlit у частині 4 необхідно визначити вхідні дані, виконавця кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись здогадатися про прихований стан. Краще використовувати невеликі, перевірювані одиниці коду замість об’ємних скриптів. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану структуру процесу. Встановлюйте людське схвалення для тих етапів, де витрачаються гроші чи змінюються дані в продакшені. Підключення елементів під час компіляції не є гарантією повноти бізнес-функціоналу.
# 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: Точка входу до додатку
На етапі введення даних у частині 5 необхідно визначити вхідні дані, відповідальну особу за кожен крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Призначте назви елементам, визначте критерії успіху та не допускайте мовчазного часткового завершення. Необхідно забезпечити людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Компіляційне підключення елементів не є гарантією повноти виконання завдань у бізнес-сенсі.
# app.py - Main entry point for the application
from ui.chat_ui import run_chat_ui
if __name__ == "__main__":
run_chat_ui()
Запуск чат-бота: два підходи
Для двоетапної реалізації Chatbot слід спочатку визначити вхідні дані, виконавця кроку та критерії завершення, перш ніж змінювати код. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-середовища до спільних середовищ. Встановіть людське схвалення для тих кроків, які вимагають фінансових витрат або змінюють дані у продакшені. Підключення під час компіляції не гарантує повності бізнес-функціоналу.
Підхід 1: Веб-додаток Streamlit
Для етапу Streamlit Web за підходом 1 необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись здогадатися про прихований стан. Конфігурацію слід тримати окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функціоналу мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь алгоритм. Необхідно передбачити людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повноти бізнес-процесу.
streamlit run app.py
Підхід 2: Jupyter Notebook
Для етапу Jupyter Notebook у підході 2 необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Необхідно документувати як шлях успішного виконання, так і шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Встановлюйте людське схвалення для тих кроків, які призводять до витрат грошей або змінюють дані у продакшені. Підключення під час компіляції не є гарантією повності функціоналу продукту.
# 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)
Висновок:
На етапі завершення необхідно визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового завершення. Необхідно отримати схвалення людини для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повності виконання завдань у бізнес-сенсі.
Посилання:
На етапі референсів необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Відображення витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Встановлюйте людське схвалення для тих кроків, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повності функціоналу продукту. На етапі референсів необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Документуйте як оптимальний, так і альтернативний шлях виконання. Повторні спроби, людське схвалення та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Чек-лист операцій
На етапі чек-листу операцій необхідно визначити вхідні дані, відповідальну особу за кожен крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан.
Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, яке оператори можуть перевірити, не читаючи весь код.
Встановіть людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повноти функціоналу.
Напишіть короткий посібник: як обмінювати ключі, як спорожнювати чергу, як скасовувати останнє завантаження даних.
Документуйте як успішний, так і відновлювальний сценарії роботи. Повторні спроби, людське схвалення та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Встановіть людське схвалення для тих етапів, де витрачаються гроші або змінюються дані виробництва. Підключення під час компіляції не є гарантією повноти бізнес-функціоналу.
Перш ніж переходити на нову версію стеку, заморозьте існуючі версії, створіть остаточний запис для критичного шляху виконання та підтвердьте кроки для скасування змін. У спільних середовищах необхідні обмеження на частоту використання, перевірки прав доступу та чіткий власник для зміни секретних ключів. Віддавайте перевагу надійності перед креативними одноразовими демонстраціями.
Примітка до версії 9744ef4a5b25: не включайте ключі постачальника до репозиторію, встановіть ліміт на токени на сеанс та зберігайте записи поруч із фіксами для оцінки, щоб подальша заміна моделей залишалася порівнянною.