« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Native function-calling strategy for agent reasoning.
3Drives the LLM through its native tool-calling interface: ``ToolDefinition``
4schemas are sent to ``complete(..., tools=...)`` and the tool calls returned
5by the model are executed directly, with results fed back as ``tool`` role
6messages. A text-marker fallback (``ACTION:`` / ``FINAL_ANSWER:``) keeps the
7strategy working with models or providers that do not honour native schemas.
9This is the strategy that consumes tool schemas from
10``ToolRegistry.list_tool_schemas()``, so the schema path is exercised end to
11end rather than existing as dead code.
13Termination:
14 - The model returns no tool calls (its message is the final answer).
15 - A text-marker ``FINAL_ANSWER:`` is parsed from the content.
16 - The maximum iteration count is reached.
17"""
19from __future__ import annotations
21import asyncio
22import time
23from typing import TYPE_CHECKING, Any, cast
25from lexigram.ai.agents.strategies.base import AbstractStrategy
26from lexigram.ai.agents.strategies.parsing import (
27 build_chat_messages_from_dict,
28 extract_final_answer,
29 extract_tool_call,
30)
31from lexigram.ai.agents.strategies.token_utils import count_tokens, token_split
32from lexigram.ai.agents.types import ReasoningStep, ToolExecutionRecord
33from lexigram.contracts.ai.agents import (
34 AgentError,
35 AgentResponse,
36 ToolDefinition,
37 ToolProtocol,
38)
39from lexigram.contracts.ai.llm import ChatMessage, Role
40from lexigram.logging import (
41 get_logger,
42)
43from lexigram.result import Err, Ok, Result
44from lexigram.serialization import loads_str
46if TYPE_CHECKING:
47 from lexigram.contracts.ai.agents import MemoryProtocol
48 from lexigram.contracts.ai.llm import CompletionProtocol, LLMClientProtocol
50logger = get_logger(__name__)
52_SYSTEM_SUFFIX = """
53You are a function-calling assistant. Use the tools available to you to
54complete the user's request. The tool schema is enforced by the model, so
55request tools through native function calls rather than free text.
57## Rules
58- Call a tool when you need information you do not already have.
59- Read each tool result before deciding the next step.
60- Once the request is satisfied, answer the user directly in natural language.
61"""
63_OBSERVATION_TEMPLATE = "OBSERVATION: {observation}"
66class FunctionCallingStrategy(AbstractStrategy):
67 """Native tool-calling reasoning strategy.
69 Sends function schemas to the LLM and executes the tool calls the model
70 returns. Falls back to text-marker action parsing for providers without
71 native tool-calling support.
73 Example::
75 from lexigram.ai.agents import Agent
76 from lexigram.ai.agents.strategies import FunctionCallingStrategy
78 agent = Agent(
79 llm=my_llm_client,
80 strategy=FunctionCallingStrategy(max_iterations=10),
81 )
82 """
84 def __init__(
85 self,
86 max_iterations: int = 10,
87 tool_timeout: float = 30.0,
88 observation_max_chars: int = 10_000,
89 timeout: float = 120.0,
90 tool_max_retries: int = 3,
91 ) -> None:
92 """Initialise the function-calling strategy.
94 Args:
95 max_iterations: Maximum number of tool-call rounds.
96 tool_timeout: Per-tool execution timeout in seconds.
97 observation_max_chars: Maximum characters for tool output before
98 truncation.
99 timeout: Per-LLM-call timeout in seconds.
100 tool_max_retries: Retry attempts for transient tool errors
101 (``ConnectionError``, ``OSError``).
102 """
103 self.max_iterations = max_iterations
104 self.tool_timeout = tool_timeout
105 self.observation_max_chars = observation_max_chars
106 self.timeout = timeout
107 self.tool_max_retries = tool_max_retries
109 async def execute(
110 self,
111 message: str,
112 tools: list[ToolProtocol],
113 history: list[dict[str, Any]],
114 llm: LLMClientProtocol,
115 **kwargs: Any,
116 ) -> Result[AgentResponse, AgentError]:
117 """Execute the function-calling loop.
119 Args:
120 message: The user's input message.
121 tools: Tools available to the agent.
122 history: Conversation history as ChatMessage objects.
123 llm: LLM client implementing ``LLMClientProtocol``.
124 **kwargs: Additional parameters:
125 - ``system_prompt`` (str): Optional system prompt prefix.
126 - ``memory``: Optional memory backend for context retrieval.
127 - ``tool_registry``: Optional tool registry whose
128 ``list_tool_schemas()`` output is merged into the schemas
129 sent to the model.
131 Returns:
132 Ok(AgentResponse) with the final answer and reasoning trace, or
133 Err(AgentError) on unrecoverable failure.
134 """
135 system_prompt: str = kwargs.get("system_prompt", "")
136 memory = kwargs.get("memory")
137 tool_registry = kwargs.get("tool_registry")
138 guard_pipeline = kwargs.get("guard_pipeline")
140 steps: list[ReasoningStep] = []
141 tool_records: list[ToolExecutionRecord] = []
142 total_tokens = 0
143 prompt_tokens = 0
144 completion_tokens = 0
145 start_time = time.monotonic()
147 tool_map: dict[str, ToolProtocol] = {t.name: t for t in tools}
148 schemas = self._build_schemas(tools, tool_registry)
150 memory_context = await self._get_memory_context(memory)
151 full_system = system_prompt + memory_context + _SYSTEM_SUFFIX
152 messages = build_chat_messages_from_dict(message, history, full_system)
154 for iteration in range(1, self.max_iterations + 1):
155 completion = await self._call_llm(llm, messages, schemas)
156 if completion is None:
157 return Err(AgentError(f"LLM failed at iteration {iteration}"))
159 step_prompt, step_completion = self._token_split(completion)
160 prompt_tokens += step_prompt
161 completion_tokens += step_completion
162 total_tokens += self._count_tokens(completion)
164 native_calls = getattr(completion, "tool_calls", None) or []
165 if native_calls:
166 if await self._handle_native_calls(
167 completion,
168 iteration,
169 messages,
170 steps,
171 tool_records,
172 tool_map,
173 guard_pipeline=guard_pipeline,
174 ):
175 continue
177 content = getattr(completion, "content", None) or ""
178 if not content.strip():
179 steps.append(
180 ReasoningStep(
181 step_number=iteration,
182 thought=content,
183 action=None,
184 observation="[No valid tool call returned — retrying]",
185 )
186 )
187 messages.append(
188 ChatMessage(
189 role=Role.ASSISTANT,
190 content=content or "",
191 tool_calls=native_calls or None,
192 )
193 )
194 messages.append(
195 ChatMessage(
196 role=Role.USER,
197 content=(
198 "Your previous response contained no tool calls or "
199 "final answer. Either call a tool natively or "
200 "answer the user directly."
201 ),
202 )
203 )
204 continue
206 final_answer = extract_final_answer(content)
207 if final_answer is not None:
208 final = final_answer
209 else:
210 tool_name, tool_args = extract_tool_call(content)
211 if tool_name is not None:
212 logger.debug(
213 "function_calling_text_fallback",
214 iteration=iteration,
215 tool=tool_name,
216 )
217 await self._handle_text_tool(
218 tool_name,
219 tool_args,
220 content,
221 iteration,
222 messages,
223 steps,
224 tool_records,
225 tool_map,
226 guard_pipeline=guard_pipeline,
227 )
228 continue
229 final = content
231 steps.append(
232 ReasoningStep(
233 step_number=iteration,
234 thought=content,
235 action="final_answer",
236 observation=final,
237 )
238 )
239 elapsed = (time.monotonic() - start_time) * 1000
240 return Ok(
241 AgentResponse(
242 message=final,
243 steps=steps,
244 tool_calls=tool_records,
245 total_tokens=total_tokens,
246 prompt_tokens=prompt_tokens,
247 completion_tokens=completion_tokens,
248 duration_ms=elapsed,
249 metadata={
250 "strategy": "function_calling",
251 "iterations": iteration,
252 },
253 )
254 )
256 elapsed = (time.monotonic() - start_time) * 1000
257 last_obs = steps[-1].observation if steps else "No response generated"
258 return Ok(
259 AgentResponse(
260 message=f"[Max iterations reached] {last_obs}",
261 steps=steps,
262 tool_calls=tool_records,
263 total_tokens=total_tokens,
264 prompt_tokens=prompt_tokens,
265 completion_tokens=completion_tokens,
266 duration_ms=elapsed,
267 metadata={
268 "strategy": "function_calling",
269 "iterations": self.max_iterations,
270 "max_iterations_reached": True,
271 },
272 )
273 )
275 # ------------------------------------------------------------------
276 # Schema building
277 # ------------------------------------------------------------------
279 def _build_schemas(
280 self,
281 tools: list[ToolProtocol],
282 tool_registry: Any,
283 ) -> list[ToolDefinition]:
284 """Build the native tool schema list sent to the LLM.
286 Merges per-tool schemas with any schemas exposed by ``tool_registry``
287 (``list_tool_schemas``), wiring the registry schema builder into the
288 live tool-calling path.
290 Args:
291 tools: Executable tools to describe.
292 tool_registry: Optional tool registry.
294 Returns:
295 List of ``ToolDefinition`` schemas.
296 """
297 schemas = [
298 ToolDefinition(
299 name=t.name,
300 description=t.description,
301 parameters=t.parameters_schema,
302 )
303 for t in tools
304 ]
305 present = {s.name for s in schemas}
306 list_schemas = getattr(tool_registry, "list_tool_schemas", None)
307 if list_schemas is not None:
308 for raw in list_schemas():
309 function = raw.get("function", {})
310 name = function.get("name")
311 if name and name not in present:
312 schemas.append(
313 ToolDefinition(
314 name=name,
315 description=function.get("description", ""),
316 parameters=function.get("parameters", {}),
317 )
318 )
319 present.add(name)
320 return schemas
322 # ------------------------------------------------------------------
323 # LLM interaction
324 # ------------------------------------------------------------------
326 async def _call_llm(
327 self,
328 llm: LLMClientProtocol,
329 messages: list[ChatMessage],
330 schemas: list[ToolDefinition],
331 ) -> CompletionProtocol | None:
332 """Call the LLM with tool schemas and return the completion.
334 Args:
335 llm: LLM client implementing ``LLMClientProtocol``.
336 messages: Chat message history.
337 schemas: Native tool schemas to advertise to the model.
339 Returns:
340 The completion object, or ``None`` when the call failed.
341 """
342 try:
343 result = await asyncio.wait_for(
344 llm.complete(
345 cast("list[Any]", messages),
346 tools=schemas or None,
347 ),
348 timeout=self.timeout,
349 )
350 except TimeoutError:
351 logger.warning("function_calling_llm_timeout", timeout=self.timeout)
352 return None
353 except (OSError, ConnectionError, RuntimeError, ValueError) as exc:
354 logger.warning("function_calling_llm_error", error=str(exc))
355 return None
357 if not result.is_ok():
358 logger.warning(
359 "function_calling_llm_err_result",
360 error=str(result.unwrap_err()),
361 )
362 return None
363 return result.unwrap()
365 def _count_tokens(self, completion: CompletionProtocol) -> int:
366 """Extract total token usage from a completion, if reported."""
367 return count_tokens(completion)
369 def _token_split(self, completion: CompletionProtocol) -> tuple[int, int]:
370 """Extract the prompt/completion token split, if reported.
372 Args:
373 completion: LLM completion result.
375 Returns:
376 Tuple of ``(prompt_tokens, completion_tokens)``. Both are
377 ``0`` when usage is missing.
378 """
379 return token_split(completion)
381 # ------------------------------------------------------------------
382 # Native tool loop
383 # ------------------------------------------------------------------
385 async def _handle_native_calls(
386 self,
387 completion: CompletionProtocol,
388 iteration: int,
389 messages: list[ChatMessage],
390 steps: list[ReasoningStep],
391 tool_records: list[ToolExecutionRecord],
392 tool_map: dict[str, ToolProtocol],
393 guard_pipeline: Any = None,
394 ) -> bool:
395 """Execute native tool calls and feed results back as tool messages.
397 The assistant message carrying the native calls is inserted before the
398 matching ``tool`` role responses so the provider can re-emit the full
399 round trip.
401 Returns:
402 ``True`` when at least one tool call was executed (loop continues),
403 ``False`` when there was nothing executable.
404 """
405 native_calls = getattr(completion, "tool_calls", None) or []
406 if not native_calls:
407 return False
409 assistant_idx = len(messages)
410 executed = False
411 for native_call in native_calls:
412 function = getattr(native_call, "function", None)
413 if function is None:
414 continue
415 tool_name = function.name
416 tool_args = self._parse_args(getattr(function, "arguments", {}))
417 record = await self._execute_tool(tool_name, tool_args, tool_map)
418 tool_records.append(record)
419 executed = True
421 observation = (
422 str(record.result) if record.succeeded else f"Error: {record.error}"
423 )
424 # Guard before truncation so detectors see the full content
425 from lexigram.ai.agents.strategies.guard_hook import guard_observation
427 observation = await guard_observation(
428 guard_pipeline, observation, tool_name=tool_name
429 )
430 if len(observation) > self.observation_max_chars:
431 observation = (
432 observation[: self.observation_max_chars] + "\n[TRUNCATED]"
433 )
435 steps.append(
436 ReasoningStep(
437 step_number=iteration,
438 thought=getattr(completion, "content", None) or "",
439 action=tool_name,
440 tool_call=record,
441 observation=observation,
442 )
443 )
444 messages.append(
445 ChatMessage(
446 role=Role.TOOL,
447 content=observation,
448 tool_call_id=native_call.id,
449 )
450 )
452 if executed:
453 messages.insert(
454 assistant_idx,
455 ChatMessage(
456 role=Role.ASSISTANT,
457 content=getattr(completion, "content", None) or "",
458 tool_calls=list(native_calls),
459 ),
460 )
461 return executed
463 async def _handle_text_tool(
464 self,
465 tool_name: str,
466 tool_args: dict[str, Any],
467 content: str,
468 iteration: int,
469 messages: list[ChatMessage],
470 steps: list[ReasoningStep],
471 tool_records: list[ToolExecutionRecord],
472 tool_map: dict[str, ToolProtocol],
473 guard_pipeline: Any = None,
474 ) -> None:
475 """Execute a tool requested through text markers (fallback path)."""
476 record = await self._execute_tool(tool_name, tool_args, tool_map)
477 tool_records.append(record)
479 observation = (
480 str(record.result) if record.succeeded else f"Error: {record.error}"
481 )
482 # Guard before truncation so detectors see the full content
483 from lexigram.ai.agents.strategies.guard_hook import guard_observation
485 observation = await guard_observation(
486 guard_pipeline, observation, tool_name=tool_name
487 )
488 if len(observation) > self.observation_max_chars:
489 observation = observation[: self.observation_max_chars] + "\n[TRUNCATED]"
491 steps.append(
492 ReasoningStep(
493 step_number=iteration,
494 thought=content,
495 action=tool_name,
496 tool_call=record,
497 observation=observation,
498 )
499 )
500 messages.append(ChatMessage(role=Role.ASSISTANT, content=content))
501 messages.append(
502 ChatMessage(
503 role=Role.USER,
504 content=_OBSERVATION_TEMPLATE.format(observation=observation),
505 )
506 )
508 # ------------------------------------------------------------------
509 # Tool execution
510 # ------------------------------------------------------------------
512 def _parse_args(self, raw: Any) -> dict[str, Any]:
513 """Parse tool-call arguments that may arrive JSON-encoded or as a dict."""
514 if isinstance(raw, dict):
515 return raw
516 if not raw:
517 return {}
518 try:
519 parsed = loads_str(raw)
520 return parsed if isinstance(parsed, dict) else {}
521 except (TypeError, ValueError):
522 return {}
524 async def _execute_tool(
525 self,
526 tool_name: str,
527 tool_args: dict[str, Any],
528 tool_map: dict[str, ToolProtocol],
529 ) -> ToolExecutionRecord:
530 """Execute a tool with timeout and retry on transient errors."""
531 if tool_name not in tool_map:
532 return ToolExecutionRecord(
533 tool_name=tool_name,
534 arguments=tool_args,
535 error=f"Unknown tool: {tool_name}. Available: {list(tool_map)}",
536 )
538 tool = tool_map[tool_name]
539 start = time.monotonic()
540 last_error: BaseException | None = None
542 for attempt in range(self.tool_max_retries):
543 try:
544 output = await asyncio.wait_for(
545 tool.execute(**tool_args),
546 timeout=self.tool_timeout,
547 )
548 duration = (time.monotonic() - start) * 1000
549 return ToolExecutionRecord(
550 tool_name=tool_name,
551 arguments=tool_args,
552 result=output,
553 duration_ms=duration,
554 )
555 except TimeoutError:
556 duration = (time.monotonic() - start) * 1000
557 return ToolExecutionRecord(
558 tool_name=tool_name,
559 arguments=tool_args,
560 error=f"Tool '{tool_name}' timed out after {self.tool_timeout}s",
561 duration_ms=duration,
562 )
563 except (ConnectionError, OSError) as exc:
564 last_error = exc
565 logger.warning(
566 "function_calling_tool_transient_error",
567 tool=tool_name,
568 attempt=attempt + 1,
569 error=str(exc),
570 )
571 if attempt < self.tool_max_retries - 1:
572 await asyncio.sleep(1.0 * (2**attempt))
573 except (RuntimeError, TypeError, ValueError, LookupError) as exc:
574 duration = (time.monotonic() - start) * 1000
575 return ToolExecutionRecord(
576 tool_name=tool_name,
577 arguments=tool_args,
578 error=f"Tool '{tool_name}' failed: {exc}",
579 duration_ms=duration,
580 )
582 duration = (time.monotonic() - start) * 1000
583 return ToolExecutionRecord(
584 tool_name=tool_name,
585 arguments=tool_args,
586 error=(
587 f"Tool '{tool_name}' failed after {self.tool_max_retries} "
588 f"retries: {last_error}"
589 ),
590 duration_ms=duration,
591 )
593 # ------------------------------------------------------------------
594 # Memory context
595 # ------------------------------------------------------------------
597 @staticmethod
598 async def _get_memory_context(memory: MemoryProtocol | None) -> str:
599 """Retrieve context from memory backend if available."""
600 if memory is None:
601 return ""
602 try:
603 past_messages = await memory.get_messages()
604 if past_messages:
605 context_str = "\n".join(str(m) for m in past_messages[-5:])
606 return f"\n\nRelevant context from memory:\n{context_str}"
607 except (RuntimeError, TypeError, ValueError, OSError, AttributeError):
608 pass
609 return ""
612__all__ = ["FunctionCallingStrategy"]