|
|
|
@ -1050,12 +1050,19 @@ class Runnable(Generic[Input, Output], ABC):
|
|
|
|
|
|
|
|
|
|
for chunk in input:
|
|
|
|
|
if not got_first_val:
|
|
|
|
|
final = chunk
|
|
|
|
|
final = _adapt_first_streaming_chunk(chunk) # type: ignore
|
|
|
|
|
got_first_val = True
|
|
|
|
|
else:
|
|
|
|
|
# Make a best effort to gather, for any type that supports `+`
|
|
|
|
|
# This method should throw an error if gathering fails.
|
|
|
|
|
final = final + chunk # type: ignore[operator]
|
|
|
|
|
try:
|
|
|
|
|
final = final + chunk # type: ignore[operator]
|
|
|
|
|
except TypeError:
|
|
|
|
|
raise TypeError(
|
|
|
|
|
f"Failed while trying to add together "
|
|
|
|
|
f"type {type(final)} and {type(chunk)}."
|
|
|
|
|
f"These types should be addable for transform to work."
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
if got_first_val:
|
|
|
|
|
yield from self.stream(final, config, **kwargs)
|
|
|
|
@ -1076,12 +1083,19 @@ class Runnable(Generic[Input, Output], ABC):
|
|
|
|
|
|
|
|
|
|
async for chunk in input:
|
|
|
|
|
if not got_first_val:
|
|
|
|
|
final = chunk
|
|
|
|
|
final = _adapt_first_streaming_chunk(chunk) # type: ignore
|
|
|
|
|
got_first_val = True
|
|
|
|
|
else:
|
|
|
|
|
# Make a best effort to gather, for any type that supports `+`
|
|
|
|
|
# This method should throw an error if gathering fails.
|
|
|
|
|
final = final + chunk # type: ignore[operator]
|
|
|
|
|
try:
|
|
|
|
|
final = final + chunk # type: ignore[operator]
|
|
|
|
|
except TypeError:
|
|
|
|
|
raise TypeError(
|
|
|
|
|
f"Failed while trying to add together "
|
|
|
|
|
f"type {type(final)} and {type(chunk)}."
|
|
|
|
|
f"These types should be addable for atransform to work."
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
if got_first_val:
|
|
|
|
|
async for output in self.astream(final, config, **kwargs):
|
|
|
|
@ -3560,7 +3574,7 @@ class RunnableLambda(Runnable[Input, Output]):
|
|
|
|
|
final: Optional[Input] = None
|
|
|
|
|
for ichunk in input:
|
|
|
|
|
if final is None:
|
|
|
|
|
final = ichunk
|
|
|
|
|
final = _adapt_first_streaming_chunk(ichunk) # type: ignore
|
|
|
|
|
else:
|
|
|
|
|
try:
|
|
|
|
|
final = final + ichunk # type: ignore[operator]
|
|
|
|
@ -3644,7 +3658,7 @@ class RunnableLambda(Runnable[Input, Output]):
|
|
|
|
|
final: Optional[Input] = None
|
|
|
|
|
async for ichunk in input:
|
|
|
|
|
if final is None:
|
|
|
|
|
final = ichunk
|
|
|
|
|
final = _adapt_first_streaming_chunk(ichunk)
|
|
|
|
|
else:
|
|
|
|
|
try:
|
|
|
|
|
final = final + ichunk # type: ignore[operator]
|
|
|
|
@ -4445,3 +4459,11 @@ def chain(
|
|
|
|
|
yield chunk
|
|
|
|
|
"""
|
|
|
|
|
return RunnableLambda(func)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _adapt_first_streaming_chunk(chunk: Any) -> Any:
|
|
|
|
|
"""This might transform the first chunk of a stream into an AddableDict."""
|
|
|
|
|
if isinstance(chunk, dict) and not isinstance(chunk, AddableDict):
|
|
|
|
|
return AddableDict(chunk)
|
|
|
|
|
else:
|
|
|
|
|
return chunk
|
|
|
|
|