Agently documentation for building AI applications with stable outputs, observable actions, and durable workflows.
Languages: English · 中文
Run a structured survey through a conversational interface. The model:
A conversational agent with a session, plus a per-turn structured output that includes:
The application loop reads the model’s structured output and decides whether to continue.
from agently import Agently
agent = (
Agently.create_agent()
.role(
"You are running a customer onboarding survey. "
"Ask one question at a time. Branch into follow-ups when needed. "
"End the survey when all required slots are filled.",
always=True,
)
.info({
"required_slots": ["company_size", "primary_use_case", "current_tools", "decision_timeline"],
"format": "Reply only via the schema.",
}, always=True)
)
agent.activate_session(session_id="survey-dialog") # multi-turn
state = {"answers": {}}
def step(user_message: str):
return (
agent
.info({"answers_so_far": state["answers"]}, always=False)
.input(user_message)
.output({
"reply_to_user": (str, "What to show the user", True),
"current_slot": (str, "Which slot is being filled", True),
"captured": {
"slot": (str, "Slot just captured (or empty)"),
"value": "captured value (any type)",
},
"survey_complete": (bool, "True only when all required slots are captured", True),
})
.start()
)
# Run
print("Hi! Let's get started. Ready?")
user_text = input("> ")
while True:
result = step(user_text)
print(result["reply_to_user"])
captured = result.get("captured") or {}
if captured.get("slot"):
state["answers"][captured["slot"]] = captured.get("value")
if result["survey_complete"]:
break
user_text = input("> ")
print("\nFinal answers:")
print(state["answers"])
info(answers_so_far, always=False) — the captured state changes every turn; passing it as request-only info means it’s always current without polluting the agent’s persistent prompt.info({"required_slots": [...]}, always=True) — the slot list doesn’t change; pin it to the agent.activate_session() handles that.survey_complete: bool — explicit termination. The application loop trusts this; the model is told only to set it when all required slots are filled.If a value should be one of an enum, use a custom .validate(...):
def value_check(result, ctx):
captured = result.get("captured") or {}
slot = captured.get("slot")
value = captured.get("value")
if slot == "decision_timeline" and value not in ("now", "this_quarter", "this_year", "exploring"):
return {"ok": False, "reason": f"unknown timeline: {value}", "validator_name": "enum"}
return True
See Output Control.
For long surveys (20+ questions), register custom resize handlers that summarize older turns into memo:
agent.set_settings("session.max_length", 12000)
agent.register_session_analysis_handler(analysis_handler)
agent.register_session_resize_handler("summarize_old_turns", resize_handler)
The default Session only trims the window; summarization logic comes from your handler. See Session Memory.
If a single user answer triggers multi-step processing (lookup, validate, score), promote the per-turn handling to a TriggerFlow. The conversation layer stays in the agent loop; one flow runs per turn. See TriggerFlow Orchestration Playbook.
survey_complete: bool as an ensure’d field.validate(...) for value checksinfo(always=True) vs always=False