Progress and questions¶
Two settings connect an Agent to the outside:
| Setting | Role | Default |
|---|---|---|
reporter |
Receives progress events. Only watches, never changes the run | Terminal |
human |
Answers agent.ask_human |
Terminal |
The terminal¶
By default the Agent prints progress and asks questions in the terminal:
[turn 1] thinking
Let me look at main.py first.
tool read_file(path="main.py")
done 1.2KB
[turn 2] thinking
The bug is on line 3.
done: is_answered (2 turns)
[turn N] thinkingstarts each model request. The model's text follows as it streams in.- Indented lines are tool calls and their results.
- The last line says why the run stopped and how many turns it took. When the Model has prices, it also shows the
cost, for example
done: is_answered (5 turns, ~$0.42).
Other last lines:
done: reached limit(50), the task may be unfinished (50 turns)
done: error TimeoutError: took too long (3 turns)
done: interrupted by user (3 turns)
| To | Use |
|---|---|
| Hide the model's streamed text, keep the other lines | Agent(..., reporter=Terminal(show_text=False)) |
| Show nothing | Agent(..., reporter=None) |
| Allow no questions | Agent(..., human=None). ask_human then raises NoHumanError |
Write a Reporter¶
Subclass Reporter and override only the events you need. Every event does nothing by default.
import logging
from alpineagents import Reporter
log = logging.getLogger("agent")
class LogReporter(Reporter):
def on_tool_start(self, state, call):
log.info("turn %d: %s(%s)", state.turn, call.name, call.args)
def on_run_end(self, state, error):
log.info("stopped by %s after %d turns", state.stopped_by, state.turn)
from alpineagents import Agent
agent = Agent(model="claude-sonnet-5", reporter=LogReporter())
| Event | When |
|---|---|
on_run_start(state) |
run starts |
on_think_start(state) |
Right before think or ask sends a request |
on_text(state, chunk) |
Each piece of model text as it streams in |
on_think_end(state, reply) |
Right after the reply ends |
on_tool_start(state, call) |
Right before a tool runs |
on_tool_end(state, call, result, outcome) |
Right after a tool call ends. outcome.kind says how: "done", "error", "denied", ... |
on_context_change(state, change) |
The context was compacted, restarted with start_from, cleared or rolled back |
on_model_event(state, event) |
The Model reported something outside the reply, such as a fallback |
on_run_end(state, error) |
run ends, always. error is None on a normal finish |
- Events can come from several threads. For example, a tool that calls
state.denyruns on a worker thread, and Agents in different threads share the default Terminal. Make the Reporter thread-safe. - Exceptions raised in a Reporter propagate.
Write a Human¶
Subclass Human and implement ask:
from typing import Literal, get_args, get_origin
from alpineagents import Human
class Unattended(Human):
"""For runs nobody watches. Says no to every yes/no question."""
def ask(self, state, prompt, returns=str):
if returns is bool:
return False
if get_origin(returns) is Literal and "no" in get_args(returns):
return "no"
raise RuntimeError(f"Nobody can answer this: {prompt}")
write_file and careful are from Ask before a tool runs:
agent = Agent(
model="claude-sonnet-5",
tools=[write_file],
loop=careful,
human=Unattended(),
)
returnsisstr,boolor aLiteral[...]of choices. Return a value of that type.- For a person reached asynchronously, such as on a web page, implement
async def aask(...)instead and useagent.aask_human.