Finished design for Python API for executing prompts with tools, including debug hooks

[simonw]

This is a blocker for lots of things, including:

• Debug/approval mechanism for CLI prompts with tools #1005

[simonw]

Here's what I have now, it's not right yet:

for token in model.chain("rs in strawberry", tools=[count_char_in_text]):
    print(token, sep="", flush=True)

This uses ChainResponse which iterates over tokens in all responses, executing tools along the way to trigger further responses:

llm/llm/models.py

Lines 1054 to 1059 in b10506b

	def __iter__(self) -> Iterator[str]:
	for response in self.responses():
	if isinstance(response, str):
	yield response
	else:
	yield from response

Which uses:

llm/llm/models.py

Lines 982 to 1052 in b10506b

	def responses(
	self, details=False
	) -> Iterator[
	Union[Response, str]
	]: # Iterator[Union[Response, AsyncResponse, str]]:
	prompt = self.prompt
	count = 0
	response = Response(
	prompt,
	self.model,
	self.stream,
	key=self._key,
	conversation=self.conversation,
	)
	while response:
	count += 1
	yield response
	self._responses.append(response)
	if count > self.chain_limit:
	raise ValueError(f"Chain limit of {self.chain_limit} exceeded. ")
	tool_calls = response.tool_calls()
	if not tool_calls:
	return
	tools_by_name = {
	tool.name: tool.implementation for tool in response.prompt.tools
	}
	tool_results = []
	for tool_call in tool_calls:
	if details:
	yield "\nTool call requested: {}({})\n".format(
	tool_call.name,
	", ".join(
	f"{k}={repr(v)}" for k, v in tool_call.arguments.items()
	),
	)
	implementation = tools_by_name.get(tool_calls[0].name)
	if not implementation:
	# TODO: send this as an error instead?
	raise ValueError(f"Tool {tool_call.name} not found")
	try:
	result = implementation(**tool_call.arguments)
	if not isinstance(result, str):
	result = json.dumps(result, default=repr)
	if details:
	yield f"\n{result}\n"
	except Exception as ex:
	if details:
	yield f"\nError:\ns{ex}\n"
	result = f"Error: {ex}"
	tool_results.append(
	ToolResult(
	name=tool_call.name,
	output=result,
	tool_call_id=tool_call.tool_call_id,
	)
	)
	if not tool_results:
	break
	response = Response(
	Prompt(
	"",
	self.model,
	tools=response.prompt.tools,
	tool_results=tool_results,
	options=self.prompt.options,
	),
	self.model,
	stream=self.stream,
	key=self._key,
	conversation=self.conversation,
	)

I also have this .details() thing but it's not well designed at all:

for token in model.chain("rs in strawberry", tools=[count_char_in_text]).details()
    print(token, sep="", flush=True)

That allows the tool debug stuff to emit a string directly:

llm/llm/models.py

Lines 1061 to 1066 in b10506b

	def details(self) -> Iterator[str]:
	for thing in self.responses(details=True):
	if isinstance(thing, str):
	yield thing
	else:
	yield from thing

Thanks to this code:

llm/llm/models.py

Lines 1010 to 1026 in b10506b

	if details:
	yield "\nTool call requested: {}({})\n".format(
	tool_call.name,
	", ".join(
	f"{k}={repr(v)}" for k, v in tool_call.arguments.items()
	),
	)
	implementation = tools_by_name.get(tool_calls[0].name)
	if not implementation:
	# TODO: send this as an error instead?
	raise ValueError(f"Tool {tool_call.name} not found")
	try:
	result = implementation(**tool_call.arguments)
	if not isinstance(result, str):
	result = json.dumps(result, default=repr)
	if details:
	yield f"\n{result}\n"

[simonw]

I think a regular model response already has access to the information needed for most of a tool execution abstraction. I could have a method like this:

tool_results = response.execute_tool_calls()

Those tool_results could then be passed to the next prompt manually.

I want the ability to intercept tool calls and show debug information about them AND give the user a chance to cancel. This could work:

response.execute_tool_calls(
    before_call=before_call,
    after_all=after_call
)

Those callbacks get passed the relevant dataclasses - the before one can raise a special llm.CancelToolCall exception to cancel the procedure.

def before_call(tool, tool_call):
    print(tool)
    print(tool_call)
    if "evil" in repr(tool_call.arguments):
        raise llm.CancelToolCall("evil")

def after_call(tool, tool_call, tool_result):
    print(tool_result)

Relevant dataclasses:

llm/llm/models.py

Lines 181 to 192 in b10506b

	@dataclass
	class ToolCall:
	name: str
	arguments: dict
	tool_call_id: Optional[str] = None
	@dataclass
	class ToolResult:
	name: str
	output: str
	tool_call_id: Optional[str] = None

[simonw]

I could make those before and after parameters part of the .chain() methods too, then I won't need that weird .details() path at all.

I think **options is an anti-pattern - I'm going to drop that from .chain() and plan to drop it from .prompt() as well in favor of an options={...} thing that can also take options=model.Options(...) (perhaps if that turns out to have nice ergonomics)

[simonw]

llm -f issue:https://github.com/simonw/llm/issues/1007 -m o4-mini \
  'show me full signature for execute_tool_calls taking into account signature of callbacks'

https://gist.github.com/simonw/51fdac66985ece3afa269be01396ce2b

    def execute_tool_calls(
        self,
        *,
        before_call: Optional[Callable[[Tool, ToolCall], None]] = None,
        after_call:  Optional[Callable[[Tool, ToolCall, ToolResult], None]] = None,
    ) -> Sequence[ToolResult]:

[simonw]

import llm

def print_backwards(s: str) -> str:
    "Print something backwards"
    backwards = s[::-1]
    print(backwards)
    return backwards

model = llm.get_model("gpt-4.1-mini")
response = model.prompt("reverse panda", tools=[print_backwards])
print(response)
response.tool_calls()

# Now execute it
tool_results = response.execute_tool_calls()
print(tool_results)

# Running a second prompt is a bit tricky, need to pass in the previous response somehow

Here's code to exercise the new hooks:

import llm
from rich.pretty import pprint

def reverse_string(s):
    return s[::-1]

model = llm.get_model("gpt-4.1-mini")
response = model.prompt("reverse panda", tools=[reverse_string])

response.execute_tool_calls(
    before_call=lambda *args: pprint(args),
    after_call=lambda *args: pprint(args)
)

Output looks like this:

(
    Tool(
        name='reverse_string',
        description=None,
        input_schema={'properties': {'s': {'type': 'string'}}, 'required': ['s'], 'type': 'object'},
        implementation=<function reverse_string at 0x102f15ee0>
    ),
    ToolCall(
        name='reverse_string',
        arguments={'s': 'panda'},
        tool_call_id='call_F1epLnBEAO6JkrHBeNPFZd3h'
    )
)
(
    Tool(
        name='reverse_string',
        description=None,
        input_schema={'properties': {'s': {'type': 'string'}}, 'required': ['s'], 'type': 'object'},
        implementation=<function reverse_string at 0x102f15ee0>
    ),
    ToolCall(
        name='reverse_string',
        arguments={'s': 'panda'},
        tool_call_id='call_F1epLnBEAO6JkrHBeNPFZd3h'
    ),
    ToolResult(name='reverse_string', output='adnap', tool_call_id='call_F1epLnBEAO6JkrHBeNPFZd3h')
)

[simonw]

I'm going to keep the chain() thing (with the **options turned into options={...} but switch it to calling the new execute_tool_calls() method.