Overview
Provides utility functions for creating simulated users and chat simulation graphs to evaluate chatbot assistants in automated testing workflows.
Description
The Simulation Utils module supplies two primary factory functions, `create_simulated_user` and `create_chat_simulator`, that together enable fully automated chatbot evaluation through simulated multi-turn conversations. The simulated user is backed by a language model (defaulting to `gpt-3.5-turbo`) driven by a configurable system prompt that defines the persona and behavior of the simulated human participant.
The `create_chat_simulator` function constructs a LangGraph `StateGraph` with two nodes: an "assistant" node that wraps the chatbot under test and a "user" node that wraps the simulated user. Messages alternate between the two participants with automatic role-swapping so the simulated user always sees the conversation from the human perspective. The graph supports a configurable maximum number of turns and an optional custom continuation predicate; by default the simulation ends after the turn limit or when the simulated user emits "FINISHED".
The module also defines a `SimulationState` TypedDict that tracks the accumulated messages and optional input parameters, and a `_prepare_example` helper that converts dataset examples into the state format expected by the graph. This design allows seamless integration with LangSmith datasets for batch evaluation of chatbot quality.
Usage
Use these utilities when you need to automate chatbot evaluation against a dataset of example inputs. They are particularly useful in CI pipelines or LangSmith evaluation runs where a simulated user replaces a real human to generate multi-turn conversations that can then be scored by an LLM judge or heuristic evaluator.
Code Reference
Source Location
Signature
def create_simulated_user(
system_prompt: str, llm: Runnable | None = None
) -> Runnable[Dict, AIMessage]:
...
def create_chat_simulator(
assistant: (
Callable[[List[AnyMessage]], str | AIMessage]
| Runnable[List[AnyMessage], str | AIMessage]
),
simulated_user: Runnable[Dict, AIMessage],
*,
input_key: str,
max_turns: int = 6,
should_continue: Optional[Callable[[SimulationState], str]] = None,
):
...
Import
from simulation_utils import create_simulated_user, create_chat_simulator
I/O Contract
create_simulated_user
| Parameter |
Type |
Required |
Description
|
| system_prompt |
`str` |
Yes |
System prompt defining the simulated user persona
|
| llm |
None` |
No |
Language model for the simulated user; defaults to `ChatOpenAI(model="gpt-3.5-turbo")`
|
| Return Type |
Description
|
| `Runnable[Dict, AIMessage]` |
A runnable chain that takes a dict with a "messages" key and returns an AI message
|
create_chat_simulator
| Parameter |
Type |
Required |
Description
|
| assistant |
Runnable` |
Yes |
The chatbot function or runnable to evaluate
|
| simulated_user |
`Runnable[Dict, AIMessage]` |
Yes |
The simulated user runnable (from `create_simulated_user`)
|
| input_key |
`str` |
Yes |
Key in the dataset example dict that contains the initial user message
|
| max_turns |
`int` |
No |
Maximum conversation turns before stopping (default: 6)
|
| should_continue |
`Callable[[SimulationState], str] | None` |
No |
Custom function returning the next node name or END
|
| Return Type |
Description
|
| `Runnable` |
A compiled graph that accepts a dataset example dict and returns a `SimulationState`
|
SimulationState
| Field |
Type |
Description
|
| messages |
`List[AnyMessage]` |
Accumulated conversation messages (uses `add_messages` reducer)
|
| inputs |
`Optional[dict[str, Any]]` |
Additional inputs from the dataset example (excluding the input_key)
|
Usage Examples
from simulation_utils import create_simulated_user, create_chat_simulator
# Create a simulated user with a custom persona
simulated_user = create_simulated_user(
system_prompt="You are a customer asking about return policies. "
"Be polite but persistent. Say FINISHED when satisfied."
)
# Define a simple assistant function
def my_assistant(messages):
# Your chatbot logic here
return "Our return policy allows returns within 30 days."
# Build the simulator
simulator = create_chat_simulator(
assistant=my_assistant,
simulated_user=simulated_user,
input_key="question",
max_turns=8,
)
# Run a simulation with a dataset example
result = simulator.invoke({"question": "What is your return policy?"})
for msg in result["messages"]:
print(f"{msg.__class__.__name__}: {msg.content}")
Related Pages