I am using a multi-agent flow. I am using create_agent as a supervisor and other sub agents as a tools. the supervisor agent code is as follows:
supervisor = create_agent(
model=model,
system_prompt=supervisor_prompt,
tools=[fetch_patient_information_tool, manage_appointment_tool, book_appointment_tool],
checkpointer=checkpointer,
name="supervisor_agent",
)
One of the tools is defined as follows:
@tool(return_direct=True)
async def fetch_patient_information_tool(request: str, runtime: ToolRuntime) -> str:
"""Use this tool to retrieve patient information such as name, mobile number, date of birth, or address.
This tool handles ONLY patient personal information queries. Call this when the user asks about their
personal details or provides a patient ID to retrieve information.
Args:
request: The user's request for patient information
"""
root_thread_id = runtime.config["configurable"]["thread_id"]
logger.info(f"[patient_agent] 🚀 Invoking with thread_id={root_thread_id}, request: {request}")
# Build config with shared checkpoint namespace for memory continuity
sub_config = extend_runtime_config(runtime, checkpoint_ns=SHARED_NS)
# 💾 Log checkpoint state BEFORE invoke to see what memory is loaded
await log_checkpoint_state(patient_agent, "patient_agent", sub_config)
# Note: checkpointer is already bound to patient_agent at creation time
result = await patient_agent.ainvoke(
{"messages": [{"role": "user", "content": request}]},
config=sub_config
)
# Convert ToolMessages to TOON format
result = convert_tool_messages_to_toon_format(result, tool_names=["mssql_get_patient_info", "mssql_get_patient_by_name_dob"])
# Update the checkpoint with TOON formatted messages
await patient_agent.aupdate_state(sub_config, result)
logger.info(f"[patient_agent] 💾 Updated checkpoint with TOON formatted messages")
logger.info(f"[patient_agent] 🛠️ Sub-agent invocation complete.")
# Log sub-agent's internal tool calls and messages AFTER invoke
log_subagent_result("patient_agent", result)
response = extract_last_ai_message(result)
logger.info(f"[patient_agent] ✅ Final response: {response[:200]}...")
response = strip_markdown(response)
return response.lower()
The sub-agent is defined as follows:
async def patient_information_retrieval_agent(tools, language_id, checkpointer=None):
"""
Create a patient information retrieval agent.
"""
logger.info(f"Creating patient information retrieval agent with checkpointer: {checkpointer}")
prompt_folder = 'english_prompts' if language_id == 2 else 'arabic_prompts'
prompt_path = os.path.join(os.path.dirname(__file__), '..', 'prompts', prompt_folder, 'patient_information_retrieval_prompt.yml')
logger.info(f"Loading patient information retrieval prompt from with language id: {language_id}")
with open(prompt_path, 'r', encoding='utf-8') as f:
prompts = yaml.safe_load(f)
return create_agent(
model=model,
tools=tools,
system_prompt=prompts['patient_information_retrieval_prompt'].format(LANGUAGE_ID=language_id),
name="patient_information_retrieval_agent",
checkpointer=checkpointer,
middleware=[
# SummarizationMiddleware(
# model = model,
# trigger=("tokens", 1000),
# keep=("messages", 5),
# ),
]
)
The other tools and sub_agents are defined in the same manner.
@pawel-twardziak If you need other information’s let me know.
