From fbbe47d90d522834aff8ab5d6ee12df8657dce77 Mon Sep 17 00:00:00 2001 From: alex alves Date: Thu, 5 Nov 2026 16:14:38 -0300 Subject: [PATCH] =?UTF-8?q?fix:=20corrige=20autentica=C3=A7=C3=A3o=20e=20c?= =?UTF-8?q?iclo=20de=20streaming=20do=20DeepAgents?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 16 +++ src/voice_assistant/agent/deep_agent.py | 78 +++++++++---- tests/test_deep_agent.py | 51 +++++++++ tests/test_deep_agent_stream.py | 146 ++++++++++++++++++++++++ 4 files changed, 270 insertions(+), 21 deletions(-) create mode 100644 tests/test_deep_agent_stream.py diff --git a/README.md b/README.md index 7ca06e3..fc7c218 100644 --- a/README.md +++ b/README.md @@ -253,6 +253,22 @@ flowchart TB **Design principle:** transport, orchestration, lifecycle, and provider-specific integrations remain separate concerns. Provider independence is established through contracts and factories, while streaming and cancellation behavior are validated at their respective integration boundaries. +### DeepAgents: autenticação e streaming + +No Mac, `API_KEY` lê o perfil OCI e a chave indicados pelo arquivo de configuração. +Para uma sessão OCI CLI, use `SECURITY_TOKEN` e um perfil com token válido. +O DeepAgents também aceita `INSTANCE_PRINCIPAL` e `RESOURCE_PRINCIPAL` em +ambientes OCI compatíveis. O padrão do construtor permanece `INSTANCE_PRINCIPAL` +quando `auth_type` é omitido. Credenciais e permissões precisam ser válidas. +Dentro do container, monte o arquivo OCI e as chaves/tokens necessários somente +para leitura, com caminhos acessíveis ao usuário do processo, e ajuste +`auth_file_location`; `~` refere-se ao usuário do container, não ao do Mac. +Não há integração MCP/RAG automática nesse bloco; a implementação usa `tools=[]`. +`agent.options.timeout_seconds` limita a espera por cada próximo evento do +DeepAgents (inclusive eventos internos), sem incluir o tempo em que o consumidor +está processando um fragmento já entregue. Não é prazo total da resposta nem +limite de inicialização/autenticação. A expiração fecha o stream e propaga erro. + ## Quickstart ### Requirements diff --git a/src/voice_assistant/agent/deep_agent.py b/src/voice_assistant/agent/deep_agent.py index ecdb672..9aba22c 100644 --- a/src/voice_assistant/agent/deep_agent.py +++ b/src/voice_assistant/agent/deep_agent.py @@ -16,7 +16,10 @@ Does not manage: from __future__ import annotations -from collections.abc import AsyncGenerator +import asyncio +import math +from collections.abc import AsyncGenerator, Iterator +from typing import Any from contextlib import aclosing from deepagents import create_deep_agent @@ -41,6 +44,23 @@ de ferramentas que não foram executadas. """ +def _text_fragments(content: str | list[str | dict[str, Any]]) -> Iterator[str]: + """Extract text without exposing reasoning, tools or other content blocks.""" + if isinstance(content, str): + if content: + yield content + return + + for block in content: + if isinstance(block, str): + if block: + yield block + elif isinstance(block, dict) and block.get("type") == "text": + text = block.get("text") + if isinstance(text, str) and text: + yield text + + class DeepAgentsAdapter(StreamingAgentAdapter): def __init__( self, @@ -51,6 +71,10 @@ class DeepAgentsAdapter(StreamingAgentAdapter): max_completion_tokens: int, timeout_seconds: float, system_prompt: str = DEFAULT_SYSTEM_PROMPT, + *, + auth_type: str = "INSTANCE_PRINCIPAL", + auth_profile: str = "DEFAULT", + auth_file_location: str = "~/.oci/config", ) -> None: if not region: raise ValueError("OCI region is required.") @@ -61,8 +85,15 @@ class DeepAgentsAdapter(StreamingAgentAdapter): if not model_id: raise ValueError("OCI model_id is required.") - if timeout_seconds <= 0: - raise ValueError("timeout_seconds must be positive.") + if ( + isinstance(timeout_seconds, bool) + or not isinstance(timeout_seconds, (int, float)) + or not math.isfinite(timeout_seconds) + or timeout_seconds <= 0 + ): + raise ValueError("timeout_seconds must be finite and positive.") + + self._stream_event_timeout = float(timeout_seconds) self.model = ChatOCIGenAI( model_id=model_id, @@ -71,7 +102,9 @@ class DeepAgentsAdapter(StreamingAgentAdapter): f"https://inference.generativeai." f"{region}.oci.oraclecloud.com" ), - auth_type="INSTANCE_PRINCIPAL", + auth_type=auth_type, + auth_profile=auth_profile, + auth_file_location=auth_file_location, is_stream=True, model_kwargs={ "temperature": temperature, @@ -142,27 +175,30 @@ class DeepAgentsAdapter(StreamingAgentAdapter): config = self._config(session_id) - async for message, metadata in self.agent.astream( - { - "messages": [ - HumanMessage(content=text) - ] - }, + events = self.agent.astream( + {"messages": [HumanMessage(content=text)]}, config=config, stream_mode="messages", - ): - # Ignore tools and subagent outputs. - if metadata.get("langgraph_node") != "model": - continue + ) + async with aclosing(events): + while True: + try: + async with asyncio.timeout(self._stream_event_timeout): + message, metadata = await anext(events) + except StopAsyncIteration: + break + except TimeoutError as exc: + raise TimeoutError( + "Deep Agent timed out waiting for the next stream event." + ) from exc - # Ignore tool-call fragments. - if getattr(message, "tool_call_chunks", None): - continue + if metadata.get("langgraph_node") != "model": + continue + if getattr(message, "tool_call_chunks", None): + continue - content = message.content - - if isinstance(content, str) and content: - yield content + for fragment in _text_fragments(message.content): + yield fragment async def aclose(self) -> None: diff --git a/tests/test_deep_agent.py b/tests/test_deep_agent.py index a898a86..7b32a0c 100644 --- a/tests/test_deep_agent.py +++ b/tests/test_deep_agent.py @@ -64,6 +64,7 @@ def test_deep_agent_stream_can_be_cancelled(): async def scenario(): adapter = object.__new__(DeepAgentsAdapter) adapter._closed = False + adapter._stream_event_timeout = 1.0 started = asyncio.Event() cancelled = asyncio.Event() @@ -341,3 +342,53 @@ def test_deep_agent_checkpoint_survives_cancellation(): started = None asyncio.run(scenario()) +@pytest.mark.parametrize( + ("options", "expected_auth"), + [ + ({}, ("INSTANCE_PRINCIPAL", "DEFAULT", "~/.oci/config")), + ( + {"auth_type": "API_KEY", "auth_profile": "LOCAL", "auth_file_location": "/config/oci"}, + ("API_KEY", "LOCAL", "/config/oci"), + ), + ( + {"auth_type": "SECURITY_TOKEN", "auth_profile": "SESSION"}, + ("SECURITY_TOKEN", "SESSION", "~/.oci/config"), + ), + ({"auth_type": "RESOURCE_PRINCIPAL"}, ("RESOURCE_PRINCIPAL", "DEFAULT", "~/.oci/config")), + ], +) +def test_deep_agent_factory_configures_authentication(monkeypatch, options, expected_auth): + import voice_assistant.agent.deep_agent as module + from voice_assistant.config.loader import load_provider_catalog + from voice_assistant.config.models import AppConfig + from voice_assistant.providers import create_agent_adapter + + model_factory = MagicMock() + agent_factory = MagicMock() + monkeypatch.setattr(module, "ChatOCIGenAI", model_factory) + monkeypatch.setattr(module, "create_deep_agent", agent_factory) + config = AppConfig.model_validate({ + "stt": {"provider": "nvidia"}, + "tts": {"provider": "voxcpm"}, + "agent": { + "provider": "deepagents", + "options": { + "region": "sa-vinhedo-1", + "compartment_id": "test-compartment", + "model_id": "test-model", + "temperature": 0.3, + "max_completion_tokens": 256, + "timeout_seconds": 60.0, + **options, + }, + }, + }) + + adapter = create_agent_adapter(config=config, catalog=load_provider_catalog()) + + 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 + diff --git a/tests/test_deep_agent_stream.py b/tests/test_deep_agent_stream.py new file mode 100644 index 0000000..ec43dbf --- /dev/null +++ b/tests/test_deep_agent_stream.py @@ -0,0 +1,146 @@ +import asyncio +from contextlib import aclosing +from types import SimpleNamespace + +import pytest + +from voice_assistant.agent.deep_agent import DeepAgentsAdapter + + +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) + return adapter + + +def event(content, node="model", tool_calls=None): + return SimpleNamespace(content=content, tool_call_chunks=tool_calls or []), {"langgraph_node": node} + + +@pytest.mark.parametrize( + ("content", "expected"), + [ + ("Olá ", ["Olá "]), + ([{"type": "text", "text": "Olá "}, {"type": "text", "text": "Lara."}], ["Olá ", "Lara."]), + (["Olá", " ", {"type": "text", "text": "Lara."}], ["Olá", " ", "Lara."]), + ([{"type": "reasoning", "text": "Interno"}, {"type": "image", "text": "Ignorar"}, {"type": "text", "text": "Resposta"}], ["Resposta"]), + ([{"type": "text"}, {"type": "text", "text": None}, {"type": "text", "text": ""}], []), + ], +) +def test_stream_extracts_only_text_blocks(content, expected): + async def scenario(): + async def source(): + yield event(content) + assert [part async for part in make_adapter(source()).stream("session", 1, "Olá")] == expected + asyncio.run(scenario()) + + +def test_stream_filters_non_model_events_and_tool_calls(): + async def scenario(): + async def source(): + yield event("Resultado interno", node="tools") + yield event("Chamada interna", tool_calls=[{"id": "call-1"}]) + yield event("Resposta final") + assert [part async for part in make_adapter(source()).stream("session", 1, "Olá")] == ["Resposta final"] + asyncio.run(scenario()) + + +@pytest.mark.parametrize("finish", ["normal", "early_close", "cancel", "error", "timeout"]) +def test_stream_closes_owned_generator(finish): + async def scenario(): + closed = asyncio.Event() + waiting = asyncio.Event() + + async def source(): + try: + yield event("Olá") + if finish == "error": + raise ValueError("provider failure") + if finish in ("cancel", "timeout"): + waiting.set() + await asyncio.Event().wait() + finally: + closed.set() + + stream = make_adapter(source(), timeout=0.1).stream("session", 1, "Olá") + assert await anext(stream) == "Olá" + if finish == "normal": + with pytest.raises(StopAsyncIteration): + await anext(stream) + elif finish == "early_close": + await stream.aclose() + elif finish == "cancel": + task = asyncio.create_task(anext(stream)) + await asyncio.wait_for(waiting.wait(), 1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + elif finish == "error": + with pytest.raises(ValueError, match="provider failure"): + await anext(stream) + else: + with pytest.raises(TimeoutError, match="next stream event"): + await anext(stream) + assert closed.is_set() + asyncio.run(scenario()) + + +def test_timeout_does_not_cancel_consumer_between_fragments(): + async def scenario(): + async def source(): + yield event("Primeiro ") + yield event("segundo") + async with aclosing(make_adapter(source(), timeout=0.01).stream("session", 1, "Olá")) as stream: + assert await anext(stream) == "Primeiro " + await asyncio.sleep(0.04) + assert await anext(stream) == "segundo" + asyncio.run(scenario()) + + +def test_timeout_is_renewed_for_each_next_event(): + async def scenario(): + async def source(): + for token in ("a", "b", "c", "d"): + await asyncio.sleep(0.02) + yield event(token) + adapter = make_adapter(source(), timeout=0.06) + assert "".join([part async for part in adapter.stream("session", 1, "Olá")]) == "abcd" + asyncio.run(scenario()) + + +@pytest.mark.parametrize("timeout", [0, -1, float("inf"), float("nan"), True, "60", None]) +def test_invalid_timeout_fails_before_model_construction(monkeypatch, timeout): + import voice_assistant.agent.deep_agent as module + + def unexpected_model(**kwargs): + pytest.fail("Invalid configuration must not create a model") + + monkeypatch.setattr(module, "ChatOCIGenAI", unexpected_model) + with pytest.raises(ValueError, match="finite and positive"): + DeepAgentsAdapter( + region="sa-vinhedo-1", compartment_id="test", model_id="test", + temperature=0.3, max_completion_tokens=256, timeout_seconds=timeout, + ) + + +def test_adapter_streams_real_deepagents_graph_with_fake_model(): + from deepagents import create_deep_agent + from langchain_core.language_models import FakeListChatModel + from langgraph.checkpoint.memory import MemorySaver + + class Model(FakeListChatModel): + def bind_tools(self, tools, *, tool_choice=None, **kwargs): + return self + + async def scenario(): + adapter = object.__new__(DeepAgentsAdapter) + adapter._closed = False + adapter._stream_event_timeout = 2.0 + adapter.agent = 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())