a cleaning company to request a quote for cleaning a couch. They asked about its size and material and requested a picture of it. They also needed my address because travel time varies significantly across the city and therefore affects the price.
After receiving the quote for the couch, I asked about apartment cleaning. This started another exchange of messages about the apartment’s size, the desired cleaning depth, and optional services. Once we agreed on a price, we chatted a little more to find a suitable appointment.
The entire booking process took around 15 minutes. This sounds like a problem worth handing to an AI agent.
In this article, we’ll build this agent using Python, LangGraph, and LangChain. We’ll create a demo for testing the complete booking flow and integrate Langfuse for observability.
The AI agent will not only speed up the process but also work 24/7 when deployed into production.
Please note that this is not a simple chatbot. It is a stateful AI agent capable of managing a multi-step business process by combining conversational intelligence with deterministic business rules.
The source code of the agent is available on GitHub at customer-service-agent. Feel free to clone and test it yourself.
What does this agent do?
The agent does everything the customer-service representative of this business does:
- Responds to customer queries and understands their needs.
- Calculates the price for the service and informs the customer.
- Handles the customer’s acceptance or rejection.
- Proposes optimized time slots.
- Confirms and records the appointment.
Why an AI agent for this task?
We could build a quotation tool that asks customers a predefined set of questions. Once they approve the price, the system could automatically schedule an appointment.
This would work, but every customer would have to follow the same process.
An AI agent provides a more natural and flexible experience. For example, a customer can write:
“Hey! I need a cleaning service for my 2 bedroom apartment in this address. I also want you to clean inside of my refrigerator. I’m available on Tuesday and Wednesday.”.
This query includes all the required details so there is no need to ask any other question or make the customer enter the same information again through a form.
The AI agent can understand If all required information is already present and proceed directly to calculating and presenting a quote.
The AI agent can also handle the orchestration of any subsequent steps such as price calculation, handling customer’s decision, searching for optimized appointment times, and booking confirmation.
It also has the potential to support additional operations, such as notifying cleaners about new appointments or contacting nearby cleaners to find someone available for an urgent request.
Structure of the agent
I created the drawing below to demonstrate the structure of our AI agent.
The starting point is the interaction between a customer and AI agent. Chatting with the customer, AI agent tries to get all the information needed for the service.
The output of the first chat is booking or service details and the agent converts this to a structured output so that it can be used later on by the price and booking engines.
The BookingDetails I used in the first version is as follows:
class BookingDetails(BaseModel):
"""Information extracted from the conversation.
``size_info`` is square footage for a house and seat count for a couch.
Fields are optional because this model also represents partial extraction.
"""
service_type: ServiceType | None = Field(default=None)
size_info: float | None = Field(default=None, gt=0)
cleaning_depth: CleaningDepth | None = Field(default=None)
add_ons: list[str] = Field(default_factory=list)
address: str | None = Field(default=None)
is_complete: bool = False
next_question: str | None = Field(
default=None,
exclude=True,
description="A concise question asking only for information still missing.",
)
The agent only asks a new question if there is some missing details. Thus, the agent decides what happens next, which is a key point that separates an AI agent from a workflow.
LangGraph
We use LangGraph because the booking workflow needs shared state and must resume across multiple customer messages. Thus, we need a stateful agent.
The state is a TypedDict as shown below:
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
booking_details: BookingDetails
calculated_price: NotRequired[float | None]
time_options: NotRequired[list[TimeOption]]
selected_slot: NotRequired[TimeOption | None]
status: BookingStatus
booking_id: NotRequired[str | None]
LangGraph manages state during graph execution. Nodes read the current state and return partial updates. Conditional routing functions inspect that state to decide which node runs next.
We can create the graph builder using this state schema:
rom langgraph.graph import StateGraph
graph = StateGraph(AgentState)
Then, we start adding nodes and edges. For example, the following code snippet shows how we can create the price calculation node.
from customer_service_agent.engines import calculate_price
def calculate_price_node(state: AgentState) -> dict[str, Any]:
return {
"calculated_price": calculate_price(
state["booking_details"]
)
}
graph.add_node("calculate_price", calculate_price_node)
The price calculation node reads the extracted booking details and returns a partial state update.
The calculate_price is a function that calculates the price using the BookingDetails available in the AgentState .
After the information-gathering node runs, a conditional routing function determines whether enough information is available to calculate a price:
def should_continue_to_pricing(
state: AgentState,
) -> Literal["calculate_price", "end"]:
if state["booking_details"].is_complete:
return "calculate_price"
return "end"
We connect that routing decision to the graph:
graph.add_conditional_edges(
"gather_info",
should_continue_to_pricing,
{
"calculate_price": "calculate_price",
"end": END,
},
)
If the booking details are complete, execution continues to the pricing node. Otherwise, the graph ends the current turn and waits for another customer message.
The remaining nodes and conditional routes are added in the same way. Finally, the graph is compiled with a checkpointer.
graph.compile(checkpointer=checkpointer or MemorySaver())
Observability
The most critical component of any system is observability. Whether it’s a production line at a factory, a traditional ML model to detect anomalies, or any AI-agent used in production.
If you do not know what goes on in your system, you just need to rely on pure luck to solve any issue. And it’s certain that your production system will have issues.
There are many alternatives to use for observability of an AI agent. We will be using LangFuse, an open-source AI engineering platform designed for tracing, monitoring, evaluating, and debugging large language model (LLM) applications.
The free tier is enough to test our project. You just need to create the secret and public keys in LangFuse dashboard and paste them in your .env file:
LANGFUSE_SECRET_KEY=
LANGFUSE_PUBLIC_KEY=
LANGFUSE_BASE_URL="https://cloud.langfuse.com"
We can create a langfuse handler as follows:
def create_langfuse_handler() -> Any | None:
"""Return a configured handler when credentials are available."""
if not os.getenv("LANGFUSE_PUBLIC_KEY") or not os.getenv("LANGFUSE_SECRET_KEY"):
return None
from langfuse.langchain import CallbackHandler
return CallbackHandler()
def flush_langfuse(handler: Any | None) -> None:
"""Upload queued telemetry without making observability application-critical."""
if handler is None:
return
try:
from langfuse import get_client
get_client().flush()
except Exception:
logger.exception("Failed to flush LangFuse telemetry")
LangFuse dashboard shows every detail for us to debug and monitor our AI agent. We can see the prompts, inputs, and outputs. We also see the token usage and cost so it’s very easy to detect if there is unexpected situation in our budget.
LangFuse dashboard looks like this:
It looks complicated at first but once you start using it, it becomes much easier.
What’s next
The project includes a simple CLI to test the end-to-end booking workflow, but its modular design gives us plenty of room for improvement.
You need an OpenAI API Key to test it but it costs almost nothing. The screenshot below shows the cost of a single run:
Every component has access to the central agent state so the system can easily be extended with:
- Frontend interfaces such as web, mobile apps, or messaging platforms like WhatsApp.
- A production database like PostgreSQL.
- Direct connections to real employee calendars and location-based services.
I consider this release as v0 and I’m planning to add more features and functionality on top of it.
Thanks for reading.
Instagram Engineering
Shopify Engineering