Fix Singleton Tuple Config Handling In _ConfigurableModel.Batch_as_completed
`_ConfigurableModel.batch_as_completed` and `abatch_as_completed` incorrectly reject singleton tuple configs, causing an AttributeError despite the annotated `Sequence[RunnableConfig]` contract. The code checks only for `list`, not the abstract `Sequence`, so non-list sequences are passed to `ensure_config` unnormalized.
The single-config branch unwraps only a concrete `list` using `isinstance(config, list)`, while the parameter is declared as `Sequence[RunnableConfig]`. Any other Sequence (e.g., tuple) skips the unwrap and is passed to `ensure_config`, which requires a mapping, resulting in AttributeError.
Run the following Python snippet with langchain 1.3.15 / langchain-core 1.6.0:
```python
from unittest.mock import patch
from langchain_core.language_models.fake_chat_models import FakeChatModel
from langchain.chat_models import init_chat_model
model = init_chat_model()
def run(config):
with patch("langchain.chat_models.base._init_chat_model_helper", return_value=FakeChatModel()):
return [output for _, output in model.batch_as_completed(["hello"], config=config)]
print("list [{...}] :", run([{"tags": ["test"]}]))
print("tuple ({...},):", run(({"tags": ["test"]},)))
```
The tuple form raises `AttributeError: 'tuple' object has no attribute 'items'`.
Fixing Code Block
from collections.abc import Sequence
from typing import Any, cast, AsyncIterator, Iterator
# Inside _ConfigurableModel class, replace existing methods with these corrected versions.
def batch_as_completed(
self,
inputs: list[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any,
) -> Iterator[tuple[int, Output]]:
"""Overridden to unwrap any Sequence, not just list, for singleton config."""
if isinstance(config, Sequence):
config = config[0]
yield from self._model(cast("RunnableConfig", config)).batch_as_completed(
inputs,
config=config,
return_exceptions=return_exceptions,
**kwargs,
)
async def abatch_as_completed(
self,
inputs: list[Input],
config: RunnableConfig | Sequence[RunnableConfig] | None = None,
*,
return_exceptions: bool = False,
**kwargs: Any,
) -> AsyncIterator[tuple[int, Output]]:
"""Overridden to unwrap any Sequence, not just list, for singleton config."""
if isinstance(config, Sequence):
config = config[0]
async for output in self._model(cast("RunnableConfig", config)).abatch_as_completed(
inputs,
config=config,
return_exceptions=return_exceptions,
**kwargs,
):
yield output
Replace the concrete `list` check with the abstract `Sequence` check from `collections.abc`. This aligns the unwrapping logic with the declared type annotation and matches the behavior of the base `Runnable` implementation, which already handles any `Sequence` correctly via `get_config_list`. The import of `Sequence` (and any missing typing imports) is included for completeness.
Edge Case Audit
The fix preserves existing behavior for lists, but note that any non-list Sequence with length > 1 will still only use the first element, ignoring the rest. This matches the previous list behavior but may be unexpected for callers passing multiple configs to these overridden methods. No new threading or concurrency issues are introduced. If a regression occurs, roll back by changing `isinstance(config, Sequence)` back to `isinstance(config, list)`. Test on Python 3.9+ where `collections.abc.Sequence` is stable.
