Skip to content

Core API Reference

Agent

lllm.core.agent.Agent dataclass

Represents a single LLM agent with a specific role and capabilities.

An Agent owns the dialogs it creates. Each dialog is keyed by a user-chosen alias (e.g. 'planning', 'talk_with_coder') that makes the code self-documenting:

agent.open('planning', prompt_args={...})
agent.receive("What's the plan?")
response = agent.respond()

agent.open('execution', prompt_args={...})
agent.switch('execution')
...

For power-user / cross-agent scenarios, call(dialog) still accepts a raw Dialog directly — but the recommended path is alias-based.

Attributes:

Name Type Description
name str

The name or role of the agent (e.g., 'assistant', 'coder').
system_prompt Prompt

The system prompt defining the agent's persona.
model str

The model identifier (e.g., 'gpt-4o').
llm_invoker BaseInvoker

The invoker instance for LLM calls.
model_args Dict[str, Any]

Additional model arguments (temp, top_p, etc.).
max_exception_retry int

Max retries for agent parsing/validation exceptions.
max_interrupt_steps int

Max consecutive tool call interrupts.
max_llm_recall int

Max retries for LLM API errors.
Source code in lllm/core/agent.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
@dataclass
class Agent:
    """
    Represents a single LLM agent with a specific role and capabilities.

    An Agent owns the dialogs it creates. Each dialog is keyed by a
    user-chosen alias (e.g. 'planning', 'talk_with_coder') that makes the
    code self-documenting:

        agent.open('planning', prompt_args={...})
        agent.receive("What's the plan?")
        response = agent.respond()

        agent.open('execution', prompt_args={...})
        agent.switch('execution')
        ...

    For power-user / cross-agent scenarios, ``call(dialog)`` still accepts
    a raw Dialog directly — but the recommended path is alias-based.

    Attributes:
        name: The name or role of the agent (e.g., 'assistant', 'coder').
        system_prompt: The system prompt defining the agent's persona.
        model: The model identifier (e.g., 'gpt-4o').
        llm_invoker: The invoker instance for LLM calls.
        model_args: Additional model arguments (temp, top_p, etc.).
        max_exception_retry: Max retries for agent parsing/validation exceptions.
        max_interrupt_steps: Max consecutive tool call interrupts.
        max_llm_recall: Max retries for LLM API errors.
    """
    name: str # the role of the agent, or a name of the agent
    system_prompt: Prompt
    model: str # the model identifier (e.g., 'gpt-4o'), by default, it from litellm model list (https://models.litellm.ai/)
    llm_invoker: BaseInvoker
    stream_handler: Optional[BaseStreamHandler] = None

    api_type: APITypes = APITypes.COMPLETION
    model_args: Dict[str, Any] = field(default_factory=dict) # additional args, like temperature, seed, etc.
    max_exception_retry: int = 3
    max_interrupt_steps: int = 5
    max_llm_recall: int = 0
    context_manager: Optional[ContextManager] = None  # applied before each LLM call; None = disabled

    # Dialog management
    _dialogs: Dict[str, Dialog] = field(default_factory=dict, repr=False)
    _active_alias: Optional[str] = field(default=None, repr=False)

    def open(self, alias: str, prompt_args: Optional[Dict[str, Any]] = None, session_name: Optional[str] = None, switch: bool = True):
        """
        Create a new dialog owned by this agent, keyed by alias.

        Args:
            alias: the alias for the new dialog.
            prompt_args: the arguments for the system prompt.
            session_name: the name of the session for logging and checkpointing.
            switch: if True, switch to the new dialog after opening. Default is True.
        """
        if alias in self._dialogs:
            raise ValueError(
                f"Dialog '{alias}' already exists on agent '{self.name}'. "
                f"Use .fork('{alias}', ...) or .close('{alias}') first."
            )
        prompt_args = dict(prompt_args) if prompt_args else {}
        dialog = Dialog(
            session_name=session_name or f"{self.name}_{alias}",
            owner=self.name,
        )
        dialog.put_prompt(
            self.system_prompt, prompt_args,
            name='system', role=Roles.SYSTEM,
        )
        self._dialogs[alias] = dialog
        if switch:
            self._active_alias = alias
        return self # for chaining

    def fork(self, alias: str, child_alias: str, last_n: int = 0, first_k: int = 1, switch: bool = True) -> 'Agent':
        """
        Branch an existing dialog into a new child dialog.

        The parent dialog's ``fork()`` handles all lineage bookkeeping
        (parent ↔ child links, split_point, ids).  Agent just stores
        the child under ``child_alias`` and switches to it.

        Args:
            alias: the source dialog to fork from.
            child_alias: the alias for the new child dialog.
            last_n: if >0, drop the last n messages from the copy.
            first_k: if >0, keep the first k messages from the copy. Only used when last_n is >0.
            switch: if True, switch to the new child dialog after forking.

        Raises:
            ValueError: if ``child_alias`` is already in use.
            KeyError: if ``alias`` doesn't exist.
        """
        if child_alias in self._dialogs:
            raise ValueError(
                f"Dialog '{child_alias}' already exists on agent '{self.name}'."
            )
        parent = self._get_dialog(alias)
        child = parent.fork(last_n, first_k)
        self._dialogs[child_alias] = child
        if switch:
            self._active_alias = child_alias
        return self # for chaining

    def close(self, alias: str) -> Dialog:
        """
        Remove a dialog from this agent and return it.

        Useful for archiving, handing off to another system, or just
        cleaning up.  If the closed dialog was active, active becomes None.
        """
        dialog = self._dialogs.pop(alias)
        if self._active_alias == alias:
            self._active_alias = None
        return dialog

    def switch(self, alias: str) -> 'Agent':
        """
        Set the active dialog by alias.  Returns self for chaining.

        Raises:
            KeyError: if ``alias`` doesn't exist.
        """
        if alias not in self._dialogs:
            raise KeyError(
                f"No dialog '{alias}' on agent '{self.name}'. "
                f"Available: {list(self._dialogs.keys())}"
            )
        self._active_alias = alias
        return self

    def _get_dialog(self, alias: str = None) -> Dialog:
        """Resolve alias → Dialog, falling back to active dialog if alias is None."""
        if alias is not None:
            if alias not in self._dialogs:
                raise KeyError(
                    f"No dialog '{alias}' on agent '{self.name}'. "
                    f"Available: {list(self._dialogs.keys())}"
                )
            return self._dialogs[alias]
        if self._active_alias is None:
            raise RuntimeError(
                f"Agent '{self.name}' has no active dialog. "
                f"Call .open(alias) or .switch(alias) first."
            )
        return self._dialogs[self._active_alias]

    @property
    def current_dialog(self) -> Dialog:
        """The currently active dialog."""
        return self._get_dialog()

    @property
    def dialogs(self) -> Dict[str, Dialog]:
        """Read-only snapshot of all managed dialogs (alias → Dialog)."""
        return dict(self._dialogs)

    @property
    def active_alias(self) -> Optional[str]:
        return self._active_alias

    # ===================================================================
    # Messaging primitives — operate on active or specified dialog
    # ===================================================================

    def receive(
        self,
        text: str,
        alias: str = None,
        role: Roles = Roles.USER,
        name: str = 'user',
    ) -> Message:
        """Put a text message into the active (or specified) dialog."""
        return self._get_dialog(alias).put_text(text, name=name, role=role)

    def receive_prompt(
        self,
        prompt: Prompt,
        prompt_args: Optional[Dict[str, Any]] = None,
        alias: str = None,
        role: Roles = Roles.USER,
        name: str = 'user',
    ) -> Message:
        """Put a structured prompt message into the dialog."""
        return self._get_dialog(alias).put_prompt(
            prompt, prompt_args, name=name, role=role,
        )

    def receive_image(
        self,
        image,
        caption: str = None,
        alias: str = None,
        role: Roles = Roles.USER,
        name: str = 'user',
    ) -> Message:
        """Put an image message into the dialog."""
        return self._get_dialog(alias).put_image(
            image, caption=caption, name=name, role=role,
        )

    def respond(
        self,
        alias: str = None,
        metadata: Optional[Dict[str, Any]] = None,
        args: Optional[Dict[str, Any]] = None,
        parser_args: Optional[Dict[str, Any]] = None,
        return_session: bool = False,
    ) -> Union[Message, Tuple[Message, AgentCallSession]]:
        """
        High-level: run the agent call loop on a dialog, return the response.

        This is the recommended way to get a response.  For full diagnostics
        (call_state with retry info, model_args, etc.), use ``call()`` directly.

        Args:
            alias: the alias of the dialog to respond to.
            metadata: additional metadata for the call.
            args: additional arguments for the prompt.
            parser_args: arguments for the output parser.
            return_session: if True, return the entire AgentCallSession instead of just the message (use session.delivery to get the final message).
        """
        dialog = self._get_dialog(alias)
        session = self._call(dialog, metadata=metadata, args=args, parser_args=parser_args)
        if return_session:
            return session
        else:
            return session.delivery


    # ===================================================================
    # Core agent call loop
    # ===================================================================

    # it performs the "Agent Call"
    def _call(
        self,
        dialog: Dialog,  # it assumes the prompt is already loaded into the dialog as the top prompt by send_message
        metadata: Optional[Dict[str, Any]] = None,  # for tracking additional information, such as frontend replay info
        args: Optional[Dict[str, Any]] = None,  # for tracking additional information, such as frontend replay info
        parser_args: Optional[Dict[str, Any]] = None,
    ) -> AgentCallSession:
        """
        Executes the agent loop, handling LLM calls, tool execution, and interrupts.

        Args:
            dialog (Dialog): The current dialog state.
            metadata (Dict[str, Any], optional): Extra metadata for the call.
            args (Dict[str, Any], optional): Additional arguments for the prompt.
            parser_args (Dict[str, Any], optional): Arguments for the output parser.

        Returns:
            Tuple[Message, Dialog, List[FunctionCall]]: The final response message, the updated dialog, and a list of executed function calls.

        Raises:
            ValueError: If the agent fails to produce a valid response after retries.
        """
        session = AgentCallSession(
            agent_name=self.name,
            max_exception_retry=self.max_exception_retry,
            max_interrupt_steps=self.max_interrupt_steps,
            max_llm_recall=self.max_llm_recall,
        )
        metadata = dict(metadata) if metadata else {}
        args = dict(args) if args else {}
        parser_args = dict(parser_args) if parser_args else {}
        # Prompt: a function maps prompt args and dialog into the expected output 
        if dialog.top_prompt is None:
            dialog.top_prompt = self.system_prompt
        interrupts = []
        _max_steps = 100 if self.max_interrupt_steps == 0 else self.max_interrupt_steps + 1  # +1 for the final response
        if self.max_interrupt_steps == 0:
            logger.warning(
                "Agent '%s': max_interrupt_steps=0 means unlimited (up to 100 iterations). "
                "Set an explicit value to cap the loop.",
                self.name,
            )
        for i in range(_max_steps):
            working_dialog = dialog.fork() # make a copy of the dialog, truncate all exception handling dialogs
            if self.context_manager is not None:
                working_dialog = self.context_manager(working_dialog)
            while True: # ensure the response is no exception
                try:
                    _model_args = self.model_args.copy()
                    _model_args.update(args)

                    invoke_result = self.llm_invoker.call(
                        working_dialog,
                        self.model,
                        _model_args,
                        parser_args=parser_args,
                        responder=self.name,
                        metadata=metadata,
                        api_type=self.api_type,
                        stream_handler=self.stream_handler,
                    )
                    session.new_invoke_trace(invoke_result, i)
                    working_dialog.append(invoke_result.message) 
                    if invoke_result.has_errors:
                        raise AgentException(invoke_result.error_message)
                    else: 
                        break
                except AgentException as e: # handle the exception from the agent
                    if not session.reach_max_exception_retry:
                        session.exception(e, i)
                        working_dialog.put_prompt(
                            dialog.top_prompt.on_exception(session), 
                            {'error_message': str(e)}, 
                            name='exception'
                        )
                        continue
                    else:
                        raise e
                except Exception as e: # handle the exception from the LLM
                    # Simplified error handling for now
                    wait_time = random.random()*15+1
                    if U.is_openai_rate_limit_error(e): # for safe
                        time.sleep(wait_time)
                    else:
                        if not session.reach_max_llm_recall:
                            session.llm_recall(e, i)
                            time.sleep(1) # wait for a while before retrying
                            continue
                        else:
                            raise e

            dialog.append(invoke_result.message) # update the dialog state
            # now handle the interruption
            if invoke_result.message.is_function_call:
                _func_names = [func_call.name for func_call in invoke_result.message.function_calls]
                # handle the function call
                session.interrupt(invoke_result.message.function_calls, i)
                for function_call in invoke_result.message.function_calls:
                    if function_call.is_repeated(interrupts):
                        result_str = f'The function {function_call.name} with identical arguments {function_call.arguments} has been called earlier, please check the previous results and do not call it again. If you do not need to call more functions, just stop calling and provide the final response.'
                    else:
                        if function_call.name not in dialog.top_prompt.functions:
                            raise KeyError(f"Function '{function_call.name}' not registered on prompt '{dialog.top_prompt.path}'")
                        function = dialog.top_prompt.functions[function_call.name]
                        function_call = function(function_call)
                        result_str = function_call.result_str
                        interrupts.append(function_call)
                    dialog.put_prompt(
                        dialog.top_prompt.on_interrupt(session),
                        {'call_results': result_str},
                        role=Roles.TOOL,
                        name=function_call.name,
                        metadata={'tool_call_id': function_call.id},
                    )

                if session.reach_max_interrupt_steps:
                    dialog.put_prompt(
                        dialog.top_prompt.on_interrupt_final(session), 
                        role=Roles.USER, 
                        name=function_call.name
                    )
            else: # the response is not a function call, it is the final response
                session.success(invoke_result.message)
                return session
        session.failure()
        raise ValueError(f'Failed to call the agent: {session}')

current_dialog property

The currently active dialog.

dialogs property

Read-only snapshot of all managed dialogs (alias → Dialog).

close(alias)

Remove a dialog from this agent and return it.

Useful for archiving, handing off to another system, or just cleaning up. If the closed dialog was active, active becomes None.

Source code in lllm/core/agent.py 
123
124
125
126
127
128
129
130
131
132
133
def close(self, alias: str) -> Dialog:
    """
    Remove a dialog from this agent and return it.

    Useful for archiving, handing off to another system, or just
    cleaning up.  If the closed dialog was active, active becomes None.
    """
    dialog = self._dialogs.pop(alias)
    if self._active_alias == alias:
        self._active_alias = None
    return dialog

fork(alias, child_alias, last_n=0, first_k=1, switch=True)

Branch an existing dialog into a new child dialog.

The parent dialog's fork() handles all lineage bookkeeping (parent ↔ child links, split_point, ids). Agent just stores the child under child_alias and switches to it.

Parameters:

Name Type Description Default
alias str

the source dialog to fork from.

 required
child_alias str

the alias for the new child dialog.

 required
last_n int

if >0, drop the last n messages from the copy.

 0
first_k int

if >0, keep the first k messages from the copy. Only used when last_n is >0.

 1
switch bool

if True, switch to the new child dialog after forking.

 True

Raises:

Type Description
ValueError

if child_alias is already in use.
KeyError

if alias doesn't exist.
Source code in lllm/core/agent.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def fork(self, alias: str, child_alias: str, last_n: int = 0, first_k: int = 1, switch: bool = True) -> 'Agent':
    """
    Branch an existing dialog into a new child dialog.

    The parent dialog's ``fork()`` handles all lineage bookkeeping
    (parent ↔ child links, split_point, ids).  Agent just stores
    the child under ``child_alias`` and switches to it.

    Args:
        alias: the source dialog to fork from.
        child_alias: the alias for the new child dialog.
        last_n: if >0, drop the last n messages from the copy.
        first_k: if >0, keep the first k messages from the copy. Only used when last_n is >0.
        switch: if True, switch to the new child dialog after forking.

    Raises:
        ValueError: if ``child_alias`` is already in use.
        KeyError: if ``alias`` doesn't exist.
    """
    if child_alias in self._dialogs:
        raise ValueError(
            f"Dialog '{child_alias}' already exists on agent '{self.name}'."
        )
    parent = self._get_dialog(alias)
    child = parent.fork(last_n, first_k)
    self._dialogs[child_alias] = child
    if switch:
        self._active_alias = child_alias
    return self # for chaining

open(alias, prompt_args=None, session_name=None, switch=True)

Create a new dialog owned by this agent, keyed by alias.

Parameters:

Name Type Description Default
alias str

the alias for the new dialog.

 required
prompt_args Optional[Dict[str, Any]]

the arguments for the system prompt.

 None
session_name Optional[str]

the name of the session for logging and checkpointing.

 None
switch bool

if True, switch to the new dialog after opening. Default is True.

 True
Source code in lllm/core/agent.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def open(self, alias: str, prompt_args: Optional[Dict[str, Any]] = None, session_name: Optional[str] = None, switch: bool = True):
    """
    Create a new dialog owned by this agent, keyed by alias.

    Args:
        alias: the alias for the new dialog.
        prompt_args: the arguments for the system prompt.
        session_name: the name of the session for logging and checkpointing.
        switch: if True, switch to the new dialog after opening. Default is True.
    """
    if alias in self._dialogs:
        raise ValueError(
            f"Dialog '{alias}' already exists on agent '{self.name}'. "
            f"Use .fork('{alias}', ...) or .close('{alias}') first."
        )
    prompt_args = dict(prompt_args) if prompt_args else {}
    dialog = Dialog(
        session_name=session_name or f"{self.name}_{alias}",
        owner=self.name,
    )
    dialog.put_prompt(
        self.system_prompt, prompt_args,
        name='system', role=Roles.SYSTEM,
    )
    self._dialogs[alias] = dialog
    if switch:
        self._active_alias = alias
    return self # for chaining

receive(text, alias=None, role=Roles.USER, name='user')

Put a text message into the active (or specified) dialog.

Source code in lllm/core/agent.py 
184
185
186
187
188
189
190
191
192
def receive(
    self,
    text: str,
    alias: str = None,
    role: Roles = Roles.USER,
    name: str = 'user',
) -> Message:
    """Put a text message into the active (or specified) dialog."""
    return self._get_dialog(alias).put_text(text, name=name, role=role)

receive_image(image, caption=None, alias=None, role=Roles.USER, name='user')

Put an image message into the dialog.

Source code in lllm/core/agent.py 
207
208
209
210
211
212
213
214
215
216
217
218
def receive_image(
    self,
    image,
    caption: str = None,
    alias: str = None,
    role: Roles = Roles.USER,
    name: str = 'user',
) -> Message:
    """Put an image message into the dialog."""
    return self._get_dialog(alias).put_image(
        image, caption=caption, name=name, role=role,
    )

receive_prompt(prompt, prompt_args=None, alias=None, role=Roles.USER, name='user')

Put a structured prompt message into the dialog.

Source code in lllm/core/agent.py 
194
195
196
197
198
199
200
201
202
203
204
205
def receive_prompt(
    self,
    prompt: Prompt,
    prompt_args: Optional[Dict[str, Any]] = None,
    alias: str = None,
    role: Roles = Roles.USER,
    name: str = 'user',
) -> Message:
    """Put a structured prompt message into the dialog."""
    return self._get_dialog(alias).put_prompt(
        prompt, prompt_args, name=name, role=role,
    )

respond(alias=None, metadata=None, args=None, parser_args=None, return_session=False)

High-level: run the agent call loop on a dialog, return the response.

This is the recommended way to get a response. For full diagnostics (call_state with retry info, model_args, etc.), use call() directly.

Parameters:

Name Type Description Default
alias str

the alias of the dialog to respond to.

 None
metadata Optional[Dict[str, Any]]

additional metadata for the call.

 None
args Optional[Dict[str, Any]]

additional arguments for the prompt.

 None
parser_args Optional[Dict[str, Any]]

arguments for the output parser.

 None
return_session bool

if True, return the entire AgentCallSession instead of just the message (use session.delivery to get the final message).

 False
Source code in lllm/core/agent.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def respond(
    self,
    alias: str = None,
    metadata: Optional[Dict[str, Any]] = None,
    args: Optional[Dict[str, Any]] = None,
    parser_args: Optional[Dict[str, Any]] = None,
    return_session: bool = False,
) -> Union[Message, Tuple[Message, AgentCallSession]]:
    """
    High-level: run the agent call loop on a dialog, return the response.

    This is the recommended way to get a response.  For full diagnostics
    (call_state with retry info, model_args, etc.), use ``call()`` directly.

    Args:
        alias: the alias of the dialog to respond to.
        metadata: additional metadata for the call.
        args: additional arguments for the prompt.
        parser_args: arguments for the output parser.
        return_session: if True, return the entire AgentCallSession instead of just the message (use session.delivery to get the final message).
    """
    dialog = self._get_dialog(alias)
    session = self._call(dialog, metadata=metadata, args=args, parser_args=parser_args)
    if return_session:
        return session
    else:
        return session.delivery

switch(alias)

Set the active dialog by alias. Returns self for chaining.

Raises:

Type Description
KeyError

if alias doesn't exist.
Source code in lllm/core/agent.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def switch(self, alias: str) -> 'Agent':
    """
    Set the active dialog by alias.  Returns self for chaining.

    Raises:
        KeyError: if ``alias`` doesn't exist.
    """
    if alias not in self._dialogs:
        raise KeyError(
            f"No dialog '{alias}' on agent '{self.name}'. "
            f"Available: {list(self._dialogs.keys())}"
        )
    self._active_alias = alias
    return self

Tactic

lllm.core.tactic.Tactic

Bases: ABC

A Tactic is a local, functional unit of agentic behavior.

It defines HOW a group of agents solve a task — the "program" that wires callers (agents) to functions (prompts).

Config format (the dict passed to __init__)::

tactic_type: analytica
global:
    model_name: gpt-4o
    model_args:
        temperature: 0.1
agent_configs:
    - name: analyzer
      system_prompt_path: analytica/analyzer_system
      model_args:
          max_completion_tokens: 20000
    - name: synthesizer
      system_prompt_path: analytica/synthesizer_system

global provides defaults merged into each agent config. agent_configs is a list; each entry must have a name.

Source code in lllm/core/tactic.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
class Tactic(ABC):
    """
    A Tactic is a local, functional unit of agentic behavior.

    It defines HOW a group of agents solve a task — the "program" that
    wires callers (agents) to functions (prompts).

    **Config format** (the dict passed to ``__init__``)::

        tactic_type: analytica
        global:
            model_name: gpt-4o
            model_args:
                temperature: 0.1
        agent_configs:
            - name: analyzer
              system_prompt_path: analytica/analyzer_system
              model_args:
                  max_completion_tokens: 20000
            - name: synthesizer
              system_prompt_path: analytica/synthesizer_system

    ``global`` provides defaults merged into each agent config.
    ``agent_configs`` is a list; each entry must have a ``name``.
    """

    name: str = None
    agent_group: List[str] = None

    # -- Auto-registration ------------------------------------------------

    def __init_subclass__(cls, register: bool = True, runtime: Optional[Runtime] = None, **kwargs):
        super().__init_subclass__(**kwargs)
        if register and getattr(cls, "name", None):
            register_tactic_class(cls, runtime=runtime or get_default_runtime())

    # -- Construction -----------------------------------------------------

    def __init__(
        self,
        config: Dict[str, Any],
        log_store: Optional[LogStore] = None,
        runtime: Optional[Runtime] = None,
        tactic_path: Optional[str] = None,
    ):
        self._runtime = runtime or get_default_runtime()
        self._sub_tactics: Dict[str, Tactic] = {}

        self.config = config
        self._log_store: Optional[LogStore] = log_store
        self._log_store_warned: bool = False
        # Absolute qualified key in the runtime, e.g. "my_pkg.tactics:folder/researcher".
        # If not supplied by build_tactic, resolved lazily on first use.
        self._tactic_path: Optional[str] = tactic_path
        self.llm_invoker = build_invoker(config)

        # Parse agent specs from the new config format
        assert self.agent_group is not None, (
            f"agent_group not set for tactic '{self.name}'"
        )
        self._agent_specs = parse_agent_configs(
            config, self.agent_group, self.name
        )

        self._max_workers: int = config.get("max_workers", 4)

        # Per-call state — set by _execute on the copy, created here for convenience and checking
        self.agents: Dict[str, Union[Agent, _TrackedAgent]] = self._create_fresh_agents()
        self._session: Optional[TacticCallSession] = None

    # -- Sub-tactic composition -------------------------------------------

    def __setattr__(self, name: str, value: Any) -> None:
        if isinstance(value, Tactic) and name not in ("_sub_tactics",):
            if hasattr(self, "_sub_tactics"):
                self._sub_tactics[name] = value
        super().__setattr__(name, value)

    @property
    def sub_tactics(self) -> Dict[str, "Tactic"]:
        return dict(self._sub_tactics)

    # -- Fresh agent creation (per-call) ----------------------------------

    def _create_fresh_agents(self) -> Dict[str, Agent]:
        return {
            agent_name: spec.build(self._runtime, self.llm_invoker)
            for agent_name, spec in self._agent_specs.items()
        }

    # -- Core execution ---------------------------------------------------

    @abstractmethod
    def call(self, task: Union[str, BaseModel], **kwargs) -> Union[str, BaseModel]:
        pass

    def _execute(
        self,
        task: Union[str, BaseModel],
        session_name: Optional[str] = None,
        tags: Optional[Dict[str, str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        return_session: bool = False,
        **kwargs,
    ) -> Union[str, BaseModel, TacticCallSession]:
        if session_name is None:
            task_str = task if isinstance(task, str) else task.model_dump_json()
            task_hash = hashlib.md5(task_str.encode()).hexdigest()[:8]
            session_name = (
                f"{self.name}_{task_hash}"
                f"_{dt.datetime.now().strftime('%Y%m%d_%H%M%S')}"
            )

        ctx = copy.copy(self)
        ctx._sub_tactics = dict(self._sub_tactics)

        # Resolve stable tactic ID lazily (once, then cached).
        # Format: "{package_name}::{tactic_name}", e.g. "my_pkg::researcher".
        # Independent of file layout, "under" prefixes, and aliases.
        if self._tactic_path is None:
            try:
                node = self._runtime.get_node(self.name, resource_type="tactic")
                self._tactic_path = _stable_tactic_id(node.namespace, self.name)
            except (KeyError, AttributeError):
                self._tactic_path = self.name  # not in package system

        session = TacticCallSession(tactic_name=self.name, tactic_path=self._tactic_path)
        session.state = "running"
        ctx._session = session

        raw_agents = ctx._create_fresh_agents()
        ctx.agents = {
            n: _TrackedAgent(agent, session, n)
            for n, agent in raw_agents.items()
        }

        logger.info("Tactic '%s' started — session_name=%s", self.name, session_name)
        try:
            result = ctx.call(task, **kwargs)
            session.success(result)
            logger.info(
                "Tactic '%s' completed — cost=%s agent_calls=%d",
                self.name,
                session.total_cost.cost,
                session.agent_call_count,
            )
        except Exception as e:
            session.failure(e)
            logger.error(
                "Tactic '%s' failed: %s",
                self.name, e, exc_info=True,
            )
            raise
        finally:
            if self._log_store is not None:
                try:
                    saved_id = self._log_store.save_session(
                        session, tags=tags, metadata=metadata
                    )
                    logger.debug("Session persisted — id=%s", saved_id)
                except Exception:
                    logger.warning(
                        "LogStore failed to save session for tactic '%s'",
                        self.name, exc_info=True,
                    )
            elif not self._log_store_warned:
                self._log_store_warned = True
                warnings.warn(
                    f"No LogStore configured for tactic '{self.name}'. "
                    "Session data will not be persisted. "
                    "Pass a LogStore instance via the log_store parameter.",
                    UserWarning,
                    stacklevel=3,
                )

        return session if return_session else result

    def __call__(self, task, session_name=None, tags=None, metadata=None,  return_session=False, **kwargs):
        return self._execute(task, session_name, tags=tags, metadata=metadata, return_session=return_session, **kwargs)

    async def acall(self, task, tags=None, metadata=None, return_session=False, **kwargs):
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            lambda: self._execute(task, tags=tags, metadata=metadata, return_session=return_session, **kwargs),
        )

    def bcall(self, tasks, max_workers=None, fail_fast=True, tags=None, metadata=None, return_sessions=False, **kwargs):
        workers = max_workers or self._max_workers
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
            futures = [
                pool.submit(self._execute, t, None, tags, metadata, return_session=return_sessions, **kwargs)
                for t in tasks
            ]
            if fail_fast:
                return [f.result() for f in futures]
            results = []
            for f in futures:
                try:
                    results.append(f.result())
                except Exception as e:
                    results.append(e)
            return results

    async def ccall(self, tasks, max_workers=None, tags=None, metadata=None, return_sessions=False, **kwargs):
        workers = max_workers or self._max_workers
        loop = asyncio.get_running_loop()
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
            def _run(idx, t):
                return idx, self._execute(t, tags=tags, metadata=metadata, return_session=return_sessions, **kwargs)
            futures = [loop.run_in_executor(pool, _run, i, t) for i, t in enumerate(tasks)]
            for coro in asyncio.as_completed(futures):
                idx, result = await coro
                yield idx, result

    # -- Quick constructor ------------------------------------------------

    @classmethod
    def quick(cls, 
        query: Optional[str] = None, 
        system_prompt: Optional[Union[str, Prompt]] = "You are a helpful assistant.", 
        model: str = "gpt-4o", 
        return_agent: bool = False,
        **model_args: Any
    ) -> Union[Message, Agent, Tuple[Message, Agent]]:
        """
        Quick constructor for a single-agent chat.

        Args:
            query: The query to send to the agent.
            system_prompt: The system prompt to use for the agent.
            model: The model to use for the agent.
            return_agent: If True, return the agent in addition to the response. If no query is provided, return the agent.
            **model_args: Additional model arguments.

        Returns:
            If return_agent is False:
                Message: The response from the agent.
            If return_agent is True:
                Tuple[Message, Agent]: The response from the agent and the agent.
            If return_agent is True and query is not provided:
                Agent: The agent.
        """
        if isinstance(system_prompt, str):
            prompt = Prompt(path="_quick/system", prompt=system_prompt)
        else:
            prompt = system_prompt
        invoker = build_invoker({"invoker": "litellm"})
        agent = Agent(
            name="assistant",
            system_prompt=prompt,
            model=model,
            llm_invoker=invoker,
            model_args=model_args,
        )
        if query is not None:
            agent.open("chat")
            agent.receive(query)
            response = agent.respond()
            if return_agent:
                return response, agent
            else:
                return response
        else:
            return agent

    def __repr__(self) -> str:
        parts = [f"Tactic(name={self.name!r}"]
        parts.append(f"agents={list(self._agent_specs.keys())}")
        if self._sub_tactics:
            parts.append(f"sub_tactics={list(self._sub_tactics.keys())}")
        return ", ".join(parts) + ")"

quick(query=None, system_prompt='You are a helpful assistant.', model='gpt-4o', return_agent=False, **model_args) classmethod

Quick constructor for a single-agent chat.

Parameters:

Name Type Description Default
query Optional[str]

The query to send to the agent.

 None
system_prompt Optional[Union[str, Prompt]]

The system prompt to use for the agent.

 'You are a helpful assistant.'
model str

The model to use for the agent.

 'gpt-4o'
return_agent bool

If True, return the agent in addition to the response. If no query is provided, return the agent.

 False
**model_args Any

Additional model arguments.

 {}

Returns:

Type Description
Union[Message, Agent, Tuple[Message, Agent]]

If return_agent is False: Message: The response from the agent.
Union[Message, Agent, Tuple[Message, Agent]]

If return_agent is True: Tuple[Message, Agent]: The response from the agent and the agent.
Union[Message, Agent, Tuple[Message, Agent]]

If return_agent is True and query is not provided: Agent: The agent.
Source code in lllm/core/tactic.py 
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
@classmethod
def quick(cls, 
    query: Optional[str] = None, 
    system_prompt: Optional[Union[str, Prompt]] = "You are a helpful assistant.", 
    model: str = "gpt-4o", 
    return_agent: bool = False,
    **model_args: Any
) -> Union[Message, Agent, Tuple[Message, Agent]]:
    """
    Quick constructor for a single-agent chat.

    Args:
        query: The query to send to the agent.
        system_prompt: The system prompt to use for the agent.
        model: The model to use for the agent.
        return_agent: If True, return the agent in addition to the response. If no query is provided, return the agent.
        **model_args: Additional model arguments.

    Returns:
        If return_agent is False:
            Message: The response from the agent.
        If return_agent is True:
            Tuple[Message, Agent]: The response from the agent and the agent.
        If return_agent is True and query is not provided:
            Agent: The agent.
    """
    if isinstance(system_prompt, str):
        prompt = Prompt(path="_quick/system", prompt=system_prompt)
    else:
        prompt = system_prompt
    invoker = build_invoker({"invoker": "litellm"})
    agent = Agent(
        name="assistant",
        system_prompt=prompt,
        model=model,
        llm_invoker=invoker,
        model_args=model_args,
    )
    if query is not None:
        agent.open("chat")
        agent.receive(query)
        response = agent.respond()
        if return_agent:
            return response, agent
        else:
            return response
    else:
        return agent

Dialog

lllm.core.dialog.Dialog dataclass

An append-only message sequence owned by a single agent.

The agent that creates a dialog seeds it with its system prompt, and that ownership is recorded on the tree_node. Other participants (user, tools, forwarded messages from other agents) append via put_* helpers, but the system-level identity of the dialog never changes.

Tree structure is maintained by a :class:DialogTreeNode owned by each Dialog. fork() creates a child Dialog whose tree_node is automatically linked to the parent's tree_node — callers (including Agent) never need to wire lineage manually.

Source code in lllm/core/dialog.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
@dataclass
class Dialog:
    """
    An append-only message sequence owned by a single agent. 

    The agent that creates a dialog seeds it with its system prompt, and that
    ownership is recorded on the ``tree_node``.  Other participants (user,
    tools, forwarded messages from other agents) append via ``put_*`` helpers,
    but the system-level identity of the dialog never changes.

    **Tree structure** is maintained by a :class:`DialogTreeNode` owned by
    each Dialog.  ``fork()`` creates a child Dialog whose tree_node is
    automatically linked to the parent's tree_node — callers (including
    Agent) never need to wire lineage manually.
    """
    session_name: str = None
    top_prompt: Optional[Prompt] = None
    runtime: Optional[Runtime] = None
    owner: Optional[str] = None

    # Message storage (append-only)
    _messages: List[Message] = field(default_factory=list)

    # Tree structure — each Dialog has exactly one node
    tree_node: DialogTreeNode = field(default=None)

    # Live Dialog-level parent/children refs (parallel to tree_node's node-level refs)
    _parent_dialog: Optional['Dialog'] = field(default=None, repr=False)
    _children_dialogs: List['Dialog'] = field(default_factory=list, repr=False)


    def __post_init__(self):
        if self.tree_node is None:
            self.tree_node = DialogTreeNode(owner=self.owner)
        if self.session_name is None:
            self.session_name = dt.datetime.now().strftime('%Y%m%d_%H%M%S') + '_' + str(uuid.uuid4())[:6]
        self.runtime = self.runtime or get_default_runtime()

    # -- Convenience proxies to tree_node ---------------------------------

    @property
    def dialog_id(self) -> str:
        return self.tree_node.dialog_id

    @property
    def parent(self) -> Optional['Dialog']:
        """Live reference to parent Dialog (None for root dialogs)."""
        return self._parent_dialog

    @property
    def children(self) -> List['Dialog']:
        """Live references to child Dialogs forked from this one."""
        return list(self._children_dialogs)

    @property
    def is_root(self) -> bool:
        return self.tree_node.is_root

    @property
    def depth(self) -> int:
        return self.tree_node.depth

    # -- Message access ---------------------------------------------------

    def append(self, message: Message):
        message.metadata['dialog_id'] = self.dialog_id
        self._messages.append(message)

    def to_dict(self):
        return {
            'messages': [message.to_dict() for message in self._messages],
            'session_name': self.session_name,
            'owner': self.owner,
            'tree_node': self.tree_node.to_dict(),
            'top_prompt_path': (
                getattr(self.top_prompt, '_qualified_key', None) or self.top_prompt.path
            ) if self.top_prompt is not None else None,
        }

    @classmethod
    def from_dict(cls, d: dict, runtime: Runtime = None):
        top_prompt_path = d.get('top_prompt_path')
        runtime = runtime or get_default_runtime()
        top_prompt = None
        if top_prompt_path is not None:
            try:
                top_prompt = runtime.get_prompt(top_prompt_path)
            except KeyError:
                logger.warning("Prompt '%s' not found in runtime during Dialog.from_dict", top_prompt_path)
        tree_node_data = d.get('tree_node')
        tree_node = DialogTreeNode.from_dict(tree_node_data) if tree_node_data else None
        return cls(
            _messages=[Message.from_dict(message) for message in d['messages']],
            session_name=d['session_name'],
            owner=d.get('owner'),
            top_prompt=top_prompt,
            runtime=runtime,
            tree_node=tree_node,
        )

    @property
    def messages(self):
        return self._messages

    # -----------------------------------------------------------------------
    # Message Operations, you can only put or fork, dialog is immutable and monotonic.
    # If you want to modify the dialog, you can use ContextManager to dynamically edit it on the fly.
    # -----------------------------------------------------------------------

    # Static/stateless puts

    def put_image(
        self,
        image: Union[str, Path, Any],
        caption: str = None,
        name: str = 'user',
        metadata: Optional[Dict[str, Any]] = None,
        role: Roles = Roles.USER,
    ) -> Message:
        """
        Expects:
        - image: a base64 encoded string, a Path object or string path, or a PIL Image object
        """
        # Resolve string to Path if it looks like a file path
        if isinstance(image, str):
            try:
                p = Path(image)
                if p.exists():
                    image = p
            except (OSError, ValueError):
                pass  # not a path; treat as base64 below
        if isinstance(image, Path):
            with image.open('rb') as f:
                image_base64 = base64.b64encode(f.read()).decode('utf-8')
        elif _is_pil_image(image):
            buf = io.BytesIO()
            fmt = image.format or 'PNG'
            image.save(buf, format=fmt)
            image_base64 = base64.b64encode(buf.getvalue()).decode('utf-8')
        elif isinstance(image, str):
            # Validate that the base64 string decodes to a known image format
            _IMAGE_MAGIC = (
                b'\x89PNG',       # PNG
                b'\xff\xd8\xff',  # JPEG
                b'RIFF',          # WebP (RIFF....WEBP)
                b'GIF8',          # GIF
            )
            try:
                raw = base64.b64decode(image)
            except Exception:
                raise ValueError(f'Invalid base64 encoded image string')
            if not any(raw.startswith(m) for m in _IMAGE_MAGIC):
                raise ValueError(
                    f'Base64 string does not appear to be a supported image '
                    f'(PNG/JPEG/WebP/GIF); got header {raw[:8]!r}'
                )
            image_base64 = image
        else:
            raise ValueError(f'Invalid image type: {type(image)}')
        payload = dict(metadata) if metadata else {}
        if caption is not None:
            payload['caption'] = caption
        message = Message(
            role=role,
            content=image_base64,
            name=name,
            modality=Modalities.IMAGE,
            metadata=payload,
        )
        self.append(message)
        return message

    def put_text(
        self,
        text: str,
        name: str = 'user',
        metadata: Optional[Dict[str, Any]] = None,
        role: Roles = Roles.USER,
    ) -> Message:
        from lllm.core.prompt import Prompt  # lazy import to avoid circular dependency
        metadata = dict(metadata) if metadata else {}
        # create a temporary prompt for the text to reset parsers and other state
        prompt = Prompt(path='__temp_prompt_'+str(uuid.uuid4())[:6], prompt=text)
        message = Message(
            role=role,
            content=text,
            name=name,
            modality=Modalities.TEXT,
            metadata=metadata
        )
        self.append(message)
        self.top_prompt = prompt
        return message

    # Stateful put, only prompt can be stateful, other messages are stateless

    def put_prompt(
        self,
        prompt: Prompt | str,
        prompt_args: Optional[Dict[str, Any]] = None,
        name: str = 'user',  # or 'user', etc.
        metadata: Optional[Dict[str, Any]] = None,
        role: Roles = Roles.USER,
    ) -> Message:
        if isinstance(prompt, str):
            prompt = self.runtime.get_prompt(prompt)
        prompt_args = dict(prompt_args) if prompt_args else {}
        metadata = dict(metadata) if metadata else {}

        content = prompt(**prompt_args)
        message = Message(
            role=role,
            content=content,
            name=name,
            modality=Modalities.TEXT,
            metadata=metadata
        )
        self.append(message)
        self.top_prompt = prompt
        return message

    # -----------------------------------------------------------------------
    # Forking — the only way to branch a dialog
    # -----------------------------------------------------------------------

    def fork(self, last_n: int = 0, first_k: int = 1) -> 'Dialog':
        """
        Create a child dialog branching from this one.

        Args:
            last_n: if >0, only preserve the last n messages from the parent dialog
                        (useful for retrying from an earlier point).
                        The first system message is always preserved.
            first_k: if >0, ensure the first k messages from the parent dialog
                        (useful for preserving the system prompt message). 
                        Should always be >=1 to at least preserve the system prompt message.

        The fork automatically:
        - Deep-copies the message prefix into the child.
        - Creates a child DialogTreeNode linked to this dialog's tree_node.
        - Records split_point on the child's tree_node.
        - Wires live Dialog-level parent/children refs.
        - Inherits session_name, top_prompt, runtime, owner.

        Returns:
            The new child Dialog.
        """
        if last_n >= len(self._messages):
            last_n = 0
        if last_n > 0:
            tail_start = len(self._messages) - last_n
            # Clamp first_k so it doesn't overlap with the tail slice
            first_k = min(first_k, tail_start) if first_k > 0 else 0
            _messages = self._messages[:first_k] + self._messages[tail_start:]
        else:
            _messages = self._messages
        split_point = len(_messages)

        # Build the child's tree node (not yet linked)
        child_tree_node = DialogTreeNode(
            owner=self.owner,
            split_point=split_point,
            last_n=last_n,
            first_k=first_k,
        )

        child = Dialog(
            _messages=[copy.deepcopy(m) for m in _messages],
            session_name=self.session_name,
            top_prompt=self.top_prompt,
            runtime=self.runtime,
            owner=self.owner,
            tree_node=child_tree_node,
        )

        # Wire tree_node parent ↔ child (sets parent_id, live refs, children_ids)
        self.tree_node.add_child(child.tree_node)

        # Wire Dialog-level live refs
        child._parent_dialog = self
        self._children_dialogs.append(child)

        return child

    # -----------------------------------------------------------------------
    # Display
    # -----------------------------------------------------------------------

    def overview(self, remove_tail: bool = False, max_length: int = 100, 
                 stream = None, divider: bool = False):
        _overview = ''
        for idx, message in enumerate(self.messages):
            if remove_tail and idx == len(self.messages)-1:
                break
            content_preview = str(message.content)[:max_length] + '...' if len(str(message.content)) > max_length else str(message.content)
            _overview += f'[{idx}. {message.name} ({message.role.msg_value})]: {content_preview}\n\n'

        _overview = _overview.strip()
        cost = self.tail.cost if self.messages else InvokeCost()
        if stream is not None:
            if divider:
                stream.divider()
            stream.write(U.html_collapse(f'Context overview', _overview), unsafe_allow_html=True)
            stream.write(str(cost))
        return _overview

    def tree_overview(self, indent: int = 0) -> str:
        """
        Recursively print the dialog tree structure from this node.
        Useful for debugging multi-fork scenarios.
        """
        prefix = '  ' * indent
        branch = '└─ ' if indent > 0 else ''
        line = (
            f"{prefix}{branch}[{self.dialog_id[:8]}] "
            f"owner={self.owner} msgs={len(self._messages)} "
            f"split@{self.tree_node.split_point}"
        )
        if self.tree_node.last_n is not None and self.tree_node.last_n > 0:
            line += f" (last_n={self.tree_node.last_n}, first_k={self.tree_node.first_k})"
        lines = [line]
        for child in self._children_dialogs:
            lines.append(child.tree_overview(indent + 1))
        return '\n'.join(lines)

    @property
    def tail(self): # last message in the dialog, use it to get last response from the LLM
        return self._messages[-1] if self._messages else None

    @property
    def head(self): # usually the system prompt message
        return self._messages[0] if self._messages else None

    @property
    def cost(self) -> InvokeCost:
        costs = [message.cost for message in self._messages]
        return InvokeCost(
            prompt_tokens=sum(c.prompt_tokens for c in costs),
            completion_tokens=sum(c.completion_tokens for c in costs),
            total_tokens=sum(c.total_tokens for c in costs),
            cached_prompt_tokens=sum(c.cached_prompt_tokens for c in costs),
            reasoning_tokens=sum(c.reasoning_tokens for c in costs),
            audio_prompt_tokens=sum(c.audio_prompt_tokens for c in costs),
            audio_completion_tokens=sum(c.audio_completion_tokens for c in costs),

            # We don't aggregate token rates for the whole dialog
            input_cost_per_token=0.0,
            output_cost_per_token=0.0,
            cache_read_input_token_cost=0.0,

            # Aggregate absolute dollar values
            prompt_cost=sum(c.prompt_cost for c in costs),
            completion_cost=sum(c.completion_cost for c in costs),
            cost=sum(c.cost for c in costs)
        )

children property

Live references to child Dialogs forked from this one.

parent property

Live reference to parent Dialog (None for root dialogs).

fork(last_n=0, first_k=1)

Create a child dialog branching from this one.

Parameters:

Name Type Description Default
last_n int

if >0, only preserve the last n messages from the parent dialog (useful for retrying from an earlier point). The first system message is always preserved.

 0
first_k int

if >0, ensure the first k messages from the parent dialog (useful for preserving the system prompt message). Should always be >=1 to at least preserve the system prompt message.

 1

The fork automatically: - Deep-copies the message prefix into the child. - Creates a child DialogTreeNode linked to this dialog's tree_node. - Records split_point on the child's tree_node. - Wires live Dialog-level parent/children refs. - Inherits session_name, top_prompt, runtime, owner.

Returns:

Type Description
'Dialog'

The new child Dialog.
Source code in lllm/core/dialog.py 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
def fork(self, last_n: int = 0, first_k: int = 1) -> 'Dialog':
    """
    Create a child dialog branching from this one.

    Args:
        last_n: if >0, only preserve the last n messages from the parent dialog
                    (useful for retrying from an earlier point).
                    The first system message is always preserved.
        first_k: if >0, ensure the first k messages from the parent dialog
                    (useful for preserving the system prompt message). 
                    Should always be >=1 to at least preserve the system prompt message.

    The fork automatically:
    - Deep-copies the message prefix into the child.
    - Creates a child DialogTreeNode linked to this dialog's tree_node.
    - Records split_point on the child's tree_node.
    - Wires live Dialog-level parent/children refs.
    - Inherits session_name, top_prompt, runtime, owner.

    Returns:
        The new child Dialog.
    """
    if last_n >= len(self._messages):
        last_n = 0
    if last_n > 0:
        tail_start = len(self._messages) - last_n
        # Clamp first_k so it doesn't overlap with the tail slice
        first_k = min(first_k, tail_start) if first_k > 0 else 0
        _messages = self._messages[:first_k] + self._messages[tail_start:]
    else:
        _messages = self._messages
    split_point = len(_messages)

    # Build the child's tree node (not yet linked)
    child_tree_node = DialogTreeNode(
        owner=self.owner,
        split_point=split_point,
        last_n=last_n,
        first_k=first_k,
    )

    child = Dialog(
        _messages=[copy.deepcopy(m) for m in _messages],
        session_name=self.session_name,
        top_prompt=self.top_prompt,
        runtime=self.runtime,
        owner=self.owner,
        tree_node=child_tree_node,
    )

    # Wire tree_node parent ↔ child (sets parent_id, live refs, children_ids)
    self.tree_node.add_child(child.tree_node)

    # Wire Dialog-level live refs
    child._parent_dialog = self
    self._children_dialogs.append(child)

    return child

put_image(image, caption=None, name='user', metadata=None, role=Roles.USER)

Expects: - image: a base64 encoded string, a Path object or string path, or a PIL Image object

Source code in lllm/core/dialog.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def put_image(
    self,
    image: Union[str, Path, Any],
    caption: str = None,
    name: str = 'user',
    metadata: Optional[Dict[str, Any]] = None,
    role: Roles = Roles.USER,
) -> Message:
    """
    Expects:
    - image: a base64 encoded string, a Path object or string path, or a PIL Image object
    """
    # Resolve string to Path if it looks like a file path
    if isinstance(image, str):
        try:
            p = Path(image)
            if p.exists():
                image = p
        except (OSError, ValueError):
            pass  # not a path; treat as base64 below
    if isinstance(image, Path):
        with image.open('rb') as f:
            image_base64 = base64.b64encode(f.read()).decode('utf-8')
    elif _is_pil_image(image):
        buf = io.BytesIO()
        fmt = image.format or 'PNG'
        image.save(buf, format=fmt)
        image_base64 = base64.b64encode(buf.getvalue()).decode('utf-8')
    elif isinstance(image, str):
        # Validate that the base64 string decodes to a known image format
        _IMAGE_MAGIC = (
            b'\x89PNG',       # PNG
            b'\xff\xd8\xff',  # JPEG
            b'RIFF',          # WebP (RIFF....WEBP)
            b'GIF8',          # GIF
        )
        try:
            raw = base64.b64decode(image)
        except Exception:
            raise ValueError(f'Invalid base64 encoded image string')
        if not any(raw.startswith(m) for m in _IMAGE_MAGIC):
            raise ValueError(
                f'Base64 string does not appear to be a supported image '
                f'(PNG/JPEG/WebP/GIF); got header {raw[:8]!r}'
            )
        image_base64 = image
    else:
        raise ValueError(f'Invalid image type: {type(image)}')
    payload = dict(metadata) if metadata else {}
    if caption is not None:
        payload['caption'] = caption
    message = Message(
        role=role,
        content=image_base64,
        name=name,
        modality=Modalities.IMAGE,
        metadata=payload,
    )
    self.append(message)
    return message

tree_overview(indent=0)

Recursively print the dialog tree structure from this node. Useful for debugging multi-fork scenarios.

Source code in lllm/core/dialog.py 
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
def tree_overview(self, indent: int = 0) -> str:
    """
    Recursively print the dialog tree structure from this node.
    Useful for debugging multi-fork scenarios.
    """
    prefix = '  ' * indent
    branch = '└─ ' if indent > 0 else ''
    line = (
        f"{prefix}{branch}[{self.dialog_id[:8]}] "
        f"owner={self.owner} msgs={len(self._messages)} "
        f"split@{self.tree_node.split_point}"
    )
    if self.tree_node.last_n is not None and self.tree_node.last_n > 0:
        line += f" (last_n={self.tree_node.last_n}, first_k={self.tree_node.first_k})"
    lines = [line]
    for child in self._children_dialogs:
        lines.append(child.tree_overview(indent + 1))
    return '\n'.join(lines)

Message

lllm.core.dialog.Message

Bases: BaseModel

Source code in lllm/core/dialog.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class Message(BaseModel):
    role: Roles
    content: Union[str, List[Dict[str, Any]]] # Content can be string or list of content parts (for images)
    name: str # name of the sender
    function_calls: List[FunctionCall] = Field(default_factory=list)
    modality: Modalities = Modalities.TEXT
    logprobs: List[TokenLogprob] = Field(default_factory=list)
    parsed: Dict[str, Any] = Field(default_factory=dict)
    model: Optional[str] = None
    usage: Dict[str, Any] = Field(default_factory=dict) # 
    metadata: Dict[str, Any] = Field(default_factory=dict) 
    api_type: APITypes = APITypes.COMPLETION

    vectors: List[float] = Field(default_factory=list) # place holder for embedding vectors of the message, can be used for training, similarity search, etc. Need special invoker to support this.
    model_config = ConfigDict(arbitrary_types_allowed=True)

    @property
    def sanitized_name(self):
        return _sanitize_name(self.name)

    @field_validator("logprobs", mode="before")
    @classmethod
    def _coerce_logprobs(cls, value):
        if not value:
            return []
        normalized: List[TokenLogprob] = []
        for entry in value:
            if isinstance(entry, TokenLogprob):
                normalized.append(entry)
                continue
            if isinstance(entry, dict):
                normalized.append(TokenLogprob(**entry))
                continue
            if isinstance(entry, (int, float)):
                normalized.append(TokenLogprob(logprob=float(entry)))
                continue
            normalized.append(TokenLogprob(token=str(entry)))
        return normalized

    @property
    def cost(self) -> InvokeCost:
        if not self.usage:
            return InvokeCost()

        p_tokens = self.usage.get('prompt_tokens', 0)
        c_tokens = self.usage.get('completion_tokens', 0)
        t_tokens = self.usage.get('total_tokens', p_tokens + c_tokens)
        p_details = self.usage.get('prompt_tokens_details', {}) or {}
        c_details = self.usage.get('completion_tokens_details', {}) or {}

        return InvokeCost(
            prompt_tokens=p_tokens,
            completion_tokens=c_tokens,
            total_tokens=t_tokens,
            cached_prompt_tokens=p_details.get('cached_tokens') or 0,
            audio_prompt_tokens=p_details.get('audio_tokens') or 0,
            reasoning_tokens=c_details.get('reasoning_tokens') or 0,
            audio_completion_tokens=c_details.get('audio_tokens') or 0,
            # Dollar costs
            input_cost_per_token=self.usage.get("input_cost_per_token", 0.0),
            output_cost_per_token=self.usage.get("output_cost_per_token", 0.0),
            cache_read_input_token_cost=self.usage.get("cache_read_input_token_cost", 0.0),
            prompt_cost=self.usage.get("prompt_cost", 0.0),
            completion_cost=self.usage.get("completion_cost", 0.0),
            cost=self.usage.get("response_cost", 0.0)
        )

    @property
    def is_function_call(self) -> bool:
        return len(self.function_calls) > 0

    def to_dict(self):
        return self.model_dump()

    @classmethod
    def from_dict(cls, d: dict):
        return cls(**d)

Prompt

lllm.core.prompt.Prompt

Bases: BaseModel

A Prompt is a complete behaviour definition for one agent turn.

It bundles four concerns:

  1. Template — the text to send to the LLM, with {variable} placeholders rendered via str.format (or a custom renderer).
  2. Output contract — an :class:OutputSpec describing how to parse and validate the LLM's response.
  3. Tools — the :class:Function and :class:MCP objects available during this turn.
  4. Handlers — template strings (or full Prompts) that define how to recover from exceptions and how to feed tool results back.

Provider-specific features (web search, computer use, citations, …) live in the generic capabilities dict so that new features never require schema changes on Prompt.

Notes

The __call__ method uses str.format by default, so literal braces in the template must be doubled: {{ and }}.

Source code in lllm/core/prompt.py 
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
class Prompt(BaseModel):
    """
    A Prompt is a complete behaviour definition for one agent turn.

    It bundles four concerns:

    1. **Template** — the text to send to the LLM, with ``{variable}``
       placeholders rendered via ``str.format`` (or a custom ``renderer``).
    2. **Output contract** — an :class:`OutputSpec` describing how to parse
       and validate the LLM's response.
    3. **Tools** — the :class:`Function` and :class:`MCP` objects available
       during this turn.
    4. **Handlers** — template strings (or full Prompts) that define how to
       recover from exceptions and how to feed tool results back.

    Provider-specific features (web search, computer use, citations, …) live
    in the generic ``capabilities`` dict so that new features never require
    schema changes on Prompt.

    Notes
    -----
    The ``__call__`` method uses ``str.format`` by default, so literal braces
    in the template must be doubled: ``{{`` and ``}}``.
    """
    path: str
    prompt: str
    metadata: Dict[str, Any] = Field(default_factory=dict) # record additional info, like version, etc.

    # -- Output contract --------------------------------------------------
    parser: Optional[BaseParser] = None
    format: Optional[Any] = None # Structured output (Pydantic model class or JSON schema dict)

    # -- Tools ------------------------------------------------------------
    function_list: List[Function] = Field(default_factory=list)
    mcp_servers_list: List[MCP] = Field(default_factory=list)

    # Provider-specific capabilities or args (like allow_web_search, computer_use_config, etc.)
    addon_args: Dict[str, Any] = Field(default_factory=dict)

    # -- Handlers ---------------------------------------------------------
    handler: BaseHandler = Field(default_factory=DefaultSimpleHandler)

    # -- Rendering --------------------------------------------------------
    renderer: BaseRenderer = Field(default_factory=StringFormatterRenderer)

    model_config = ConfigDict(arbitrary_types_allowed=True)

    # -- Internal (populated in model_post_init) --------------------------
    _functions: Dict[str, Function] = PrivateAttr(default_factory=dict)
    _mcp_servers: Dict[str, MCP] = PrivateAttr(default_factory=dict)
    _template_vars: set = PrivateAttr(default_factory=set)

    def model_post_init(self, __context):
        self._functions = {f.name: f for f in self.function_list}
        self._mcp_servers = {m.server_label: m for m in self.mcp_servers_list}

        _parser = string.Formatter()
        self._template_vars = {
            field_name.split('.')[0].split('[')[0]
            for _, field_name, _, _ in _parser.parse(self.prompt)
            if field_name is not None
        }

    @property
    def functions(self) -> Dict[str, Function]:
        return self._functions

    @property
    def mcp_servers(self) -> Dict[str, MCP]:
        return self._mcp_servers


    @property
    def template_vars(self) -> set[str]:
        """Variable names required by this template (e.g. {topic} → 'topic')."""
        return self._template_vars

    def validate_args(self, prompt_args: dict[str, Any]) -> list[str]:
        """Return list of missing required template variables for the prompt."""
        return [v for v in self.template_vars if v not in prompt_args]


    # -- Rendering --------------------------------------------------------

    def __call__(self, **kwargs: Any) -> str:
        """Render the template with the given variables."""
        if not self.template_vars:  # empty set is falsy
            return self.prompt
        missing_args = self.validate_args(kwargs)
        if missing_args:
            raise ValueError(f"Missing required template variables: {missing_args} for prompt {self.path}, please provide: {self.template_vars}")
        return self.renderer.render(self.prompt, **kwargs)


    def parse(self, content: str, **runtime_args: Any) -> Dict[str, Any]:
        if self.parser is not None:
            parsed = self.parser.parse(content, **runtime_args)
            if 'raw' not in parsed:
                parsed['raw'] = content
            return parsed
        return {
            'raw': content,
        }

    # -- Tool management --------------------------------------------------

    def link_function(self, name: str, fn: Callable) -> None:
        """Attach a Python callable to an already-declared Function by name."""
        if name not in self.functions:
            raise KeyError(
                f"Function '{name}' not declared on prompt '{self.path}'. "
                f"Available: {sorted(self.functions)}"
            )
        self.functions[name].link_function(fn)

    def get_function(self, name: str) -> Function:
        """Retrieve a declared Function by name, with a clear error."""
        if name not in self.functions:
            raise KeyError(
                f"Function '{name}' not found on prompt '{self.path}'. "
                f"Available: {sorted(self.functions)}"
            )
        return self.functions[name]

    def register_mcp_server(self, server: MCP) -> None:
        self.mcp_servers[server.server_label] = server


    # -- Handler management -----------------------------------------------

    def on_exception(self, session: AgentCallSession) -> Prompt:
        return self.handler.on_exception(self, session)

    def on_interrupt(self, session: AgentCallSession) -> Prompt:
        return self.handler.on_interrupt(self, session)

    def on_interrupt_final(self, session: AgentCallSession) -> Prompt:
        return self.handler.on_interrupt_final(self, session)

    # -- Capability accessors (convenience, read-only) --------------------

    @property
    def allow_web_search(self) -> bool:
        return bool(self.addon_args.get("web_search", False))

    @property
    def computer_use_config(self) -> Dict[str, Any]:
        return self.addon_args.get("computer_use", {})

    # -- Composition ------------------------------------------------------

    def extend(self, **overrides: Any) -> Prompt:
        """
        Create a new Prompt inheriting all fields, with *overrides* applied.

        A new ``path`` is required — prompts that share a path would collide
        in the registry::

            child = parent.extend(
                path="child/analysis",
                prompt="More specific: {task}",
                output=OutputSpec(parser=strict_parser),
            )
        """
        if "path" not in overrides:
            raise ValueError("extend() requires a new 'path'")
        # Build from current field values directly, not via serialization
        current = {
            name: getattr(self, name)
            for name in type(self).model_fields
            if name not in ("_functions", "_mcp_servers")
        }
        current.update(overrides)
        return Prompt(**current)

    # -- Metadata for logging / tracking ----------------------------------

    def info_dict(self) -> Dict[str, Any]:
        """
        Return a JSON-serializable snapshot suitable for experiment tracking.
        """
        return {
            "path": self.path,
            "prompt_hash": hashlib.sha256(self.prompt.encode()).hexdigest()[:12],
            "metadata": self.metadata,
            "functions": [f.name for f in self.function_list],
            "mcp_servers": [m.server_label for m in self.mcp_servers_list],
            "addon_args": self.addon_args,
            "has_parser": self.parser is not None,
            "has_format": self.format is not None,
        }

template_vars property

Variable names required by this template (e.g. {topic} → 'topic').

__call__(**kwargs)

Render the template with the given variables.

Source code in lllm/core/prompt.py 
681
682
683
684
685
686
687
688
def __call__(self, **kwargs: Any) -> str:
    """Render the template with the given variables."""
    if not self.template_vars:  # empty set is falsy
        return self.prompt
    missing_args = self.validate_args(kwargs)
    if missing_args:
        raise ValueError(f"Missing required template variables: {missing_args} for prompt {self.path}, please provide: {self.template_vars}")
    return self.renderer.render(self.prompt, **kwargs)

extend(**overrides)

Create a new Prompt inheriting all fields, with overrides applied.

A new path is required — prompts that share a path would collide in the registry::

child = parent.extend(
    path="child/analysis",
    prompt="More specific: {task}",
    output=OutputSpec(parser=strict_parser),
)
Source code in lllm/core/prompt.py 
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
def extend(self, **overrides: Any) -> Prompt:
    """
    Create a new Prompt inheriting all fields, with *overrides* applied.

    A new ``path`` is required — prompts that share a path would collide
    in the registry::

        child = parent.extend(
            path="child/analysis",
            prompt="More specific: {task}",
            output=OutputSpec(parser=strict_parser),
        )
    """
    if "path" not in overrides:
        raise ValueError("extend() requires a new 'path'")
    # Build from current field values directly, not via serialization
    current = {
        name: getattr(self, name)
        for name in type(self).model_fields
        if name not in ("_functions", "_mcp_servers")
    }
    current.update(overrides)
    return Prompt(**current)

get_function(name)

Retrieve a declared Function by name, with a clear error.

Source code in lllm/core/prompt.py 
712
713
714
715
716
717
718
719
def get_function(self, name: str) -> Function:
    """Retrieve a declared Function by name, with a clear error."""
    if name not in self.functions:
        raise KeyError(
            f"Function '{name}' not found on prompt '{self.path}'. "
            f"Available: {sorted(self.functions)}"
        )
    return self.functions[name]

info_dict()

Return a JSON-serializable snapshot suitable for experiment tracking.

Source code in lllm/core/prompt.py 
774
775
776
777
778
779
780
781
782
783
784
785
786
787
def info_dict(self) -> Dict[str, Any]:
    """
    Return a JSON-serializable snapshot suitable for experiment tracking.
    """
    return {
        "path": self.path,
        "prompt_hash": hashlib.sha256(self.prompt.encode()).hexdigest()[:12],
        "metadata": self.metadata,
        "functions": [f.name for f in self.function_list],
        "mcp_servers": [m.server_label for m in self.mcp_servers_list],
        "addon_args": self.addon_args,
        "has_parser": self.parser is not None,
        "has_format": self.format is not None,
    }

Attach a Python callable to an already-declared Function by name.

Source code in lllm/core/prompt.py 
703
704
705
706
707
708
709
710
def link_function(self, name: str, fn: Callable) -> None:
    """Attach a Python callable to an already-declared Function by name."""
    if name not in self.functions:
        raise KeyError(
            f"Function '{name}' not declared on prompt '{self.path}'. "
            f"Available: {sorted(self.functions)}"
        )
    self.functions[name].link_function(fn)

validate_args(prompt_args)

Return list of missing required template variables for the prompt.

Source code in lllm/core/prompt.py 
674
675
676
def validate_args(self, prompt_args: dict[str, Any]) -> list[str]:
    """Return list of missing required template variables for the prompt."""
    return [v for v in self.template_vars if v not in prompt_args]

Function

lllm.core.prompt.Function

Bases: BaseModel

Declarative description of a callable tool.

The schema (name, description, properties, required) describes the tool to the LLM. The implementation is attached separately via :meth:link_function or by using the :func:tool decorator which does both in one step.

Source code in lllm/core/prompt.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
class Function(BaseModel):
    """
    Declarative description of a callable tool.

    The *schema* (name, description, properties, required) describes the tool
    to the LLM.  The *implementation* is attached separately via
    :meth:`link_function` or by using the :func:`tool` decorator which does
    both in one step.
    """

    name: str
    description: str
    properties: Dict[str, Any]
    required: List[str] = Field(default_factory=list)
    additional_properties: bool = False
    strict: bool = True

    # Implementation (attached at runtime or via decorator)
    function: Optional[Callable] = None
    processor: Callable = _default_function_call_processor

    model_config = ConfigDict(arbitrary_types_allowed=True)

    # -- Linking ----------------------------------------------------------

    def link_function(self, fn: Any) -> None:
        """Attach the Python callable that backs this tool."""
        if not callable(fn):
            raise TypeError(f"Expected a callable for function '{self.name}', got {type(fn)!r}")
        self.function = fn

    @property
    def linked(self) -> bool:
        return self.function is not None

    # -- Execution --------------------------------------------------------

    def __call__(self, function_call: FunctionCall) -> FunctionCall:
        assert self.function is not None, f"Function '{self.name}' not linked"
        try:
            result = self.function(**function_call.arguments)
        except Exception as e:
            function_call.error_message = str(e)
            function_call.result_str = f'Error: {e}'
            return function_call
        function_call.result = result
        function_call.result_str = self.processor(result, function_call)
        return function_call

    def to_tool(self, invoker: Invokers = Invokers.LITELLM) -> Optional[Dict[str, Any]]:
        # This logic might be moved to invoker specific implementations later
        if invoker == Invokers.LITELLM:
            return {
                "type": "function",
                "function": {
                    "name": self.name,
                    "description": self.description,
                    "parameters": {
                        "type": "object",
                        "properties": self.properties,
                        "required": self.required,
                        "additionalProperties": self.additional_properties,
                    },
                    "strict": self.strict
                }
            }
        raise NotImplementedError(f"Invoker {invoker} not supported for tool conversion yet")

    @classmethod
    def from_callable(
        cls,
        fn: Callable,
        *,
        name: Optional[str] = None,
        description: Optional[str] = None,
        prop_desc: Optional[Dict[str, str]] = None,
        strict: bool = True,
        processor: Callable = _default_function_call_processor,
    ) -> Function:
        """
        Build a :class:`Function` by inspecting *fn*'s signature and
        docstring.  Type hints are converted to JSON Schema types.

        Parameters without a type annotation default to ``"string"``.
        Parameters whose names end with ``*`` in the docstring (or that
        lack defaults) are treated as required.

        For example:
        ```python
        @tool(
            description="Get the current weather in a given location"
            prop_desc={
                "location": "The city and state, e.g. San Francisco, CA",
                "unit": "The unit of temperature, e.g. celsius, fahrenheit",
            }
        )
        def get_weather(location: str, unit: str = "celsius") -> str:
            ... # whatever you want to return, be sure to return a string at the end
        ```
        """
        func_name = name or fn.__name__
        func_desc = description or (inspect.getdoc(fn) or func_name)
        sig = inspect.signature(fn)
        hints = get_type_hints(fn) if hasattr(fn, "__annotations__") else {}

        prop_desc: Dict[str, str] = prop_desc or {}
        properties: Dict[str, Any] = {}
        required: List[str] = []

        for param_name, param in sig.parameters.items():
            if param_name in ("self", "cls"):
                continue
            py_type = hints.get(param_name, str)
            # Handle Optional[X] → extract X
            origin = getattr(py_type, "__origin__", None)
            if origin is Union:
                args = [a for a in py_type.__args__ if a is not type(None)]
                py_type = args[0] if args else str

            json_type = _PY_TYPE_TO_JSON.get(py_type, "string")
            prop: Dict[str, Any] = {"type": json_type}

            # Use default as example in description
            if param_name in prop_desc:
                prop["description"] = prop_desc[param_name]
            if param.default is not inspect.Parameter.empty:
                if "description" in prop:
                    prop["description"] += f" (default: {param.default!r})"
                else:
                    prop["description"] = f"(default: {param.default!r})"
            else:
                required.append(param_name)

            properties[param_name] = prop

        return cls(
            name=func_name,
            description=func_desc,
            properties=properties,
            required=required,
            strict=strict,
            function=fn,
            processor=processor,
        )

from_callable(fn, *, name=None, description=None, prop_desc=None, strict=True, processor=_default_function_call_processor) classmethod

Build a :class:Function by inspecting fn's signature and docstring. Type hints are converted to JSON Schema types.

Parameters without a type annotation default to "string". Parameters whose names end with * in the docstring (or that lack defaults) are treated as required.

For example: 

@tool(
    description="Get the current weather in a given location"
    prop_desc={
        "location": "The city and state, e.g. San Francisco, CA",
        "unit": "The unit of temperature, e.g. celsius, fahrenheit",
    }
)
def get_weather(location: str, unit: str = "celsius") -> str:
    ... # whatever you want to return, be sure to return a string at the end

Source code in lllm/core/prompt.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
@classmethod
def from_callable(
    cls,
    fn: Callable,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    prop_desc: Optional[Dict[str, str]] = None,
    strict: bool = True,
    processor: Callable = _default_function_call_processor,
) -> Function:
    """
    Build a :class:`Function` by inspecting *fn*'s signature and
    docstring.  Type hints are converted to JSON Schema types.

    Parameters without a type annotation default to ``"string"``.
    Parameters whose names end with ``*`` in the docstring (or that
    lack defaults) are treated as required.

    For example:
    ```python
    @tool(
        description="Get the current weather in a given location"
        prop_desc={
            "location": "The city and state, e.g. San Francisco, CA",
            "unit": "The unit of temperature, e.g. celsius, fahrenheit",
        }
    )
    def get_weather(location: str, unit: str = "celsius") -> str:
        ... # whatever you want to return, be sure to return a string at the end
    ```
    """
    func_name = name or fn.__name__
    func_desc = description or (inspect.getdoc(fn) or func_name)
    sig = inspect.signature(fn)
    hints = get_type_hints(fn) if hasattr(fn, "__annotations__") else {}

    prop_desc: Dict[str, str] = prop_desc or {}
    properties: Dict[str, Any] = {}
    required: List[str] = []

    for param_name, param in sig.parameters.items():
        if param_name in ("self", "cls"):
            continue
        py_type = hints.get(param_name, str)
        # Handle Optional[X] → extract X
        origin = getattr(py_type, "__origin__", None)
        if origin is Union:
            args = [a for a in py_type.__args__ if a is not type(None)]
            py_type = args[0] if args else str

        json_type = _PY_TYPE_TO_JSON.get(py_type, "string")
        prop: Dict[str, Any] = {"type": json_type}

        # Use default as example in description
        if param_name in prop_desc:
            prop["description"] = prop_desc[param_name]
        if param.default is not inspect.Parameter.empty:
            if "description" in prop:
                prop["description"] += f" (default: {param.default!r})"
            else:
                prop["description"] = f"(default: {param.default!r})"
        else:
            required.append(param_name)

        properties[param_name] = prop

    return cls(
        name=func_name,
        description=func_desc,
        properties=properties,
        required=required,
        strict=strict,
        function=fn,
        processor=processor,
    )

Attach the Python callable that backs this tool.

Source code in lllm/core/prompt.py 
200
201
202
203
204
def link_function(self, fn: Any) -> None:
    """Attach the Python callable that backs this tool."""
    if not callable(fn):
        raise TypeError(f"Expected a callable for function '{self.name}', got {type(fn)!r}")
    self.function = fn

FunctionCall

lllm.core.const.FunctionCall

Bases: BaseModel

One invocation of a tool, including its result once executed.

Source code in lllm/core/const.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class FunctionCall(BaseModel):
    """One invocation of a tool, including its result once executed."""

    id: str
    name: str
    arguments: Dict[str, Any]
    result: Any = None
    result_str: Optional[str] = None
    error_message: Optional[str] = None

    @property
    def success(self):
        return self.error_message is None and self.result_str is not None

    def __str__(self):
        _str = f'Calling function: {self.name} with arguments: {self.arguments}\n'
        if self.success:
            _str += f'Return:\n---\n{self.result_str}\n---\n'
        return _str

    def equals(self, other: FunctionCall) -> bool:
        if self.name != other.name:
            return False
        if set(self.arguments.keys()) != set(other.arguments.keys()):
            return False
        for k, v in self.arguments.items():
            if other.arguments[k] != v:
                return False
        return True

    def is_repeated(self, function_calls: List[FunctionCall]) -> bool:
        return any(self.equals(fc) for fc in function_calls)