Message Steering

Send real-time guidance to agents during long-running execution

Message Steering lets you send guidance to a running agent without restarting it.

from praisonaiagents import Agent

agent = Agent(
    name="researcher",
    instructions="You are a research assistant",
    message_steering=True,
)

agent.start("Write a brief overview of agentic RAG.")
agent.steer("Focus on business impact, keep under 200 words", priority=10)

The user steers a long-running task mid-flight; guidance is queued and applied at the next safe boundary.

Quick Start

1. Simple Usage

   Enable with message_steering=True and use agent.steer() to send guidance.

   from praisonaiagents import Agent
   
   agent = Agent(
       name="researcher",
       instructions="You are a research assistant",
       message_steering=True,
       llm="gpt-4o-mini",
   )
   
   agent.steer("Focus on business impact, keep under 200 words", priority=10)
   
   agent.start("Summarise the latest AI trends")

2. Threaded Pattern

   Send steering messages while the agent is actively running.

   import threading, time
   from praisonaiagents import Agent
   
   agent = Agent(name="writer", instructions="You write articles", message_steering=True)
   
   def run():
       return agent.start("Write a comprehensive article about quantum computing")
   
   t = threading.Thread(target=run); t.start()
   
   time.sleep(0.5); agent.steer("Make it beginner-friendly")
   time.sleep(1.0); agent.steer("Include practical applications", priority=15)
   time.sleep(1.5); agent.steer("Keep under 300 words total", priority=20)
   
   t.join()

3. Async Pattern

   Use async execution with real-time steering.

   import asyncio
   from praisonaiagents import Agent
   
   async def main():
       agent = Agent(name="analyst", instructions="You analyse data", message_steering=True)
       task = asyncio.create_task(agent.astart("Analyse impact of AI on job markets"))
   
       await asyncio.sleep(0.3); agent.steer("Focus on positive opportunities")
       await asyncio.sleep(0.7); agent.steer("Include specific examples", priority=12)
   
       print(await task)
   
   asyncio.run(main())

4. Steering a Long Multi-Tool Run

   Steer an agent while it is deep inside a multi-step tool loop — the guidance lands on the next loop iteration, not only when a chat step returns.

   import threading, time
   from praisonaiagents import Agent
   from praisonaiagents.tools import duckduckgo  # any tool works — this is illustrative
   
   agent = Agent(
       name="researcher",
       instructions="Research topics and write a structured report.",
       tools=[duckduckgo],
       message_steering=True,
   )
   
   def run():
       return agent.start("Research solar market trends and write a 500-word summary")
   
   t = threading.Thread(target=run); t.start()
   
   # Agent is now in a multi-step tool loop (search → read → search → summarise).
   # The steer lands on the next loop iteration, not only when the chat step returns.
   time.sleep(1.0); agent.steer("Focus on the EU market only", priority=20)   # URGENT
   time.sleep(2.0); agent.steer("Skip anything before 2024", priority=30)     # INTERRUPT
   
   t.join()

   Example user flow:

   User: research solar market trends and write a 500-word summary
   Agent: (tool loop: searching → reading → summarising …)
   User: (mid-run) focus on the EU market only          [URGENT]
   Agent: (next iteration picks up guidance, re-scopes the next search)
   User: (mid-run) skip anything before 2024            [INTERRUPT]
   Agent: ✅ 500-word EU Solar Market Trends (2024+) summary

How It Works

Message steering injects guidance at the top of every chat step and every tool-loop iteration, so a long multi-tool run picks up new user guidance on its very next model call — without restarting the run.

sequenceDiagram
    participant User
    participant Agent
    participant Loop as Tool Loop
    participant LLM
    participant Tools

    User->>Agent: agent.start("long task")
    Agent->>LLM: prompt
    LLM-->>Agent: tool_calls
    Agent->>Loop: enter tool-calling loop

    Note over User,Loop: user steers mid-run
    User->>Agent: agent.steer("focus on EU market", priority=20)

    Loop->>Loop: drain pending steering notes
    Loop->>LLM: prompt + [URGENT USER GUIDANCE]: focus on EU market
    LLM-->>Loop: revised tool_calls
    Loop->>Loop: cancel_token.is_set()? (before tool dispatch)
    Loop->>Tools: dispatch tools (only if not cancelled)
    Tools-->>Loop: results
    Loop-->>Agent: final response
    Agent-->>User: result

Steering messages are drained at the start of each chat step and each tool-loop iteration, then injected as user messages with a priority-aware header. The agent picks them up on its very next model call, whether it is between chat turns or deep inside a multi-tool loop.

Mid-Tool-Loop Steering

Steering messages land on the very next tool-loop iteration, so multi-step tool runs pick up new guidance without a restart.

• Steering messages queued via agent.steer(...) are checked at the top of every tool-loop iteration, so multi-step tool runs pick them up without a restart.
• INTERRUPT and URGENT priorities keep their headers when injected mid-loop, so the model sees [INTERRUPT USER GUIDANCE] / [URGENT USER GUIDANCE] instead of being silently downgraded to plain guidance.
• /stop (via cancel_token or agent.interrupt_controller) is re-checked immediately before dispatching tool calls, so a cancel mid-iteration skips running the pending tools instead of executing them first.

Priority Levels

CLI Usage

Enable message steering from command line:

praisonai --message-steering "Write a detailed report about AI trends"

The --message-steering flag sets message_steering=True on the agent.

YAML Usage

Enable steering per role in YAML configuration:

framework: praisonai
roles:
  researcher:
    role: Research Assistant
    goal: Conduct thorough research
    message_steering: true   # Enable per-role
    tasks:
      research_task:
        description: Research AI trends and write report
        expected_output: Comprehensive research report

Configuration Options

When message_steering=False (default), there’s zero overhead – the feature has no performance impact when disabled.

Agent Methods

Custom Steering Backends

Implement MessageSteeringProtocol to plug custom backends:

from praisonaiagents import Agent, MessageSteeringProtocol, SteeringMessage, SteeringPriority

class RedisSteering:  # Implements MessageSteeringProtocol structurally
    def queue_message(self, message: str, priority: int = 5) -> str: ...
    def get_pending_messages(self): ...
    def process_steering(self, context=None) -> bool: ...
    def clear_messages(self) -> int: ...
    def has_pending_messages(self) -> bool: ...
    @property
    def enabled(self) -> bool: ...

agent = Agent(name="bg", message_steering=RedisSteering())

Mid-loop drains rely on repeated has_pending_messages() + dequeue calls, so a custom backend must implement both correctly for mid-run injection to work.

Best Practices

Related