diff --git a/src/voice_assistant/agent/deep_agent.py b/src/voice_assistant/agent/deep_agent.py index 9aba22c..b57c757 100644 --- a/src/voice_assistant/agent/deep_agent.py +++ b/src/voice_assistant/agent/deep_agent.py @@ -1,4 +1,3 @@ - """ Deep Agents implementation of the AgentVoice Agent port. @@ -24,6 +23,7 @@ from contextlib import aclosing from deepagents import create_deep_agent from langchain_core.messages import HumanMessage +from langchain_core.runnables import RunnableConfig from langchain_oci import ChatOCIGenAI from langgraph.checkpoint.memory import MemorySaver @@ -95,12 +95,11 @@ class DeepAgentsAdapter(StreamingAgentAdapter): self._stream_event_timeout = float(timeout_seconds) - self.model = ChatOCIGenAI( + self._model = ChatOCIGenAI( model_id=model_id, compartment_id=compartment_id, service_endpoint=( - f"https://inference.generativeai." - f"{region}.oci.oraclecloud.com" + f"https://inference.generativeai.{region}.oci.oraclecloud.com" ), auth_type=auth_type, auth_profile=auth_profile, @@ -112,18 +111,19 @@ class DeepAgentsAdapter(StreamingAgentAdapter): }, ) - self.checkpointer = MemorySaver() + self._checkpointer = MemorySaver() - self.agent = create_deep_agent( - model=self.model, + self._graph = create_deep_agent( + model=self._model, tools=[], system_prompt=system_prompt, - checkpointer=self.checkpointer, + checkpointer=self._checkpointer, ) self._closed = False - def _config(self, session_id: str) -> dict: + @staticmethod + def _session_config(session_id: str) -> RunnableConfig: if not session_id: raise ValueError("session_id is required.") @@ -133,34 +133,6 @@ class DeepAgentsAdapter(StreamingAgentAdapter): } } - async def process( - self, - session_id: str, - turn_id: int, - text: str, - ) -> str: - parts: list[str] = [] - - async with aclosing( - self.stream( - session_id=session_id, - turn_id=turn_id, - text=text, - ) - ) as chunks: - async for chunk in chunks: - parts.append(chunk) - - response = "".join(parts).strip() - - if not response: - raise RuntimeError( - "Deep Agent returned an empty response." - ) - - return response - - async def stream( self, session_id: str, @@ -173,9 +145,9 @@ class DeepAgentsAdapter(StreamingAgentAdapter): if not text.strip(): raise ValueError("Agent input text cannot be empty.") - config = self._config(session_id) + config = self._session_config(session_id) - events = self.agent.astream( + events = self._graph.astream( {"messages": [HumanMessage(content=text)]}, config=config, stream_mode="messages", @@ -200,10 +172,9 @@ class DeepAgentsAdapter(StreamingAgentAdapter): for fragment in _text_fragments(message.content): yield fragment - async def aclose(self) -> None: if self._closed: return - await self.model.aclose() + await self._model.aclose() self._closed = True diff --git a/tests/test_deep_agent.py b/tests/test_deep_agent.py index 7b32a0c..824b42a 100644 --- a/tests/test_deep_agent.py +++ b/tests/test_deep_agent.py @@ -18,20 +18,20 @@ def test_deep_agent_session_config(): adapter = object.__new__(DeepAgentsAdapter) - assert adapter._config("session-a") == { + assert adapter._session_config("session-a") == { "configurable": { "thread_id": "session-a", } } - assert adapter._config("session-b") == { + assert adapter._session_config("session-b") == { "configurable": { "thread_id": "session-b", } } with pytest.raises(ValueError): - adapter._config("") + adapter._session_config("") @@ -41,8 +41,8 @@ def test_deep_agent_cleanup_is_idempotent(): from voice_assistant.agent.deep_agent import DeepAgentsAdapter adapter = object.__new__(DeepAgentsAdapter) - adapter.model = MagicMock() - adapter.model.aclose = AsyncMock() + adapter._model = MagicMock() + adapter._model.aclose = AsyncMock() adapter._closed = False async def scenario(): @@ -51,7 +51,7 @@ def test_deep_agent_cleanup_is_idempotent(): asyncio.run(scenario()) - adapter.model.aclose.assert_awaited_once() + adapter._model.aclose.assert_awaited_once() assert adapter._closed is True @@ -90,7 +90,7 @@ def test_deep_agent_stream_can_be_cancelled(): self.content = content self.tool_call_chunks = [] - adapter.agent = FakeAgent() + adapter._graph = FakeAgent() async def consume(): async for _ in adapter.stream( @@ -389,6 +389,6 @@ def test_deep_agent_factory_configures_authentication(monkeypatch, options, expe model_factory.assert_called_once() kwargs = model_factory.call_args.kwargs assert (kwargs["auth_type"], kwargs["auth_profile"], kwargs["auth_file_location"]) == expected_auth - assert adapter.model is model_factory.return_value - assert agent_factory.call_args.kwargs["model"] is adapter.model + assert adapter._model is model_factory.return_value + assert agent_factory.call_args.kwargs["model"] is adapter._model diff --git a/tests/test_deep_agent_stream.py b/tests/test_deep_agent_stream.py index ec43dbf..0c9367b 100644 --- a/tests/test_deep_agent_stream.py +++ b/tests/test_deep_agent_stream.py @@ -11,7 +11,7 @@ def make_adapter(events, timeout=0.05): adapter = object.__new__(DeepAgentsAdapter) adapter._closed = False adapter._stream_event_timeout = timeout - adapter.agent = SimpleNamespace(astream=lambda *args, **kwargs: events) + adapter._graph = SimpleNamespace(astream=lambda *args, **kwargs: events) return adapter @@ -138,9 +138,42 @@ def test_adapter_streams_real_deepagents_graph_with_fake_model(): adapter = object.__new__(DeepAgentsAdapter) adapter._closed = False adapter._stream_event_timeout = 2.0 - adapter.agent = create_deep_agent( + adapter._graph = create_deep_agent( model=Model(responses=["Olá Lara."]), tools=[], checkpointer=MemorySaver(), ) response = await adapter.process("session", 1, "Diga olá") assert response == "Olá Lara." asyncio.run(scenario()) + + +def test_process_rejects_empty_text_response_and_closes_stream(): + async def scenario(): + closed = asyncio.Event() + + async def source(): + try: + yield event([{"type": "reasoning", "text": "Interno"}]) + finally: + closed.set() + + with pytest.raises(RuntimeError, match="empty streaming response"): + await make_adapter(source()).process("session", 1, "Olá") + assert closed.is_set() + asyncio.run(scenario()) + + +def test_process_propagates_provider_failure_and_closes_stream(): + async def scenario(): + closed = asyncio.Event() + + async def source(): + try: + yield event("Texto parcial") + raise ValueError("provider failure") + finally: + closed.set() + + with pytest.raises(ValueError, match="provider failure"): + await make_adapter(source()).process("session", 1, "Olá") + assert closed.is_set() + asyncio.run(scenario())