diff --git a/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py b/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py index 31430f7174..3aba946c88 100644 --- a/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py +++ b/src/praisonai-agents/praisonaiagents/agent/chat_mixin.py @@ -31,6 +31,11 @@ _turn_loop_guard = contextvars.ContextVar("praisonai_turn_loop_guard", default=None) +def _agent_end_status(db_status: str) -> str: + """``agent_end`` status for replay: completed, failed or cancelled.""" + return "failed" if db_status == "error" else db_status + + def _enter_turn(agent, cancel): """Mark a ``chat``/``achat`` turn in flight and bind its cancel token to this context.""" with _turns_lock: @@ -3618,7 +3623,7 @@ def chat(self, prompt: str, temperature: Optional[float] = None, tools: Optional end_durable_run(durable_token) durable_context.close() try: - _trace_emitter.agent_end(self.name) + _trace_emitter.agent_end(self.name, {"status": _agent_end_status(db_status)}) finally: if getattr(self, "_active_turn_token", None) is _cancel: self._active_turn_token = None @@ -4467,7 +4472,7 @@ async def achat(self, prompt: str, temperature: Optional[float] = None, tools: O await durable_context.aclose() finally: try: - _trace_emitter.agent_end(self.name) + _trace_emitter.agent_end(self.name, {"status": _agent_end_status(db_status)}) finally: if getattr(self, "_active_turn_token", None) is _cancel: self._active_turn_token = None @@ -6775,6 +6780,57 @@ def _chat_completion_with_retry_core(self, messages, temperature=None, tools=Non raise RuntimeError("Retry loop completed without returning or raising an exception") async def _achat_completion_with_retry(self, messages, temperature=None, tools=None, stream=None, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, stream_callback=None, emit_events=True): + """Async model call with retry, recorded as ``llm_request``/``llm_response`` like the sync path. + + ``Agent.astart``/``achat`` recorded no model calls before this, so a replay + of an async run showed the agent but never what it asked the model. + """ + from ..trace.context_events import get_context_emitter + + _trace_emitter = get_context_emitter() + call = self._achat_completion_with_retry_dispatch( + messages, temperature, tools, stream, reasoning_steps, task_name, + task_description, task_id, response_format, + stream_callback=stream_callback, emit_events=emit_events, + ) + if not _trace_emitter._enabled: # zero overhead when tracing is off + return await call + _trace_emitter.llm_request( + self.name, + messages_count=len(messages), + tokens_used=0, + model=self.llm if isinstance(self.llm, str) else None, + messages=messages, + ) + started = time.time() + try: + response = await call + except Exception as e: + _trace_emitter.llm_response( + self.name, + duration_ms=(time.time() - started) * 1000, + finish_reason="error", + response_content=str(e), + ) + raise + prompt_tokens = completion_tokens = 0 + cost_usd = 0.0 + usage = getattr(response, "usage", None) + if usage: + prompt_tokens = getattr(usage, "prompt_tokens", 0) or 0 + completion_tokens = getattr(usage, "completion_tokens", 0) or 0 + cost_usd = self._calculate_llm_cost(prompt_tokens, completion_tokens, response=response) + _trace_emitter.llm_response( + self.name, + duration_ms=(time.time() - started) * 1000, + response_content=self._extract_llm_response_content(response), + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + cost_usd=cost_usd, + ) + return response + + async def _achat_completion_with_retry_dispatch(self, messages, temperature=None, tools=None, stream=None, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, stream_callback=None, emit_events=True): """ Async wrapper for _execute_unified_achat_completion that adds jittered exponential backoff retry logic. diff --git a/src/praisonai-agents/praisonaiagents/trace/context_events.py b/src/praisonai-agents/praisonaiagents/trace/context_events.py index 926338944c..b94cd4a5a1 100644 --- a/src/praisonai-agents/praisonaiagents/trace/context_events.py +++ b/src/praisonai-agents/praisonaiagents/trace/context_events.py @@ -85,6 +85,16 @@ def get_context_emitter() -> "ContextTraceEmitter": return _default_disabled_emitter +def _agent_key(name: Optional[str]) -> str: + """The id an agent has in traces and graphs: its name, or "Agent" when unnamed. + + ``Agent(instructions=...)`` has ``name=None``; without this its events had no + node id and its tools were recorded as ``tool:None/``, so nothing + matched the ``agent:Agent`` node in its graph. + """ + return name or "Agent" + + def set_context_emitter(emitter: "ContextTraceEmitter") -> contextvars.Token: """ Set the context trace emitter for the current async context. @@ -786,7 +796,7 @@ def agent_start( session_id=self._session_id, agent_name=agent_name, data=metadata or {}, - node_id=f"agent:{agent_name}" if agent_name else None, + node_id=f"agent:{_agent_key(agent_name)}", )) def agent_end( @@ -803,7 +813,7 @@ def agent_end( session_id=self._session_id, agent_name=agent_name, data=metadata or {}, - node_id=f"agent:{agent_name}" if agent_name else None, + node_id=f"agent:{_agent_key(agent_name)}", )) def message_added( @@ -853,7 +863,7 @@ def tool_call_start( "tool_name": tool_name, "tool_args": self._redact_dict(tool_args) if tool_args else {}, }, - node_id=f"tool:{agent_name}/{tool_name}", + node_id=f"tool:{_agent_key(agent_name)}/{tool_name}", )) def tool_call_end( @@ -901,7 +911,7 @@ def tool_call_end( "duration_ms": duration_ms, "error": error, }, - node_id=f"tool:{agent_name}/{tool_name}", + node_id=f"tool:{_agent_key(agent_name)}/{tool_name}", )) def _smart_truncate_result(self, result: str, tool_name: str) -> str: @@ -1030,7 +1040,7 @@ def _llm_request_event( tokens_used=tokens_used, tokens_budget=tokens_budget, data=data, - node_id=f"agent:{agent_name}" if agent_name else None, + node_id=f"agent:{_agent_key(agent_name)}", ) def llm_response( @@ -1114,7 +1124,7 @@ def _llm_response_event( completion_tokens=actual_completion_tokens, cost_usd=cost_usd, data=data, - node_id=f"agent:{agent_name}" if agent_name else None, + node_id=f"agent:{_agent_key(agent_name)}", ) def context_snapshot( diff --git a/src/praisonai-agents/praisonaiagents/workflows/diagram.py b/src/praisonai-agents/praisonaiagents/workflows/diagram.py index 09b6c36aac..e06feb190c 100644 --- a/src/praisonai-agents/praisonaiagents/workflows/diagram.py +++ b/src/praisonai-agents/praisonaiagents/workflows/diagram.py @@ -87,7 +87,8 @@ def _tool_name(tool: Any) -> str: def _is_agent(obj: Any) -> bool: - return hasattr(obj, "chat") and isinstance(getattr(obj, "name", None), str) + # An unnamed Agent (``Agent(instructions=...)``) has ``name=None``; it is still an agent. + return hasattr(obj, "chat") and hasattr(obj, "name") # --------------------------------------------------------------------------- @@ -204,8 +205,11 @@ def ops_to_mermaid(self) -> str: def _add_agent(g: _Graph, agent: Any, parent: Optional[str] = None, _seen: Optional[set] = None) -> str: """Add an agent node with its tools and hand-off targets; return its id.""" - name = getattr(agent, "name", None) or type(agent).__name__ - agent_id = f"agent:{name}" + from ..trace.context_events import _agent_key + + name = getattr(agent, "name", None) or type(agent).__name__ # label + key = _agent_key(getattr(agent, "name", None)) # the id the trace uses + agent_id = f"agent:{key}" seen = _seen if _seen is not None else set() if agent_id in seen: return agent_id @@ -227,7 +231,7 @@ def _add_agent(g: _Graph, agent: Any, parent: Optional[str] = None, _seen: Optio tool_name = _tool_name(tool) if tool_name in handoff_tools: continue # drawn as the hand-off edge, not as a tool - tool_id = f"tool:{name}/{tool_name}" + tool_id = f"tool:{key}/{tool_name}" g.node(tool_id, "tool", tool_name, parent=parent) g.edge(agent_id, tool_id, "uses") for target in targets: diff --git a/src/praisonai-agents/tests/unit/agent/test_db_run_lifecycle.py b/src/praisonai-agents/tests/unit/agent/test_db_run_lifecycle.py index 5bdc6fb08b..5ac824f497 100644 --- a/src/praisonai-agents/tests/unit/agent/test_db_run_lifecycle.py +++ b/src/praisonai-agents/tests/unit/agent/test_db_run_lifecycle.py @@ -388,7 +388,8 @@ async def respond(**kwargs): assert db.starts[0]["run_id"] == db.ends[0]["run_id"] assert db.ends[0]["status"] == "completed" assert agent._active_turn_token is None - emitter.agent_end.assert_called_once_with(agent.name) + # The answer came back; only the final save was cancelled. + emitter.agent_end.assert_called_once_with(agent.name, {"status": "completed"}) token.close.assert_called_once() diff --git a/src/praisonai-agents/tests/unit/trace/test_unnamed_agent_trace.py b/src/praisonai-agents/tests/unit/trace/test_unnamed_agent_trace.py new file mode 100644 index 0000000000..f701299270 --- /dev/null +++ b/src/praisonai-agents/tests/unit/trace/test_unnamed_agent_trace.py @@ -0,0 +1,174 @@ +"""Unnamed agents in traces and graphs, agent_end status, and async model-call events. + +Deterministic: the model call is replaced, so nothing reaches a provider. +""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from praisonaiagents.trace.context_events import ( + ContextListSink, + ContextTraceEmitter, + reset_context_emitter, + set_context_emitter, +) + + +@pytest.fixture +def sink(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-not-used") + s = ContextListSink() + token = set_context_emitter(ContextTraceEmitter(sink=s, session_id="t", enabled=True)) + yield s + reset_context_emitter(token) + + +def _of(sink, kind): + return [e for e in sink.get_events() if e.event_type.value == kind] + + +def _fake_completion(text="Solar sails ride sunlight."): + message = SimpleNamespace(content=text, tool_calls=None, role="assistant", reasoning_content=None) + return SimpleNamespace( + choices=[SimpleNamespace(message=message, finish_reason="stop", index=0)], + usage=SimpleNamespace(prompt_tokens=11, completion_tokens=7, total_tokens=18), + model="gpt-4o-mini", + ) + + +def search(query: str) -> str: + """Search the web.""" + return "result" + + +# --- to_graph and ids --------------------------------------------------------- + +def test_to_graph_accepts_an_unnamed_agent(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-not-used") + from praisonaiagents import Agent + from praisonaiagents.workflows.diagram import to_graph + + graph = to_graph(Agent(instructions="Answer briefly.", tools=[search])) + ids = [n["id"] for n in graph["nodes"]] + assert ids[:3] == ["start:begin", "agent:Agent", "tool:Agent/search"] or set(ids) >= { + "start:begin", "agent:Agent", "tool:Agent/search", "end:done"} + assert next(n for n in graph["nodes"] if n["id"] == "agent:Agent")["label"] == "Agent" + + +def test_named_agent_graph_is_unchanged(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-not-used") + from praisonaiagents import Agent + from praisonaiagents.workflows.diagram import to_graph + + ids = [n["id"] for n in to_graph(Agent(name="Writer", instructions="x", tools=[search]))["nodes"]] + assert "agent:Writer" in ids and "tool:Writer/search" in ids + + +def test_emitter_ids_for_an_unnamed_agent(): + s = ContextListSink() + em = ContextTraceEmitter(sink=s, session_id="t", enabled=True) + em.agent_start(None) + em.tool_call_start(None, "search", {"query": "x"}) + em.tool_call_end(None, "search", "ok", 1.0) + em.agent_end(None) + assert [e.node_id for e in s.get_events()] == [ + "agent:Agent", "tool:Agent/search", "tool:Agent/search", "agent:Agent"] + + +def test_unnamed_agent_run_matches_its_graph(sink): + from praisonaiagents import Agent + from praisonaiagents.workflows.diagram import to_graph + + agent = Agent(instructions="Answer briefly.") + with patch.object(agent, "_chat_completion", return_value=_fake_completion()): + agent.chat("Why do solar sails work?") + graph_ids = {n["id"] for n in to_graph(agent)["nodes"]} + for kind in ("agent_start", "agent_end"): + (event,) = _of(sink, kind) + assert event.agent_name is None # the name is not invented on the event + assert event.node_id == "agent:Agent" and event.node_id in graph_ids + + +# --- agent_end status --------------------------------------------------------- + +def test_agent_end_records_completed(sink): + from praisonaiagents import Agent + + agent = Agent(name="Writer", instructions="x") + with patch.object(agent, "_chat_completion", return_value=_fake_completion()): + assert agent.chat("hi") + assert _of(sink, "agent_end")[0].data == {"status": "completed"} + + +def test_agent_end_records_failed(sink): + from praisonaiagents import Agent + + agent = Agent(name="Writer", instructions="x") + with patch.object(agent, "_chat_completion", side_effect=RuntimeError("provider down")): + try: + result = agent.chat("hi") + except Exception: + result = None + assert result is None + assert _of(sink, "agent_end")[0].data == {"status": "failed"} + + +# --- async model calls -------------------------------------------------------- + +def test_astart_records_the_model_call(sink): + from praisonaiagents import Agent + + agent = Agent(name="Writer", instructions="x", llm="gpt-4o-mini") + + async def fake_dispatch(*args, **kwargs): + return _fake_completion() + + with patch.object(agent, "_achat_completion_with_retry_dispatch", side_effect=fake_dispatch): + answer = asyncio.run(agent.astart("Why do solar sails work?")) + assert "sunlight" in str(answer) + (req,) = _of(sink, "llm_request") + (resp,) = _of(sink, "llm_response") + assert req.node_id == resp.node_id == "agent:Writer" + assert resp.prompt_tokens == 11 and resp.completion_tokens == 7 and resp.cost_usd > 0 + assert "sunlight" in (resp.data.get("response_content") or "") + assert _of(sink, "agent_end")[0].data == {"status": "completed"} + + +def test_async_model_error_is_recorded(sink): + from praisonaiagents import Agent + + agent = Agent(name="Writer", instructions="x", llm="gpt-4o-mini") + + async def boom(*args, **kwargs): + raise RuntimeError("rate limited") + + with patch.object(agent, "_achat_completion_with_retry_dispatch", side_effect=boom): + try: + asyncio.run(agent.astart("hi")) + except Exception: + pass + responses = _of(sink, "llm_response") + assert responses and responses[0].data.get("finish_reason") == "error" + assert "rate limited" in (responses[0].data.get("response_content") or "") + + +def test_async_path_records_nothing_when_tracing_is_off(monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test-not-used") + from praisonaiagents import Agent + + s = ContextListSink() + token = set_context_emitter(ContextTraceEmitter(sink=s, session_id="t", enabled=False)) + try: + agent = Agent(name="Writer", instructions="x", llm="gpt-4o-mini") + + async def fake_dispatch(*args, **kwargs): + return _fake_completion() + + with patch.object(agent, "_achat_completion_with_retry_dispatch", side_effect=fake_dispatch) as d: + asyncio.run(agent.astart("hi")) + assert d.called and s.get_events() == [] + finally: + reset_context_emitter(token)