Skip to main content

​
Agent Loop Documentation

​
Overview

The agent loop is a core feature of the TextLayer system that enables autonomous, iterative conversations between users and AI agents. It allows the AI to engage in multi-step reasoning, tool usage, and problem-solving without requiring user intervention after each step.

​
Architecture Overview

The agent loop follows the Command Pattern architecture established in TextLayer, with clear separation between:
  • Controllers: Handle HTTP requests and route to appropriate commands
  • Commands: Contain business logic for processing chat messages
  • Services: Provide core functionality like LLM communication and tool execution
  • Tools: Specific capabilities the agent can use (SQL queries, reasoning, etc.)

​
Key Components

​
1. Core Agent Loop Implementation

​
ChatClient.agent_loop()

Located in: backend/textlayer/services/llm/client/chat.py 
def agent_loop(
    self,
    messages: List[Dict[str, Any]],
    tools: Optional[Toolkit] = None,
    max_steps: int = 10,
) -> List[Dict[str, Any]]:
This is the heart of the agent loop system. It implements a controlled iterative conversation flow with the following key features: Core Loop Logic:
  • Iterates up to max_steps times (default: 10)
  • Processes chat messages in each iteration
  • Handles tool calls and responses
  • Implements safety mechanisms for error handling and timeouts
Safety Mechanisms:
  • Error Handling: Tracks consecutive errors (max 3) before breaking
  • Timeout Protection: 300-second timeout to prevent infinite loops
  • Thread Completion Detection: Checks if conversation is naturally finished
  • Graceful Termination: Adds termination messages when limits are reached
Flow Control: 
while current_step < max_steps:
    # Safety checks
    if consecutive_errors >= 3:
        break

    if self._is_thread_finished(messages):
        break

    # Timeout check
    reason = self._check_termination_conditions(start_time, request_timeout)
    if reason:
        # Add termination message and final response
        break

    # Process next chat iteration
    if processed := self.chat(messages, stream=False, tools=tools, max_steps=max_steps - current_step):
        messages.extend(processed)

    current_step += 1

​
Helper Methods:

_is_thread_finished(): Determines if conversation has naturally concluded 
def _is_thread_finished(self, messages: List[Dict[str, Any]]) -> bool:
    if not messages:
        return True
    last_message = messages[-1]
    return last_message.get("finish_reason") == "stop" and last_message.get("role") == "assistant"
_check_termination_conditions(): Checks for timeout conditions _terminate_thread(): Creates termination messages when limits are reached

​
2. Command Layer

​
ProcessChatMessageCommand

Located in: backend/textlayer/commands/threads/process_chat_message.py This command encapsulates the business logic for processing chat messages and provides the interface between the controller layer and the LLM services. Key Features:
  • Supports both regular chat and agent loop modes via agent_loop boolean parameter
  • Validates input parameters (messages, stream, max_steps)
  • Sets up the LLM session with fallback models
  • Configures the toolkit of available tools
  • Applies data analysis system prompt
Constructor Parameters: 
def __init__(
    self,
    messages: List[Dict[str, Any]],
    stream: bool = False,
    model: Optional[str] = None,
    max_steps: Optional[int] = None,
    agent_loop: bool = False,  # Key parameter for enabling agent loop
):
Execution Logic: 
if self.agent_loop:
    # Agent loop doesn't support streaming
    return llm_session.agent_loop(
        messages=formatted_messages,
        tools=toolkit,
        max_steps=self.max_steps,
    )
else:
    return llm_session.chat(
        messages=formatted_messages,
        stream=self.stream,
        tools=toolkit,
        max_steps=self.max_steps,
    )

​
3. Controller Layer

​
ThreadController

Located in: backend/textlayer/controllers/thread_controller.py Provides a clean interface for the route layer to process chat messages: 
def process_chat_message(
    self,
    messages: List[Dict[str, Any]],
    stream: bool = False,
    model: Optional[str] = None,
    max_steps: Optional[int] = None,
    agent_loop: bool = False,  # Exposed parameter
) -> Union[List[Dict[str, Any]], Generator[str, None, None]]:

​
4. Route Layer

​
Thread Routes

Located in: backend/textlayer/routes/thread_routes.py Currently, the HTTP API endpoints (/chat and /chat/stream) have agent_loop=False hardcoded, meaning the agent loop is not exposed via the REST API. This suggests it’s primarily used for internal processing or CLI operations.

​
5. CLI Interface

​
CLI Handler

Located in: backend/textlayer/cli/threads/process_chat_message.py Provides a clean interface for programmatic access to the agent loop: 
@observe(name="agent_chat", capture_input=False)
def process_chat_message(messages: List[Dict[str, str]], max_steps: int = 10) -> Dict[str, str]:
Key Features:
  • Enables agent loop mode (agent_loop=True)
  • Returns the last assistant message from the conversation
  • Includes comprehensive error handling
  • Integrates with Langfuse for observability

​
Data Flow

​
1. Initialization Flow

User Input → CLI/Controller → ProcessChatMessageCommand → ChatClient.agent_loop()

​
2. Agent Loop Iteration Flow

1. Check safety conditions (errors, timeout, completion)
2. Call ChatClient.chat() with current messages + tools
3. LLM processes messages and may make tool calls
4. Tools execute and return results
5. Results added to message history
6. Repeat until completion or limits reached

​
Usage Examples

​
1. CLI Usage

from textlayer.cli.threads.process_chat_message import process_chat_message

messages = [
    {"role": "user", "content": "How many customers do we have?"}
]

result = process_chat_message(messages, agent_loop=True, max_steps=5)
# Returns the final assistant message after agent loop processing

​
2. Programmatic Usage

from textlayer.controllers.thread_controller import ThreadController

controller = ThreadController()
messages = [
    {"role": "user", "content": "Analyze customer churn patterns"}
]

all_messages = controller.process_chat_message(
    messages=messages,
    agent_loop=True,
    max_steps=10
)

​
Development Considerations

  • Agent loop is not exposed via REST API (hardcoded to False)
  • Streaming is disabled in agent loop mode
  • Tool execution is synchronous within each iteration

​
Best Practices

When working with the agent loop system:
  1. Set appropriate max_steps: Balance between allowing sufficient iterations and preventing runaway loops
  2. Monitor tool execution: Use Langfuse observability to track tool calls and performance
  3. Handle errors gracefully: The system includes built-in error handling, but consider application-specific error scenarios
  4. Test iteratively: Use the CLI interface for testing before integrating into larger applications

​
Integration with Other Components

The agent loop integrates seamlessly with other TextLayer Core components:
  • Vaul Toolkit: Provides the tools that the agent can use during iterations
  • LLMOps: Monitoring and observability through Langfuse integration
  • FLEX Stack: Built on the same architectural principles
For more information on building tools that work with the agent loop, see the Build a Tool guide.
⌘I