Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-agents/src/lexigram/ai/agents/strategies/function_calling.py: 16%

184 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-25 07:19 +0800

1"""Native function-calling strategy for agent reasoning. 

2 

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. 

8 

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. 

12 

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""" 

18 

19from __future__ import annotations 

20 

21import asyncio 

22import time 

23from typing import TYPE_CHECKING, Any, cast 

24 

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 

45 

46if TYPE_CHECKING: 

47 from lexigram.contracts.ai.agents import MemoryProtocol 

48 from lexigram.contracts.ai.llm import CompletionProtocol, LLMClientProtocol 

49 

50logger = get_logger(__name__) 

51 

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. 

56 

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""" 

62 

63_OBSERVATION_TEMPLATE = "OBSERVATION: {observation}" 

64 

65 

66class FunctionCallingStrategy(AbstractStrategy): 

67 """Native tool-calling reasoning strategy. 

68 

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. 

72 

73 Example:: 

74 

75 from lexigram.ai.agents import Agent 

76 from lexigram.ai.agents.strategies import FunctionCallingStrategy 

77 

78 agent = Agent( 

79 llm=my_llm_client, 

80 strategy=FunctionCallingStrategy(max_iterations=10), 

81 ) 

82 """ 

83 

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. 

93 

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 

108 

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. 

118 

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. 

130 

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") 

139 

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() 

146 

147 tool_map: dict[str, ToolProtocol] = {t.name: t for t in tools} 

148 schemas = self._build_schemas(tools, tool_registry) 

149 

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) 

153 

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}")) 

158 

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) 

163 

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 

176 

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 

205 

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 

230 

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 ) 

255 

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 ) 

274 

275 # ------------------------------------------------------------------ 

276 # Schema building 

277 # ------------------------------------------------------------------ 

278 

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. 

285 

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. 

289 

290 Args: 

291 tools: Executable tools to describe. 

292 tool_registry: Optional tool registry. 

293 

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 

321 

322 # ------------------------------------------------------------------ 

323 # LLM interaction 

324 # ------------------------------------------------------------------ 

325 

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. 

333 

334 Args: 

335 llm: LLM client implementing ``LLMClientProtocol``. 

336 messages: Chat message history. 

337 schemas: Native tool schemas to advertise to the model. 

338 

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 

356 

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() 

364 

365 def _count_tokens(self, completion: CompletionProtocol) -> int: 

366 """Extract total token usage from a completion, if reported.""" 

367 return count_tokens(completion) 

368 

369 def _token_split(self, completion: CompletionProtocol) -> tuple[int, int]: 

370 """Extract the prompt/completion token split, if reported. 

371 

372 Args: 

373 completion: LLM completion result. 

374 

375 Returns: 

376 Tuple of ``(prompt_tokens, completion_tokens)``. Both are 

377 ``0`` when usage is missing. 

378 """ 

379 return token_split(completion) 

380 

381 # ------------------------------------------------------------------ 

382 # Native tool loop 

383 # ------------------------------------------------------------------ 

384 

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. 

396 

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. 

400 

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 

408 

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 

420 

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 

426 

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 ) 

434 

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 ) 

451 

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 

462 

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) 

478 

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 

484 

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]" 

490 

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 ) 

507 

508 # ------------------------------------------------------------------ 

509 # Tool execution 

510 # ------------------------------------------------------------------ 

511 

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 {} 

523 

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 ) 

537 

538 tool = tool_map[tool_name] 

539 start = time.monotonic() 

540 last_error: BaseException | None = None 

541 

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 ) 

581 

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 ) 

592 

593 # ------------------------------------------------------------------ 

594 # Memory context 

595 # ------------------------------------------------------------------ 

596 

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 "" 

610 

611 

612__all__ = ["FunctionCallingStrategy"]