Chat flows by example¶
examples/chat_flows.py is a scripted city help desk (no model, no API key): one agent picks
a Flow per question, one per pattern. Run python examples/chat_flows.py, or replay it here.
Open the demo full screen. The Graph lights up node by node.
The agent¶
chat runs the picked flow in the turn's run, so tool calls show as cards; session.state
keeps the museum for later turns. Each flow ends in reply, where a stand-in model writes the
answer, so the Cost tab has tokens and dollars.
def pick_flow(text: str) -> Flow[Any]:
"""The flow for a question, by its words."""
words = text.lower()
for keys, flow in (
(("book",), booking_flow),
(("compare", "prices at"), compare_flow),
(("park", "playground"), park_flow),
(("plan",), day_flow),
):
if any(key in words for key in keys):
return flow
return ask_flow
async def chat(message: ChatMessage, session: ChatSession) -> AsyncIterator[str]:
"""Runs the question's flow in this turn's run, then streams the answer in chunks."""
flow = pick_flow(message.text)
start: dict[str, Any] = {"question": message.text}
if flow is booking_flow and session.state.get("place"):
start["place"] = session.state["place"] # the museum asked about earlier in this chat
runner = FlowRunner(session.executor, model=WRITER, trace_state=True)
result = await runner.run(flow, start, ctx=session.context())
place = getattr(result.state, "place", "")
if place:
session.state["place"] = place # remembered for later turns; editable in the page
session.state["questions"] = session.state.get("questions", 0) + 1
words = result.state.answer.split(" ")
for at in range(0, len(words), 3):
yield " ".join(words[at : at + 3]) + (" " if at + 3 < len(words) else "")
await asyncio.sleep(0.05)
async def write(request: LLMRequest) -> dict[str, Any]:
"""Stands in for a model: it returns the draft it is given, with made-up token counts."""
draft = request.messages[-1].content
return {"text": draft, "usage": {"input_tokens": 60 + len(draft) // 4, "output_tokens": 40}}
WRITER = CallableModel(write, provider="local", model="writer")
PRICES = Prices.empty().merged(
{"version": "demo", "models": {"writer": {"provider": "local", "input": 1.0, "output": 4.0}}}
)
Node = Callable[[Any, ToolContext], Awaitable[dict[str, Any]]]
def reply(draft: Callable[[Any], str]) -> Node:
"""The last node of every flow: the writer model turns the flow's draft into the answer."""
async def write_answer(state: Any, ctx: ToolContext) -> dict[str, Any]:
model = node_model(ctx)
if model is None:
raise ConfigError("the reply node needs FlowRunner(model=...)")
message = Message(role="user", content=draft(state))
return {"answer": (await model.generate(LLMRequest(messages=(message,)))).text}
return write_answer
1. Router: "Is the City Museum open on Monday?"¶
classify reads the question. A Route sends it to one branch: hours, prices or help.
flowchart TD
aa_start((start)) --> n_classify
n_classify["classify"]
n_hours(["hours<br/>tool: opening_hours"])
n_prices(["prices<br/>tool: ticket_price"])
n_help["help"]
n_reply["reply"]
n_classify --> n_hours
n_classify --> n_prices
n_classify -->|"other"| n_help
n_hours --> n_reply
n_prices --> n_reply
n_help --> n_reply
n_reply --> aa_end((end))
classDef pause stroke-dasharray: 5 5class Ask(BaseModel):
question: str
place: str = ""
day: str = ""
topic: str = ""
facts: dict[str, Any] = Field(default_factory=dict)
answer: str = ""
def classify(state: Ask) -> dict[str, Any]:
text = state.question.lower()
place = next((name for name in MUSEUMS if name.lower() in text), "City Museum")
day = next((d for d in DAYS if d.lower() in text), "Saturday")
topic = "hours" if "open" in text else "prices" if "price" in text else "other"
return {"topic": topic, "place": place, "day": day}
def ask_draft(s: Ask) -> str:
f = s.facts
if s.topic == "hours" and f["open"]:
return f"Yes: the **{s.place}** is open on {s.day}, {f['hours']}."
if s.topic == "hours":
return f"No: the **{s.place}** is closed on {s.day}s. Other days: {f['hours']}."
if s.topic == "prices":
return f"An adult ticket at the {s.place} costs **{f['adult_eur']:.2f} EUR**."
return "I can check hours and prices, compare museums, book tickets and find parks."
ask_flow = (
Flow("ask", state=Ask)
.node("classify", classify)
.node("hours", Call("opening_hours", lambda s: {"place": s.place, "day": s.day}, "facts"))
.node("prices", Call("ticket_price", lambda s: {"place": s.place}, "facts"))
.node("help", lambda s: {"facts": {}})
.node("reply", reply(ask_draft))
.route(
"classify",
Route(lambda s: s.topic, {"hours": "hours", "prices": "prices", "other": "help"}),
)
.edge("hours", "reply")
.edge("prices", "reply")
.edge("help", "reply")
.end("reply")
)
2. Fan-out, fan-in: "Compare ticket prices at three museums"¶
prices is a Map: three ticket_price calls run at once. combine is a Join that waits
for them all, then sorts the prices into a table.
flowchart TD
aa_start((start)) --> n_plan
n_plan["plan"]
n_prices[["prices<br/>fan-out over lookups<br/>tool: ticket_price"]]
n_combine{{"combine<br/>fan-in: prices"}}
n_reply["reply"]
n_plan --> n_prices
n_prices --> n_combine
n_combine --> n_reply
n_reply --> aa_end((end))
classDef pause stroke-dasharray: 5 5class Compare(BaseModel):
question: str
lookups: list[dict[str, Any]] = Field(default_factory=list)
prices: Annotated[list[dict[str, Any]], Merge.append] = Field(default_factory=list)
table: str = ""
answer: str = ""
def pick_museums(state: Compare) -> dict[str, Any]:
return {"lookups": [{"place": name} for name in list(MUSEUMS)[:3]]}
def cheapest_first(state: Compare) -> dict[str, Any]:
rows = sorted(state.prices, key=lambda p: p["adult_eur"])
lines = ["| Museum | Adult |", "|---|---|"]
lines += [f"| {p['place']} | {p['adult_eur']:.2f} EUR |" for p in rows]
return {"table": "\n".join(lines)}
compare_flow = (
Flow("compare_prices", state=Compare)
.node("plan", pick_museums)
.node("prices", Map("lookups", tool="ticket_price", into="prices"), concurrency=3)
.node("combine", Join(["prices"], fn=cheapest_first))
.node("reply", reply(lambda s: f"Cheapest first:\n\n{s.table}"))
.edge("plan", "prices")
.edge("prices", "combine")
.edge("combine", "reply")
.end("reply")
)
3. Approval: "Book two tickets for Saturday"¶
book_tickets has confirm=True, so the call waits for Approve or Deny in its card. A
decision is used once. In the recording the demo approves it; a denial fails the turn.
flowchart TD
aa_start((start)) --> n_details
n_details["details"]
n_price(["price<br/>tool: ticket_price"])
n_book(["book<br/>tool: book_tickets"])
n_reply["reply"]
n_details --> n_price
n_price --> n_book
n_book --> n_reply
n_reply --> aa_end((end))
classDef pause stroke-dasharray: 5 5class Booking(BaseModel):
question: str
place: str = "City Museum"
day: str = "Saturday"
people: int = 2
price: dict[str, Any] = Field(default_factory=dict)
booking: dict[str, Any] = Field(default_factory=dict)
answer: str = ""
def booking_details(state: Booking) -> dict[str, Any]:
text = state.question.lower()
day = next((d for d in DAYS if d.lower() in text), state.day)
people = next((n for word, n in (("one", 1), ("two", 2), ("three", 3)) if word in text), 2)
return {"day": day, "people": people}
def booking_draft(s: Booking) -> str:
total = s.price["adult_eur"] * s.people
b = s.booking
trip = f"{b['place']}, {b['day']}"
return f"Booked: **{b['people']} tickets**, {trip} ({total:.2f} EUR, ref `{b['ref']}`)."
booking_flow = (
Flow("booking", state=Booking)
.node("details", booking_details)
.node("price", Call("ticket_price", lambda s: {"place": s.place}, "price"))
.node(
"book",
Call(
"book_tickets",
lambda s: {"place": s.place, "day": s.day, "people": s.people},
"booking",
),
)
.node("reply", reply(booking_draft))
.edge("details", "price")
.edge("price", "book")
.edge("book", "reply")
.end("reply")
)
4. Loop: "Find a park with a playground near the river"¶
search looks within 0.5 km. check doubles the distance when no park has a playground,
and the route goes back to search, at most 3 times. Here it widens twice.
flowchart TD
aa_start((start)) --> n_search
n_search(["search<br/>tool: find_parks"])
n_check["check"]
n_reply["reply"]
n_search --> n_check
n_check -.->|"widen (max 3)"| n_search
n_check -->|"done"| n_reply
n_reply --> aa_end((end))
classDef pause stroke-dasharray: 5 5class ParkSearch(BaseModel):
question: str
within_km: float = 0.5
found: list[dict[str, Any]] = Field(default_factory=list)
matches: list[str] = Field(default_factory=list)
answer: str = ""
def enough_parks(state: ParkSearch) -> dict[str, Any]:
matches = [p["name"] for p in state.found if p["playground"]]
return {"matches": matches} if matches else {"within_km": state.within_km * 2}
def park_draft(s: ParkSearch) -> str:
if not s.matches:
return "No park with a playground by the river is close enough."
return f"**{s.matches[0]}** has a playground by the river, within {s.within_km:g} km."
park_flow = (
Flow("park_search", state=ParkSearch)
.node(
"search", Call("find_parks", lambda s: {"near": "river", "within_km": s.within_km}, "found")
)
.node("check", enough_parks)
.node("reply", reply(park_draft))
.edge("search", "check")
.route(
"check",
Route(lambda s: "done" if s.matches else "widen", {"widen": "search", "done": "reply"}),
max_loops=3,
on_exhausted="reply",
)
.end("reply")
)
5. All together: "Plan a rainy day for a family"¶
The forecast routes to indoor plans. check fans out over two museums and shortlist joins
them. With fewer than two family picks, it loops back for the next two.
flowchart TD
aa_start((start)) --> n_forecast
n_forecast(["forecast<br/>tool: forecast"])
n_outdoor["outdoor"]
n_candidates["candidates"]
n_check[["check<br/>fan-out over lookups<br/>tool: family_check"]]
n_shortlist{{"shortlist<br/>fan-in: check"}}
n_reply["reply"]
n_forecast -->|"rain"| n_candidates
n_forecast -->|"sun"| n_outdoor
n_outdoor --> n_reply
n_candidates --> n_check
n_check --> n_shortlist
n_shortlist -.->|"more (max 2)"| n_candidates
n_shortlist -->|"done"| n_reply
n_reply --> aa_end((end))
classDef pause stroke-dasharray: 5 5INDOOR = (("Harbour Gallery", "City Museum"), ("Science Hall", "Clock Tower"))
class DayPlan(BaseModel):
question: str
weather: dict[str, Any] = Field(default_factory=dict)
round: int = 0
lookups: list[dict[str, Any]] = Field(default_factory=list)
checked: Annotated[list[dict[str, Any]], Merge.append] = Field(default_factory=list)
picks: list[str] = Field(default_factory=list)
answer: str = ""
def next_places(state: DayPlan) -> dict[str, Any]:
batch = INDOOR[min(state.round, len(INDOOR) - 1)]
return {"lookups": [{"place": name} for name in batch], "round": state.round + 1}
def shortlist(state: DayPlan) -> dict[str, Any]:
return {"picks": [c["place"] for c in state.checked if c["family"]]}
def day_draft(s: DayPlan) -> str:
if not s.picks:
return f"It will be {s.weather['sky']}, up to {s.weather['high_c']} C: a day in the parks."
stops = " then ".join(f"**{place}**" for place in s.picks)
return f"Rain all day (up to {s.weather['high_c']} C), so stay inside: {stops}."
day_flow = (
Flow("day_plan", state=DayPlan)
.node("forecast", Call("forecast", {"day": "Sunday"}, "weather"))
.node("outdoor", lambda s: {"picks": []})
.node("candidates", next_places)
.node("check", Map("lookups", tool="family_check", into="checked"), concurrency=2)
.node("shortlist", Join(["check"], fn=shortlist))
.node("reply", reply(day_draft))
.route("forecast", Route(lambda s: s.weather["sky"], {"rain": "candidates", "sun": "outdoor"}))
.edge("candidates", "check")
.edge("check", "shortlist")
.route(
"shortlist",
Route(
lambda s: "done" if len(s.picks) >= 2 else "more",
{"more": "candidates", "done": "reply"},
),
max_loops=2,
on_exhausted="reply",
)
.edge("outdoor", "reply")
.end("reply")
)
--check runs all five questions offline and prints the answers. See also Chat with your
agent and Navigate the graph.