Agent Conversation Loop

The AgentLoop is the central orchestrator of Mistral Vibe. This document details how conversations are managed and executed.

AgentLoop Class Overview

File: vibe/core/agent_loop.py Class: AgentLoop at line 105

Constructor

# agent_loop.py:106-114
def __init__(
    self,
    config: VibeConfig,
    agent_name: str = BuiltinAgentName.DEFAULT,
    message_observer: Callable[[LLMMessage], None] | None = None,
    max_turns: int | None = None,
    max_price: float | None = None,
    backend: BackendLike | None = None,
    enable_streaming: bool = False,
) -> None:

Initialization Sequence

AgentLoop.__init__() [agent_loop.py:105]
    │
    ├─1─► Store base config [agent_loop.py:116]
    │
    ├─2─► Create AgentManager [agent_loop.py]
    │     AgentManager(config_getter, initial_agent=agent_name)
    │     Discovers built-in + custom agents
    │
    ├─3─► Create ToolManager [agent_loop.py]
    │     ToolManager(config)
    │     Discovers and registers all available tools
    │
    ├─4─► Create SkillManager [agent_loop.py]
    │     SkillManager(config_getter)
    │     Discovers SKILL.md files from search paths
    │
    ├─5─► Create APIToolFormatHandler
    │     Handles tool call serialization/deserialization
    │
    ├─6─► Select/create backend
    │     backend_factory = lambda: backend or self._select_backend()
    │
    ├─7─► Setup middleware pipeline
    │     _setup_middleware(max_turns, max_price)
    │     Now includes PlanAgentMiddleware
    │
    ├─8─► Build system prompt
    │     get_universal_system_prompt(tool_manager, config, skill_manager, agent_manager)
    │
    ├─9─► Initialize messages with system prompt
    │     messages = [LLMMessage(role=Role.system, content=system_prompt)]
    │
    ├─10► Initialize stats
    │     AgentStats with pricing from active model
    │
    ├─11► Create SessionLogger
    │     SessionLogger(config.session_logging, session_id)
    │
    └─12► Spawn session migration thread
          Thread(target=migrate_sessions_entrypoint, daemon=True)

Key Properties

Main Entry Point: act()

Method: act() at agent_loop.py

async def act(self, msg: str) -> AsyncGenerator[BaseEvent]:
    self._clean_message_history()
    async for event in self._conversation_loop(msg):
        yield event

This is an async generator that:

1. Cleans any malformed message history
2. Delegates to _conversation_loop()
3. Yields events as they're produced

The Conversation Loop

Method: _conversation_loop() at agent_loop.py

This is the heart of the agent — a multi-turn conversation loop that handles LLM calls, tool execution, and middleware.

Flow Diagram

_conversation_loop(user_msg)
    │
    ├─1─► Append user message
    │     messages.append(LLMMessage(role=Role.user, content=user_msg))
    │
    ├─2─► Yield UserMessageEvent
    │
    ├─3─► Increment steps
    │     stats.steps += 1
    │
    └─4─► Enter main loop
          │
          while not should_break_loop:
          │
          ├─a─► Run middleware before turn
          │     result = await middleware_pipeline.run_before_turn(context)
          │     │
          │     └── Handle result
          │         - STOP: yield event, return
          │         - COMPACT: trigger compaction
          │         - INJECT_MESSAGE: append to last message
          │         - CONTINUE: proceed
          │
          ├─b─► Increment steps
          │
          ├─c─► Perform LLM turn
          │     async for event in _perform_llm_turn():
          │         if is_user_cancellation_event(event):
          │             user_cancelled = True
          │         yield event
          │
          ├─d─► Check break condition
          │     should_break = (
          │         last_message.role != Role.tool AND
          │         finish_reason is not None
          │     )
          │
          ├─e─► Flush messages to observer
          │
          ├─f─► Save interaction via session_logger
          │
          ├─g─► Handle user cancellation
          │     If cancelled, flush and return
          │
          ├─h─► Run middleware after turn
          │     Similar handling to before_turn
          │
          └─i─► Final flush and save

Loop Termination Conditions

The loop terminates when any of these occur:

1. Normal completion: Last message is NOT a tool response and LLM returned a finish_reason
2. Middleware STOP: Turn limit reached or price limit exceeded
3. User cancellation: User interrupted the operation

LLM Turn Execution

Method: _perform_llm_turn()

_perform_llm_turn()
    │
    ├─1─► Call LLM (streaming or non-streaming)
    │     │
    │     ├── If streaming: _stream_assistant_events()
    │     │   Yields AssistantEvents and ReasoningEvents incrementally
    │     │
    │     └── Else: _get_assistant_event()
    │         Returns single AssistantEvent
    │
    ├─2─► Get last message and chunk
    │
    ├─3─► Parse and resolve tool calls
    │     parsed = format_handler.parse_message(last_message)
    │     resolved = format_handler.resolve_tool_calls(parsed, tool_manager)
    │
    ├─4─► Calculate tokens per second
    │
    └─5─► Handle tool calls if any
          if resolved.tool_calls or resolved.failed_calls:
              async for event in _handle_tool_calls(resolved):
                  yield event

Streaming with Reasoning Content

When streaming, the agent now handles ReasoningEvent for models that produce reasoning/thinking tokens:

# During streaming:
if chunk.message.reasoning_content:
    yield ReasoningEvent(content=reasoning_content, message_id=message_id)
if chunk.message.content:
    yield AssistantEvent(content=content, message_id=message_id)

Tool Call Handling

Method: _handle_tool_calls()

_handle_tool_calls(resolved)
    │
    ├─1─► Handle failed tool calls
    │     For each failed call:
    │     - Yield ToolResultEvent with error
    │     - Increment stats.tool_calls_failed
    │     - Append error message to history
    │
    └─2─► Process valid tool calls
          │
          for tool_call in resolved.tool_calls:
          │
          ├─a─► Yield ToolCallEvent
          │
          ├─b─► Get tool instance
          │     tool_instance = tool_manager.get(tool_call.tool_name)
          │
          ├─c─► Check execution permission
          │     decision = await _should_execute_tool(tool, args, tool_call_id)
          │
          ├─d─► If SKIP:
          │     - Increment tool_calls_rejected
          │     - Yield ToolResultEvent with skipped=True
          │     - Append skip message to history
          │     - continue
          │
          └─e─► Execute tool:
                │
                ├── Create InvokeContext
                │   InvokeContext(
                │       tool_call_id=call_id,
                │       approval_callback=self.approval_callback,
                │       agent_manager=self.agent_manager,
                │       user_input_callback=self.user_input_callback,
                │   )
                │
                ├── tool_instance.run(args, ctx=ctx)
                │   Returns AsyncGenerator[ToolStreamEvent | ToolResult, None]
                │
                ├── Yield ToolStreamEvent items (streaming output)
                │
                ├── Yield ToolResultEvent with final result
                │
                └── Handle errors (CancelledError, ToolError, etc.)

Permission Checking

Method: _should_execute_tool()

_should_execute_tool(tool, args, tool_call_id)
    │
    ├─1─► Auto-approve check
    │     If auto_approve: return EXECUTE
    │
    ├─2─► Validate args
    │
    ├─3─► Check allowlist/denylist patterns
    │     result = tool.check_allowlist_denylist(validated_args)
    │     │
    │     ├── ALWAYS: return EXECUTE
    │     └── NEVER: return SKIP with denylist feedback
    │
    ├─4─► Check tool permission config
    │     perm = tool_manager.get_tool_config(tool_name).permission
    │     │
    │     ├── ALWAYS: return EXECUTE
    │     └── NEVER: return SKIP
    │
    └─5─► Ask user for approval
          return await _ask_approval(tool_name, args, tool_call_id)

Agent Switching

Method: switch_agent()

Switches to a different agent profile, rebuilds system prompt, and optionally resets history:

def switch_agent(self, name: str) -> None:
    self.agent_manager.switch_profile(name)
    # Rebuild system prompt with new agent's config
    # Reset tool manager with new config

Context Management

Auto-Compaction

Method: compact()

Triggered by AutoCompactMiddleware when context exceeds threshold:

compact()
    │
    ├─1─► Clean message history
    │
    ├─2─► Save current interaction via session_logger
    │     session_logger.save_interaction(..., agent_profile=self.agent_profile)
    │
    ├─3─► Find last user message
    │
    ├─4─► Request summary from LLM
    │     summary_request = UtilityPrompt.COMPACT.read()
    │     summary_result = await _chat()
    │
    ├─5─► Replace history with summary
    │     messages = [system_message, summary_message]
    │
    ├─6─► Recalculate context tokens
    │
    └─7─► Reset session and middleware

History Clearing

Method: clear_history()

clear_history()
    │
    ├─1─► Save current interaction
    ├─2─► Keep only system message: messages = messages[:1]
    ├─3─► Reset stats
    ├─4─► Reset middleware pipeline
    ├─5─► Reset tool manager state
    └─6─► Reset session logger with new session ID

Event Types

Events yielded by the agent (defined in types.py):

AssistantEvent Structure

class AssistantEvent(BaseEvent):
    content: str
    stopped_by_middleware: bool = False
    message_id: str | None = None

ReasoningEvent Structure

class ReasoningEvent(BaseEvent):
    content: str
    message_id: str | None = None

ToolCallEvent Structure

class ToolCallEvent(BaseEvent):
    tool_name: str
    tool_class: type[BaseTool]
    args: BaseModel
    tool_call_id: str

ToolResultEvent Structure

class ToolResultEvent(BaseEvent):
    tool_name: str
    tool_class: type[BaseTool] | None
    result: BaseModel | None = None
    error: str | None = None
    skipped: bool = False
    skip_reason: str | None = None
    duration: float | None = None
    tool_call_id: str

ToolStreamEvent Structure

class ToolStreamEvent(BaseEvent):
    tool_name: str
    message: str
    tool_call_id: str

State Management

AgentStats

Class: AgentStats at types.py:26

Tracks conversation statistics:

class AgentStats(BaseModel):
    steps: int = 0                           # LLM turn count
    session_prompt_tokens: int = 0           # Total input tokens
    session_completion_tokens: int = 0       # Total output tokens
    tool_calls_agreed: int = 0               # Approved tool calls
    tool_calls_rejected: int = 0             # Rejected tool calls
    tool_calls_failed: int = 0               # Failed tool calls
    tool_calls_succeeded: int = 0            # Successful tool calls
    context_tokens: int = 0                  # Current context size
    last_turn_prompt_tokens: int = 0
    last_turn_completion_tokens: int = 0
    last_turn_duration: float = 0.0
    tokens_per_second: float = 0.0
    input_price_per_million: float = 0.0
    output_price_per_million: float = 0.0
 
    @computed_field
    @property
    def session_cost(self) -> float:         # Calculated cost
        ...

Source File References