strict

libs/core/langchain_core/runnables/base.py

@@ -25,6 +25,7 @@ from typing import (
     List,
     Mapping,
     Optional,
+    Protocol,
     Sequence,
     Set,
     Tuple,
@@ -5519,12 +5520,36 @@ class RunnableBinding(RunnableBindingBase[Input, Output]):
         return attr
 
 
+class _RunnableCallableSync(Protocol[Input, Output]):
+    def __call__(self, __in: Input, *, config: RunnableConfig) -> Output: ...
+
+
+class _RunnableCallableAsync(Protocol[Input, Output]):
+    def __call__(self, __in: Input, *, config: RunnableConfig) -> Awaitable[Output]: ...
+
+
+class _RunnableCallableIterator(Protocol[Input, Output]):
+    def __call__(
+        self, __in: Iterator[Input], *, config: RunnableConfig
+    ) -> Iterator[Output]: ...
+
+
+class _RunnableCallableAsyncIterator(Protocol[Input, Output]):
+    def __call__(
+        self, __in: AsyncIterator[Input], *, config: RunnableConfig
+    ) -> AsyncIterator[Output]: ...
+
+
 RunnableLike = Union[
     Runnable[Input, Output],
     Callable[[Input], Output],
     Callable[[Input], Awaitable[Output]],
     Callable[[Iterator[Input]], Iterator[Output]],
     Callable[[AsyncIterator[Input]], AsyncIterator[Output]],
+    _RunnableCallableSync[Input, Output],
+    _RunnableCallableAsync[Input, Output],
+    _RunnableCallableIterator[Input, Output],
+    _RunnableCallableAsyncIterator[Input, Output],
     Mapping[str, Any],
 ]
 

libs/core/pyproject.toml

@@ -14,10 +14,16 @@ repository = "https://github.com/langchain-ai/langchain"
 [tool.mypy]
 disallow_untyped_defs = "True"
 exclude = [ "notebooks", "examples", "example_data", "langchain_core/pydantic", "tests/unit_tests/utils/test_function_calling.py",]
+
 [[tool.mypy.overrides]]
 module = [ "numpy", "pytest",]
 ignore_missing_imports = true
 
+
+[[tool.mypy.overrides]]
+module = "tests.unit_tests.runnables.test_lint_base.py"
+strict = true
+
 [tool.poetry.urls]
 "Source Code" = "https://github.com/langchain-ai/langchain/tree/master/libs/core"
 "Release Notes" = "https://github.com/langchain-ai/langchain/releases?q=tag%3A%22langchain-core%3D%3D0%22&expanded=true"

libs/core/tests/unit_tests/runnables/test_lint_base.py

@@ -0,0 +1,50 @@
+from typing import AsyncIterator, Iterator
+
+from langchain_core.runnables.base import (
+    coerce_to_runnable,
+)
+from langchain_core.runnables.config import RunnableConfig
+
+
+async def test_coerce_to_runnable():
+    def _foo(input: str, config: RunnableConfig) -> str:
+        assert config
+        return input
+
+    assert coerce_to_runnable(_foo).invoke("foo") == "foo"
+
+    async def _afoo(input: str, config: RunnableConfig) -> str:
+        assert config
+        return input
+
+    assert (await coerce_to_runnable(_afoo).ainvoke("foo")) == "foo"
+
+    def _foo_iter(input: Iterator[str], config: RunnableConfig) -> Iterator[str]:
+        assert config
+        for batch in input:
+            for item in batch:
+                yield f"foo_{item}"
+
+    assert list(coerce_to_runnable(_foo_iter).stream(["bar", "baz"])) == [
+        "foo_bar",
+        "foo_baz",
+    ]
+
+    async def _afoo_iter(
+        input: AsyncIterator[str], config: RunnableConfig
+    ) -> AsyncIterator[str]:
+        assert config
+        async for batch in input:
+            async for item in batch:
+                yield f"afoo_{item}"
+
+    async def async_generator(items):
+        for item in items:
+            yield item
+
+    result = []
+    async for item in coerce_to_runnable(_afoo_iter).astream(
+        async_generator(["bar", "baz"])
+    ):
+        result.append(item)
+    assert result == ["afoo_bar", "afoo_baz"]