updates

Author: dixitaniket

src/opengradient/agents/og_langchain.py

@@ -1,9 +1,8 @@
 # mypy: ignore-errors
 import asyncio
 import json
-from queue import Queue
-from threading import Thread
-from typing import Any, AsyncIterator, Callable, Dict, Iterator, List, Optional, Sequence, Union, cast
+from enum import Enum
+from typing import Any, AsyncIterator, Awaitable, Callable, Dict, Iterator, List, Optional, Sequence, Union, cast
 
 from langchain_core.callbacks.manager import AsyncCallbackManagerForLLMRun, CallbackManagerForLLMRun
 from langchain_core.language_models.base import LanguageModelInput
@@ -33,8 +32,6 @@
 
 __all__ = ["OpenGradientChatModel"]
 
-_STREAM_END = object()
-
 
 def _extract_content(content: Any) -> str:
     """Normalize content to a plain string.
@@ -122,28 +119,29 @@ def _parse_tool_call_chunk(tool_call: Dict[str, Any], default_index: int) -> Too
     )
 
 
-def _run_coro_sync(coro: Any) -> Any:
+def _run_coro_sync(coro_factory: Callable[[], Awaitable[Any]]) -> Any:
     try:
         asyncio.get_running_loop()
     except RuntimeError:
-        return asyncio.run(coro)
-
-    queue: Queue[Any] = Queue(maxsize=1)
+        return asyncio.run(coro_factory())
 
-    def _runner() -> None:
-        try:
-            queue.put(asyncio.run(coro))
-        except BaseException as exc:  # noqa: BLE001
-            queue.put(exc)
+    raise RuntimeError(
+        "Synchronous LangChain calls cannot run inside an active event loop for this adapter. "
+        "Use `ainvoke`/`astream` instead of `invoke`/`stream`."
+    )
 
-    thread = Thread(target=_runner, daemon=True)
-    thread.start()
-    outcome = queue.get()
-    thread.join()
 
-    if isinstance(outcome, BaseException):
-        raise outcome
-    return outcome
+def _validate_model_string(model: Union[TEE_LLM, str]) -> Union[TEE_LLM, str]:
+    if isinstance(model, Enum):
+        model_str = str(model.value)
+    else:
+        model_str = str(model)
+    if "/" not in model_str:
+        raise ValueError(
+            f"Unsupported model value '{model_str}'. "
+            "Expected provider/model format (for example: 'openai/gpt-5')."
+        )
+    return model
 
 
 class OpenGradientChatModel(BaseChatModel):
@@ -176,6 +174,7 @@ def __init__(
         resolved_model_cid = model_cid or model
         if resolved_model_cid is None:
             raise ValueError("model_cid (or model) is required.")
+        resolved_model_cid = _validate_model_string(resolved_model_cid)
         super().__init__(
             model_cid=resolved_model_cid,
             max_tokens=max_tokens,
@@ -213,7 +212,7 @@ async def aclose(self) -> None:
 
     def close(self) -> None:
         if self._owns_client:
-            _run_coro_sync(self._llm.close())
+            _run_coro_sync(self._llm.close)
 
     def bind_tools(
         self,
@@ -309,9 +308,11 @@ def _build_chat_kwargs(self, sdk_messages: List[Dict[str, Any]], stop: Optional[
         x402_settlement_mode = kwargs.get("x402_settlement_mode", self.x402_settlement_mode)
         if isinstance(x402_settlement_mode, str):
             x402_settlement_mode = x402SettlementMode(x402_settlement_mode)
+        model = kwargs.get("model", self.model_cid)
+        model = _validate_model_string(model)
 
         return {
-            "model": kwargs.get("model", self.model_cid),
+            "model": model,
             "messages": sdk_messages,
             "stop_sequence": stop,
             "max_tokens": kwargs.get("max_tokens", self.max_tokens),
@@ -346,7 +347,7 @@ def _generate(
     ) -> ChatResult:
         sdk_messages = self._convert_messages_to_sdk(messages)
         chat_kwargs = self._build_chat_kwargs(sdk_messages, stop, stream=False, **kwargs)
-        chat_output = _run_coro_sync(self._llm.chat(**chat_kwargs))
+        chat_output = _run_coro_sync(lambda: self._llm.chat(**chat_kwargs))
         if not isinstance(chat_output, TextGenerationOutput):
             raise RuntimeError("Expected non-streaming chat output but received streaming generator.")
         return self._build_chat_result(chat_output)
@@ -374,33 +375,30 @@ def _stream(
     ) -> Iterator[ChatGenerationChunk]:
         sdk_messages = self._convert_messages_to_sdk(messages)
         chat_kwargs = self._build_chat_kwargs(sdk_messages, stop, stream=True, **kwargs)
-        queue: Queue[Any] = Queue()
-
-        def _runner() -> None:
-            async def _run() -> None:
-                stream = await self._llm.chat(**chat_kwargs)
-                async for chunk in cast(AsyncIterator[StreamChunk], stream):
-                    queue.put(self._stream_chunk_to_generation(chunk))
-
-            try:
-                asyncio.run(_run())
-            except BaseException as exc:  # noqa: BLE001
-                queue.put(exc)
-            finally:
-                queue.put(_STREAM_END)
-
-        thread = Thread(target=_runner, daemon=True)
-        thread.start()
-
-        while True:
-            item = queue.get()
-            if item is _STREAM_END:
-                break
-            if isinstance(item, BaseException):
-                raise item
-            yield cast(ChatGenerationChunk, item)
-
-        thread.join()
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            pass
+        else:
+            raise RuntimeError(
+                "Synchronous stream cannot run inside an active event loop for this adapter. "
+                "Use `astream` instead."
+            )
+
+        loop = asyncio.new_event_loop()
+        try:
+            stream = loop.run_until_complete(self._llm.chat(**chat_kwargs))
+            stream_iter = cast(AsyncIterator[StreamChunk], stream)
+
+            while True:
+                try:
+                    chunk = loop.run_until_complete(stream_iter.__anext__())
+                except StopAsyncIteration:
+                    break
+                yield self._stream_chunk_to_generation(chunk)
+        finally:
+            loop.run_until_complete(loop.shutdown_asyncgens())
+            loop.close()
 
     async def _astream(
         self,

tests/langchain_adapter_test.py

@@ -64,6 +64,10 @@ def test_initialization_without_private_key_or_client_raises(self):
         with pytest.raises(ValueError, match="private_key is required"):
             OpenGradientChatModel(private_key=None, model_cid=TEE_LLM.GPT_5)
 
+    def test_initialization_with_invalid_model_string_raises(self):
+        with pytest.raises(ValueError, match="provider/model format"):
+            OpenGradientChatModel(private_key="0x" + "a" * 64, model_cid="gpt-5")
+
     def test_identifying_params(self, model):
         """Test _identifying_params returns model name."""
         assert model._identifying_params == {"model_name": TEE_LLM.GPT_5, "temperature": 0.0, "max_tokens": 300}
@@ -168,6 +172,24 @@ def test_empty_chat_output(self, model, mock_llm_client):
 
         assert result.generations[0].message.content == ""
 
+    def test_generate_with_invalid_model_kwarg_raises(self, model):
+        with pytest.raises(ValueError, match="provider/model format"):
+            model._generate([HumanMessage(content="Hi")], model="gpt-5")
+
+    def test_sync_generate_inside_running_loop_raises(self, model):
+        async def run_test():
+            with pytest.raises(RuntimeError, match="Use `ainvoke`/`astream`"):
+                model._generate([HumanMessage(content="Hi")])
+
+        asyncio.run(run_test())
+
+    def test_sync_stream_inside_running_loop_raises(self, model):
+        async def run_test():
+            with pytest.raises(RuntimeError, match="Use `astream`"):
+                next(model._stream([HumanMessage(content="Hi")]))
+
+        asyncio.run(run_test())
+
 
 class TestMessageConversion:
     def test_converts_all_message_types(self, model, mock_llm_client):