Chatbot & Pydantic AI

Overview

The Aventur chatbot (“Avril”) is an AI assistant for Aventur Wealth clients. It is built on Pydantic AI and uses AWS Bedrock (Claude) as the LLM backend. The chatbot handles both free-form text messages and structured “data input” (e.g. form submissions), with tools for company info, user data, goals, holdings, health score, consent, addresses, relations, and support tickets.

Key characteristics:

• Pydantic AI for agent definition, tool calling, and streaming
• pydantic-ai-slim[bedrock] for Bedrock integration (see pyproject.toml)
• Two main agents: text input (conversational + tools) and data input (post-form confirmation)
• Conversation state in Redis (cache) + S3 (persistence)
• Per-conversation consent (V3 API); optional Bedrock guardrail for advice detection
• Advice classification via a separate Lambda before sending to the LLM

Dependencies

From backend/pyproject.toml:

"pydantic-ai-slim[bedrock]>=0.2.11,<1.0.0",
"pydantic-settings>=2.5.2",
"pydantic>=2.10.0",

• pydantic-ai-slim[bedrock]: Pydantic AI runtime with Bedrock provider; used for all agent runs.
• pydantic / pydantic-settings: Used for request/response DTOs, settings, and validation across the app.

Architecture

High-level flow

1. API (V2 or V3) receives a message (text or data layout).
2. PostMessage service:
   • Resolves or creates conversation and builds ConversationStateManager.
   • Dispatches to text input or data input path.
3. Text path: Optional advice classification → optional guardrail → text input agent (Pydantic AI) with tools; streamed response and tool outputs are coordinated via a queue.
4. Data path: OperationDispatcher runs the operation (e.g. update personal details), then data input agent generates a short confirmation.
5. Conversation history (including system prompt, greeting, and messages) is stored and later passed back as message_history to the agent.

Main components

Pydantic AI usage

1. Bedrock model

Location: gateway_accessors/aws/bedrock/accessor.py

• BedrockGateway uses pydantic_ai.providers.bedrock.BedrockProvider and pydantic_ai.models.bedrock.BedrockConverseModel.
• get_model(model_name: ModelName) returns a BedrockConverseModel used by all agents.

Model names (gateway_accessors/aws/bedrock/constants.py):

• Primary: claude_sonnet_4_6_inference_eu (e.g. eu.anthropic.claude-sonnet-4-6).
• Others (Haiku, older Sonnets) are available in the enum for other use cases.

2. User text input agent

Location: app/chatbot/agents/user_text_input/agent.py

• Purpose: Main conversational agent; answers questions and calls tools (context, company, user data, goals, holdings, health score, consent, support, etc.).
• Creation: create_text_input_agent(llm_gateway: BedrockGateway) -> Agent[AgentDependencies].
• Pydantic AI usage:
  • Agent(instructions=..., model=..., deps_type=AgentDependencies, output_type=str, model_settings=ModelSettings(temperature=0.0, parallel_tool_calls=True), tools=...)
  • Tools are wrapped with make_enum_safe() so IntEnum parameters get documented for the LLM.
• Run pattern: agent.iter(user_prompt=..., deps=..., message_history=..., model_settings=...) so the service can drive the iterator, interleave tool results from a queue, and stream text deltas. Optional model_settings pass Bedrock guardrail config when “reference words” are detected.

Instructions: Short behavioral prompt (e.g. be succinct, run tools in one go, don’t echo user data, run get_current_date_time for time-relative queries). The full system prompt (identity, tone, company info, examples, guardrails) is not passed here; it is in the conversation seed as the first message (see below).

3. User data input agent

Location: app/chatbot/agents/user_data_input/agent.py

• Purpose: After a structured operation (e.g. “update personal details”), generates a brief confirmation to the user.
• Creation: create_data_input_agent(user_id, llm_gateway, results: OperationResults) -> Agent[None, Any].
• Pydantic AI usage:
  • Agent(instructions=..., model=..., output_type=str, model_settings=ModelSettings(temperature=0.0), tools=[null_tool]).
  • Instructions are built from OperationResults (e.g. “The user has submitted data… outcome: SUCCESS…”).
• Run pattern: agent.run_stream(user_prompt="Continue as per your instructions.", message_history=...); response is streamed and then conversation is updated (without adding the synthetic “Continue” user message to history).

4. Insight message agent

Location: app/chatbot/agents/insight_message/agent.py

• Purpose: Generates a short “insight” about the user’s financial health from timeseries data (used outside the chat UI, e.g. health score views).
• Pydantic AI usage: Agent(model=..., output_type=InsightMessageDTO, model_settings=ModelSettings(temperature=0.0)) with no tools; agent.run(user_prompt=..., output_type=InsightMessageDTO).

5. Conversation labeller agent

Location: app/chatbot/agents/conversation_labeller/agent.py

• Purpose: Placeholder/labeller agent (model only, null_tool).
• Pydantic AI usage: Agent(model=..., output_type=str, model_settings=..., tools=[null_tool]).

6. Enum lookup agent(s)

Location: app/chatbot/agents/enum_lookup/agent.py

• Purpose: Resolve names to IDs or list indices (e.g. country names → IDs, goal names → indices) during tool execution.
• Pydantic AI usage: Small agents created with create_lookup_agent(ctx, instructions, lookup_tool); run via agent.iter(user_prompt=..., deps=ctx.deps) and the service inspects tool-call nodes and ToolReturnPart to get the list of IDs/indices. Handles UnexpectedModelBehavior, ModelRetry, AgentRunError.

Agent dependencies (deps)

Location: app/chatbot/agents/models.py

@dataclass
class AgentDependencies:
    logger: LoggingService
    llm_gateway: BedrockGateway
    user_id: int
    jwt_token: str
    internal_client: InternalClient
    tool_output_queue: asyncio.Queue[OutboundDTO]

• Injected into the text input agent as deps so tools can call internal APIs (with user_id, jwt_token, internal_client), log, and enqueue structured UI payloads (OutboundDTO) for the client.
• Lookup agents receive the same AgentDependencies via RunContext[AgentDependencies] so they share the same Bedrock gateway and logger.

System prompt and conversation seed

• System prompt content is defined in app/chatbot/context/system_prompt.py (identity, tone, format, markdown, company info, examples, guardrails) and exposed as SYSTEM_PROMPT (a SystemPrompt Pydantic model).
• Conversation seed is built in conversation_history_interface.create_conversation_seed(system_prompt, opening_greeting):
  • First message: SystemPromptPart with format_as_xml(system_prompt.model_dump(exclude_none=True), root_tag="initial_context").
  • Second message: assistant TextPart with the opening greeting (e.g. “Hi there, I’m Avril. How can I help?”).
• This seed is the start of every new conversation’s message_history. When the user sends a message, format_for_agent() returns the agent-facing history (no OutboundDTO blobs), which is passed as message_history into agent.iter(...).
• So the text input agent gets the full system context from the first message in history, plus its own short instructions on the Agent (e.g. tool ordering and brevity).

Tools (text input agent)

Tools are grouped in modules and re-exported as lists; the text input agent uses a single combined list, with each tool wrapped by make_enum_safe() for IntEnum docs.

Terminal tools: Consent tools are “terminal”: when the next node is a single ToolReturnPart for a terminal tool, the service stops the agent loop and does not stream a final model reply (to avoid an extra LLM call). Conversation is still persisted with a synthetic tool result.

IntEnum handling: intenum_helpers.make_enum_safe(tool) introspects tool parameters for IntEnums (e.g. Goals), generates value-to-name docs, and uses a Pydantic AI Tool with a prepare that appends this to the tool description so the model can map user intent to IDs.

Message flow (text input)

1. PostMessage.process_message gets a MessageDTO; if it’s not UserTextDTO, it goes to the data-input path.
2. Text path:
   • Advice classification Lambda runs on the user message; if classified as advice, the service yields a canned “financial advice” response and support ticket form, then returns.
   • If “reference words” are present and guardrail config is set, model_settings includes bedrock_guardrail_config for the next run.
   • create_text_input_agent is called; ConversationStateManager.format_for_agent returns message_history (seed + prior turns, agent-facing only).
   • agent.iter(user_prompt=dto.message, deps=agent_dependencies, message_history=starting_message_history, model_settings=model_settings) is started.
   • The service runs an async loop: wait for either (a) the next node from the agent iterator or (b) an item from tool_output_queue. Tool results are yielded to the client and enqueued as outbound DTOs; when the next node is a ModelRequest (model turn), the service streams PartStartEvent / PartDeltaEvent (text) to the client and tracks timestamps. If the node is a lone terminal-tool return, the loop exits without another model call.
   • Optional guardrail: If the streamed text was exactly the guardrail UUID, the service treats it as advice and runs the same advice-detected path (canned message + support ticket).
   • Finally, new_messages (and any pending tool outbound DTOs) are merged and saved via ConversationStateManager.save_conversation.

Message flow (data input)

1. PostMessage recognises a data layout (e.g. personal details, addresses, goals, support ticket).
2. OperationDispatcher.dispatch(dto) runs the corresponding internal API calls (PUT/POST with user_id, jwt_token).
3. Results are summarized as OperationResults (operation group, type, list of result texts, optional context).
4. If any result is not “success”, a SetIncompleteDTO is yielded.
5. run_data_input_agent builds instructions from OperationResults, creates create_data_input_agent(…, results) with null_tool, then runs with agent.run_stream(user_prompt="Continue as per your instructions.", message_history=...) and streams the assistant text. New messages from the run (excluding the synthetic user message) are saved to the conversation.

API surface

• V3 (preferred): api/routers/api_v3/chatbot/
  • GET /chatbot/messages – get messages (per-conversation consent checked).
  • POST /chatbot/conversations/{conversation_id}/message – send message (streaming); body is a discriminated union (RequestSchema: user text or data layouts).
  • GET /chatbot/new-conversation – new conversation (optional from_insight with health_score_id and goal_id).
  • PUT /chatbot/conversations/{conversation_id}/consent – grant consent for that conversation.
• V2 (deprecated): same logical operations with global consent and flat message list; still uses the same PostMessage and agents.

Request body discriminator: data_layout (e.g. user_text_layout, user__personal_details_layout, …). chatbot_requests.RequestSchema maps each layout to the corresponding DTO (e.g. UserTextDTO, User_PersonalDetailsDTO).

Configuration and dependency injection

• ChatbotServices (in common/containers/chatbot_containers.py) wires:
  • PostMessage with llm_gateway=gateway.bedrock_gateway, conversation_history_store, advice_classification_lambda_client, and optional guardrail_id / guardrail_version from config.
  • ConversationHistoryStore with Redis (cache) and S3 (storage).
  • GetMessages, NewConversation, Consent services.
• Bedrock and guardrail IDs/versions typically come from config (e.g. env / settings).

Constants and enums

• app/chatbot/constants.py: CHATBOT_ENDPOINT_ROOT, OPERATIONS_ENDPOINT_ROOT, OperationGroup (e.g. User_PersonalDetails, User_Goals), OperationResultText (SUCCESS, NO_CONTENT, INSUFFICIENT_PERMISSIONS, etc.), EXECUTION_RESULT_MAP (HTTP status → OperationResultText).
• app/chatbot/context/tool_context.py: GOAL_TYPE_CONTEXT string for goal-type descriptions used in tools/prompts.

Testing and evals

• Unit tests: tests/unit/chatbot/ – agents (user_text_input, user_data_input, insight_message, enum_lookup), tools, services, interface.
• Evals: evals/agents/ – e.g. user_text_input, user_data_input, insight_message, enum_lookup; case definitions and message histories live under evals/agents/.
• Integration: tests/integration/ includes gateway accessors (e.g. Bedrock).
• Acceptance: tests/acceptance/chatbot/ – message and consent flows.

Summary

This document describes how the chatbot is implemented and how it uses Pydantic AI and Bedrock end to end.

Interview angle

• “Design a production chatbot.” - conversation state with a compaction strategy, retrieval for grounding, structured output where it feeds a system, guardrails on input and output, streaming for perceived latency, and per-turn tracing with cost. The state and cost parts are what separate production from demo.
• “How do you manage growing conversation history?” - keep the system prompt and recent turns verbatim, summarise the middle, and note that rewriting history invalidates prefix caching from that point, so compact infrequently in larger batches. See ../12_context_engineering/01_context_management.md.
• “How do you keep it on topic and safe?” - a scoped system prompt, input classification for out-of-scope requests, output validation before display, and an escalation path to a human. Prompt instructions alone are not a control.
• “What do you monitor?” - task completion or escalation rate, tokens and cost per conversation, p95 latency and time-to-first-token, refusal rate in both directions, and a sampled groundedness score.