Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-agents/src/lexigram/ai/agents/executor/executor.py: 15%
223 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""AgentExecutorImpl — runs agents with governance, memory, tracing, and metrics.
3The executor is the bridge between the agent framework and the rest
4of the Lexigram infrastructure.
5"""
7from __future__ import annotations
9from dataclasses import dataclass, replace
10from datetime import UTC
11from typing import Any
13from lexigram.ai.agents.exceptions import BudgetExceededError
14from lexigram.ai.agents.executor.streaming import astream as _astream
15from lexigram.ai.agents.observability import AgentMetrics, AgentTracer
16from lexigram.ai.agents.strategies.guard_hook import (
17 ToolObservationBlockedError,
18 ToolObservationGuardError,
19)
20from lexigram.contracts.ai.agents import (
21 AgentError,
22 AgentExecutorProtocol,
23 AgentProtocol,
24 AgentResponse,
25 MemoryProtocol,
26)
27from lexigram.contracts.ai.governance import AIGovernanceProtocol
28from lexigram.contracts.ai.guards import GuardPipelineProtocol
29from lexigram.contracts.ai.llm import (
30 ChatMessage,
31 CostEstimatorProtocol,
32 LLMClientProtocol,
33 Role,
34)
35from lexigram.contracts.ai.memory import WorkingMemoryProtocol
36from lexigram.contracts.ai.session import SessionManagerProtocol
37from lexigram.contracts.ai.skills import (
38 SkillExecutorProtocol,
39 SkillRegistryProtocol,
40)
41from lexigram.contracts.events.protocols import EventBusProtocol
42from lexigram.logging import (
43 get_logger,
44)
45from lexigram.result import Err, Ok, Result
47logger = get_logger(__name__)
50@dataclass
51class AgentObservability:
52 """Composite for agent observability components.
54 Groups metrics, tracing, and event publishing into a single injectable unit.
55 """
57 metrics: AgentMetrics | None = None
58 tracer: AgentTracer | None = None
59 event_bus: EventBusProtocol | None = None
62@dataclass
63class AgentSafetyInfra:
64 """Composite for agent safety infrastructure.
66 Groups governance and guard pipeline into a single injectable unit.
67 """
69 governance: AIGovernanceProtocol | None = None
70 guard_pipeline: GuardPipelineProtocol | None = None
73class AgentExecutorImpl(AgentExecutorProtocol):
74 """Runs an agent with full infrastructure integration.
76 Wraps agent strategy execution with:
77 1. **Governance** — budget/rate limit checks
78 2. **Memory** — conversation history load/save (legacy + working memory)
79 3. **Metrics** — execution duration, tokens, tool calls
80 4. **Tracing** — distributed spans per execution and tool call
81 5. **Events** — domain events for agent lifecycle
82 6. **Resilience** — circuit breakers on tool calls (via ToolRegistry)
83 7. **Sessions** — stateful multi-turn conversation management
84 8. **Skills** — composable skill execution and discovery
86 Usage::
88 executor = AgentExecutorImpl(llm=llm_client)
89 result = await executor.run(
90 agent=my_agent,
91 message="Where is my order?",
92 session_id="session-123",
93 )
94 """
96 def __init__(
97 self,
98 llm: LLMClientProtocol | None = None,
99 memory: MemoryProtocol | None = None,
100 working_memory: WorkingMemoryProtocol | None = None,
101 session_manager: SessionManagerProtocol | None = None,
102 skill_executor: SkillExecutorProtocol | None = None,
103 skill_registry: SkillRegistryProtocol | None = None,
104 observability: AgentObservability | None = None,
105 safety: AgentSafetyInfra | None = None,
106 cost_estimator: CostEstimatorProtocol | None = None,
107 ) -> None:
108 """Initialize the agent executor.
110 Args:
111 llm: LLM client for agent reasoning.
112 memory: Conversation memory for multi-turn sessions (legacy).
113 working_memory: Working memory for context assembly.
114 session_manager: Session manager for stateful conversations.
115 skill_executor: Skill executor for running skills.
116 skill_registry: Skill registry for discovering available skills.
117 observability: Composite for metrics, tracer, and event bus.
118 Defaults to isolated ``AgentMetrics`` and ``AgentTracer``.
119 safety: Composite for governance and guard pipeline.
120 cost_estimator: Estimates monetary cost of LLM usage for
121 governance tracking. When omitted, cost is not tracked
122 (no fabricated estimates).
123 """
124 self._llm = llm
125 self._memory = memory
126 self._working_memory = working_memory
127 self._session_manager = session_manager
128 self._skill_executor = skill_executor
129 self._skill_registry = skill_registry
130 self._cost_estimator = cost_estimator
132 self._metrics = (
133 observability.metrics
134 if observability and observability.metrics
135 else AgentMetrics()
136 )
137 self._tracer = (
138 observability.tracer
139 if observability and observability.tracer
140 else AgentTracer()
141 )
142 self._event_bus = observability.event_bus if observability else None
143 self._governance = safety.governance if safety else None
144 self._guard_pipeline = safety.guard_pipeline if safety else None
146 async def run(
147 self,
148 agent: AgentProtocol,
149 message: str,
150 session_id: str | None = None,
151 user_id: str | None = None,
152 **kwargs: Any,
153 ) -> Result[AgentResponse, AgentError]:
154 """Execute an agent with full infrastructure integration.
156 Args:
157 agent: The agent to execute.
158 message: User's input message.
159 session_id: Session ID for multi-turn memory.
160 user_id: User ID for governance tracking.
161 **kwargs: Additional parameters passed to the strategy.
163 Returns:
164 ``Ok(AgentResponse)`` on success,
165 ``Err(AgentError)`` on failure.
166 """
167 logger.info(
168 "agent_execution_start",
169 agent=agent.name,
170 session_id=session_id,
171 user_id=user_id,
172 message_length=len(message),
173 )
175 # Publish start event
176 await self._publish_event(
177 "AgentExecutionStarted",
178 {
179 "agent_name": agent.name,
180 "session_id": session_id,
181 "user_id": user_id,
182 },
183 )
185 # 1. Governance check
186 if self._governance:
187 model = getattr(self._llm, "model", "unknown")
188 provider = getattr(self._llm, "provider", "unknown")
189 try:
190 allowed = await self._governance.check_request(
191 model=model,
192 provider=provider,
193 user_id=user_id,
194 )
195 if not allowed:
196 self._metrics.record_governance_denied(agent.name)
197 logger.warning(
198 "agent_governance_denied",
199 agent=agent.name,
200 user_id=user_id,
201 )
202 await self._publish_event(
203 "AgentExecutionFailed",
204 {
205 "agent_name": agent.name,
206 "error": "Governance denied",
207 "error_type": "BudgetExceededError",
208 },
209 )
210 return Err(
211 BudgetExceededError(
212 "Agent request denied by governance policy",
213 )
214 )
215 except (RuntimeError, OSError) as e:
216 logger.warning("governance_check_failed", error=str(e))
218 # Agent-level pipeline (AgentBuilder.with_guard_pipeline) wins over the
219 # DI-provided one; the DI pipeline remains the standard default.
220 agent_pipeline = getattr(agent, "guard_pipeline", None)
221 guard_pipeline = (
222 agent_pipeline if agent_pipeline is not None else self._guard_pipeline
223 )
225 # 1.5 Guard check for input
226 if guard_pipeline:
227 try:
228 guard_res = await guard_pipeline.check_input(
229 content=message,
230 metadata={"session_id": session_id, "user_id": user_id},
231 )
232 if not guard_res.is_ok():
233 return Err(
234 AgentError(f"Guard pipeline failure: {guard_res.unwrap_err()}")
235 )
237 agg = guard_res.unwrap()
238 blocked = bool(getattr(agg, "blocked", False))
239 if blocked:
240 blocking_result = getattr(agg, "blocking_result", None)
241 reason = (
242 getattr(blocking_result, "reason", None)
243 or "Input blocked by security guards"
244 )
245 return Err(
246 AgentError(f"Input blocked by security guards: {reason}")
247 )
249 message = str(getattr(agg, "final_content", message))
250 except (RuntimeError, OSError) as e:
251 logger.exception("input_guard_evaluation_failed", agent=agent.name)
252 return Err(AgentError(f"Guard evaluation failed: {e}", cause=e))
254 # 2a. Resume or create session
255 session_state = None
256 if self._session_manager and session_id:
257 try:
258 session_state = await self._session_manager.resume(session_id)
259 if session_state is None and user_id:
260 session_state = await self._session_manager.create(
261 user_id=user_id,
262 metadata={"agent": agent.name},
263 )
264 session_id = session_state.session_id
265 logger.debug(
266 "session_loaded",
267 session_id=session_id,
268 turns=session_state.turn_count if session_state else 0,
269 )
270 except (RuntimeError, TypeError, AttributeError, LookupError) as e:
271 logger.warning("session_load_failed", error=str(e))
273 # 2b. Assemble context via working memory (preferred) or legacy memory
274 history: list[ChatMessage] = []
275 if self._working_memory:
276 try:
277 entries = await self._working_memory.assemble(
278 query=message,
279 token_budget=4096,
280 owner_id=user_id or session_id or "anonymous",
281 session_id=session_id,
282 )
283 history = [
284 ChatMessage(role=Role(e.role), content=e.content) for e in entries
285 ]
286 except (RuntimeError, TypeError, AttributeError, LookupError) as e:
287 logger.warning("working_memory_assemble_failed", error=str(e))
288 elif self._memory and session_id:
289 try:
290 if hasattr(self._memory, "get_messages"):
291 msgs = await self._memory.get_messages()
292 history = [
293 ChatMessage(role=Role(m.role), content=m.content) for m in msgs
294 ]
295 except (RuntimeError, TypeError, AttributeError, LookupError) as e:
296 logger.warning("memory_load_failed", error=str(e))
298 # 2c. Merge skills as additional tools
299 tools: list[Any] = list(agent.tools) if agent.tools else []
300 if self._skill_registry:
301 try:
302 skill_schemas = self._skill_registry.get_schemas()
303 tools.extend(skill_schemas)
304 except (RuntimeError, TypeError, AttributeError, LookupError) as e:
305 logger.warning("skill_registry_merge_failed", error=str(e))
307 # 3. Execute strategy with tracing
308 strategy = getattr(agent, "strategy", None)
309 if strategy is None:
310 from lexigram.ai.agents.strategies.react import ReActStrategy
312 strategy = ReActStrategy()
314 if self._llm is None:
315 logger.error("agent_executor_missing_llm", agent=agent.name)
316 await self._publish_event(
317 "AgentExecutionFailed",
318 {
319 "agent_name": agent.name,
320 "error": "No LLM client configured",
321 "error_type": "ConfigurationError",
322 },
323 )
324 return Err(AgentError("Agent executor has no LLM client configured"))
326 try:
327 async with self._tracer.trace_execution(
328 agent.name,
329 message,
330 session_id,
331 ) as span:
332 strategy_history = [
333 {"role": m.role, "content": m.content} for m in history
334 ]
336 result = await strategy.execute(
337 message=message,
338 tools=tools,
339 history=strategy_history,
340 llm=self._llm,
341 system_prompt=agent.system_prompt,
342 temperature=getattr(agent, "temperature", 0.7),
343 tool_registry=kwargs.get("tool_registry"),
344 memory=getattr(agent, "memory", None),
345 guard_pipeline=guard_pipeline,
346 **kwargs,
347 )
349 if span and hasattr(result, "is_ok"):
350 if result.is_ok():
351 response = result.unwrap()
352 if hasattr(span, "set_attribute"):
353 span.set_attribute("agent.steps", response.step_count)
354 span.set_attribute("agent.tokens", response.total_tokens)
355 span.set_attribute(
356 "agent.tool_calls", response.tool_call_count
357 )
359 except (ToolObservationBlockedError, ToolObservationGuardError) as e:
360 logger.warning(
361 "agent.guard_blocked_observation",
362 agent=agent.name,
363 error=str(e),
364 )
365 self._metrics.record_error(agent.name, type(e).__name__)
366 await self._publish_event(
367 "AgentExecutionFailed",
368 {
369 "agent_name": agent.name,
370 "error": str(e),
371 "error_type": type(e).__name__,
372 },
373 )
374 if isinstance(e, ToolObservationBlockedError):
375 message_text = f"Tool observation blocked by guards: {e}"
376 else:
377 message_text = f"Tool observation guard evaluation failed: {e}"
378 return Err(AgentError(message_text, cause=e))
380 except (RuntimeError, OSError) as e:
381 logger.exception("strategy_execution_failed", agent=agent.name)
382 self._metrics.record_error(agent.name, type(e).__name__)
383 await self._publish_event(
384 "AgentExecutionFailed",
385 {
386 "agent_name": agent.name,
387 "error": str(e),
388 "error_type": type(e).__name__,
389 },
390 )
391 return Err(AgentError(f"Strategy failed: {e}", cause=e))
393 if not result.is_ok():
394 error = result.unwrap_err()
395 error_type = type(error).__name__
396 self._metrics.record_error(agent.name, error_type)
397 await self._publish_event(
398 "AgentExecutionFailed",
399 {
400 "agent_name": agent.name,
401 "error": str(error),
402 "error_type": error_type,
403 },
404 )
405 if isinstance(error, AgentError):
406 return Err(error)
407 return Err(AgentError(str(error), cause=error))
409 response = result.unwrap()
411 # 3.5 Guard check for output
412 if guard_pipeline:
413 try:
414 out_guard_res = await guard_pipeline.check_output(
415 content=response.message,
416 original_input=message,
417 metadata={"session_id": session_id, "user_id": user_id},
418 )
419 if not out_guard_res.is_ok():
420 return Err(
421 AgentError(
422 f"Guard pipeline failure: {out_guard_res.unwrap_err()}"
423 )
424 )
426 agg = out_guard_res.unwrap()
427 blocked = bool(getattr(agg, "blocked", False))
428 if blocked:
429 blocking_result = getattr(agg, "blocking_result", None)
430 reason = (
431 getattr(blocking_result, "reason", None)
432 or "Output blocked by security guards"
433 )
434 return Err(
435 AgentError(f"Output blocked by security guards: {reason}")
436 )
438 response = replace(
439 response,
440 message=str(getattr(agg, "final_content", response.message)),
441 )
442 except (RuntimeError, OSError) as e:
443 logger.exception("output_guard_evaluation_failed", agent=agent.name)
444 return Err(AgentError(f"Guard evaluation failed: {e}", cause=e))
446 # 4. Record metrics
447 self._metrics.record_execution(agent.name, response)
448 for tc in response.tool_calls:
449 self._metrics.record_tool_call(agent.name, tc)
451 # 5. Save to memory (working memory, session, and/or legacy)
452 if self._working_memory:
453 try:
454 from datetime import datetime
456 from lexigram.contracts.ai.memory import MemoryEntry
458 owner = user_id or session_id or "anonymous"
459 user_entry = MemoryEntry(
460 id=f"{session_id or 'no-session'}-user-{response.step_count}",
461 owner_id=owner,
462 content=message,
463 role="user",
464 timestamp=datetime.now(UTC),
465 )
466 assistant_entry = MemoryEntry(
467 id=f"{session_id or 'no-session'}-assistant-{response.step_count}",
468 owner_id=owner,
469 content=response.message,
470 role="assistant",
471 timestamp=datetime.now(UTC),
472 )
473 await self._working_memory.add(user_entry)
474 await self._working_memory.add(assistant_entry)
475 except (RuntimeError, TypeError, AttributeError, LookupError) as e:
476 logger.warning("working_memory_save_failed", error=str(e))
478 if self._session_manager and session_id:
479 try:
480 from datetime import datetime
482 from lexigram.contracts.ai.session import SessionTurn
484 user_turn = SessionTurn(
485 turn_id=f"{session_id}-user-{response.step_count}",
486 role="user",
487 content=message,
488 timestamp=datetime.now(UTC),
489 )
490 assistant_turn = SessionTurn(
491 turn_id=f"{session_id}-assistant-{response.step_count}",
492 role="assistant",
493 content=response.message,
494 timestamp=datetime.now(UTC),
495 tokens_used=response.total_tokens,
496 tool_calls=[
497 {"name": tc.tool_name, "result": tc.result}
498 for tc in response.tool_calls
499 ],
500 )
501 await self._session_manager.add_turn(session_id, user_turn)
502 await self._session_manager.add_turn(session_id, assistant_turn)
503 except (RuntimeError, TypeError, AttributeError, LookupError) as e:
504 logger.warning("session_turn_save_failed", error=str(e))
506 if self._memory and session_id:
507 try:
508 if hasattr(self._memory, "add"):
509 await self._memory.add("user", message)
510 await self._memory.add("assistant", response.message)
511 elif hasattr(self._memory, "add_message"):
512 await self._memory.add_message(
513 ChatMessage(role=Role.USER, content=message),
514 )
515 await self._memory.add_message(
516 ChatMessage(role=Role.ASSISTANT, content=response.message),
517 )
518 except (RuntimeError, TypeError, AttributeError, LookupError) as e:
519 logger.warning("memory_save_failed", error=str(e))
521 # 6. Track cost (only when a real estimator is configured)
522 if self._governance and self._cost_estimator and response.total_tokens > 0:
523 try:
524 model = getattr(self._llm, "model", "unknown")
525 estimated_cost = self._cost_estimator.estimate_cost(
526 model=model,
527 total_tokens=response.total_tokens,
528 provider=getattr(self._llm, "provider", None),
529 prompt_tokens=response.prompt_tokens,
530 completion_tokens=response.completion_tokens,
531 )
532 await self._governance.track_cost(
533 cost=estimated_cost,
534 model=model,
535 user_id=user_id,
536 )
537 response = replace(response, total_cost=estimated_cost)
538 except (RuntimeError, OSError) as e:
539 logger.warning("cost_tracking_failed", error=str(e))
541 response = replace(response, session_id=session_id)
543 # 7. Publish completion event
544 await self._publish_event(
545 "AgentExecutionCompleted",
546 {
547 "agent_name": agent.name,
548 "session_id": session_id,
549 "steps": response.step_count,
550 "tool_calls": response.tool_call_count,
551 "total_tokens": response.total_tokens,
552 "prompt_tokens": response.prompt_tokens,
553 "completion_tokens": response.completion_tokens,
554 "total_cost": response.total_cost,
555 "duration_ms": response.duration_ms,
556 },
557 )
559 logger.info(
560 "agent_execution_complete",
561 agent=agent.name,
562 steps=response.step_count,
563 tool_calls=response.tool_call_count,
564 tokens=response.total_tokens,
565 duration_ms=round(response.duration_ms, 2),
566 )
568 return Ok(response)
570 async def astream(
571 self,
572 agent: AgentProtocol,
573 message: str,
574 session_id: str | None = None,
575 user_id: str | None = None,
576 **kwargs: Any,
577 ):
578 """Stream agent execution events.
580 Yields AgentEvent objects as the agent executes, enabling
581 real-time monitoring of thoughts, tool calls, and messages.
583 Args:
584 agent: The agent to execute.
585 message: User's input message.
586 session_id: Session ID for multi-turn memory.
587 user_id: User ID for governance tracking.
588 **kwargs: Additional parameters passed to the strategy.
590 Yields:
591 AgentEvent objects with type, data, and run_id.
592 """
593 async for event in _astream(
594 self,
595 agent,
596 message,
597 session_id,
598 user_id,
599 **kwargs,
600 ):
601 yield event
603 async def _publish_event(
604 self,
605 event_type: str,
606 data: dict[str, Any],
607 ) -> None:
608 """Publish an agent domain event if EventBusProtocol is available."""
609 if not self._event_bus:
610 return
612 try:
613 from lexigram.ai.agents import events as agent_events
615 event_cls = getattr(agent_events, event_type, None)
616 if event_cls:
617 event_obj = event_cls(**data)
618 await self._event_bus.publish(event_obj)
619 except (ImportError, AttributeError):
620 pass
621 except (RuntimeError, OSError, ConnectionError, TypeError) as e:
622 logger.debug("event_publish_failed", event_type=event_type, error=str(e))