Skip to content

Commit

Permalink
core[patch]: add response kwarg to on_llm_error
Browse files Browse the repository at this point in the history
# Dependencies
None

# Twitter handle
@HKydlicek

---------

Co-authored-by: Erick Friis <erick@langchain.dev>
  • Loading branch information
hynky1999 and efriis committed Dec 4, 2023
1 parent 1750cc4 commit aa8ae31
Show file tree
Hide file tree
Showing 9 changed files with 172 additions and 25 deletions.
16 changes: 14 additions & 2 deletions libs/core/langchain_core/callbacks/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,13 @@ def on_llm_error(
parent_run_id: Optional[UUID] = None,
**kwargs: Any,
) -> Any:
"""Run when LLM errors."""
"""Run when LLM errors.
Args:
error (BaseException): The error that occurred.
kwargs (Any): Additional keyword arguments.
- response (LLMResult): The response which was generated before
the error occurred.
"""


class ChainManagerMixin:
Expand Down Expand Up @@ -351,7 +357,13 @@ async def on_llm_error(
tags: Optional[List[str]] = None,
**kwargs: Any,
) -> None:
"""Run when LLM errors."""
"""Run when LLM errors.
Args:
error (BaseException): The error that occurred.
kwargs (Any): Additional keyword arguments.
- response (LLMResult): The response which was generated before
the error occurred.
"""

async def on_chain_start(
self,
Expand Down
9 changes: 9 additions & 0 deletions libs/core/langchain_core/callbacks/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -623,6 +623,9 @@ def on_llm_error(
Args:
error (Exception or KeyboardInterrupt): The error.
kwargs (Any): Additional keyword arguments.
- response (LLMResult): The response which was generated before
the error occurred.
"""
handle_event(
self.handlers,
Expand Down Expand Up @@ -689,6 +692,12 @@ async def on_llm_error(
Args:
error (Exception or KeyboardInterrupt): The error.
kwargs (Any): Additional keyword arguments.
- response (LLMResult): The response which was generated before
the error occurred.
"""
await ahandle_event(
self.handlers,
Expand Down
28 changes: 19 additions & 9 deletions libs/core/langchain_core/language_models/chat_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -223,8 +223,8 @@ def stream(
name=config.get("run_name"),
batch_size=1,
)
generation: Optional[ChatGenerationChunk] = None
try:
generation: Optional[ChatGenerationChunk] = None
for chunk in self._stream(
messages, stop=stop, run_manager=run_manager, **kwargs
):
Expand All @@ -235,12 +235,15 @@ def stream(
generation += chunk
assert generation is not None
except BaseException as e:
run_manager.on_llm_error(e)
run_manager.on_llm_error(
e,
response=LLMResult(
generations=[[generation]] if generation else []
),
)
raise e
else:
run_manager.on_llm_end(
LLMResult(generations=[[generation]]),
)
run_manager.on_llm_end(LLMResult(generations=[[generation]]))

async def astream(
self,
Expand Down Expand Up @@ -277,8 +280,8 @@ async def astream(
name=config.get("run_name"),
batch_size=1,
)
generation: Optional[ChatGenerationChunk] = None
try:
generation: Optional[ChatGenerationChunk] = None
async for chunk in self._astream(
messages, stop=stop, run_manager=run_manager, **kwargs
):
Expand All @@ -289,7 +292,12 @@ async def astream(
generation += chunk
assert generation is not None
except BaseException as e:
await run_manager.on_llm_error(e)
await run_manager.on_llm_error(
e,
response=LLMResult(
generations=[[generation]] if generation else []
),
)
raise e
else:
await run_manager.on_llm_end(
Expand Down Expand Up @@ -366,7 +374,7 @@ def generate(
)
except BaseException as e:
if run_managers:
run_managers[i].on_llm_error(e)
run_managers[i].on_llm_error(e, response=LLMResult(generations=[]))
raise e
flattened_outputs = [
LLMResult(generations=[res.generations], llm_output=res.llm_output)
Expand Down Expand Up @@ -433,7 +441,9 @@ async def agenerate(
for i, res in enumerate(results):
if isinstance(res, BaseException):
if run_managers:
await run_managers[i].on_llm_error(res)
await run_managers[i].on_llm_error(
res, response=LLMResult(generations=[])
)
exceptions.append(res)
if exceptions:
if run_managers:
Expand Down
25 changes: 19 additions & 6 deletions libs/core/langchain_core/language_models/llms.py
Original file line number Diff line number Diff line change
Expand Up @@ -384,8 +384,8 @@ def stream(
name=config.get("run_name"),
batch_size=1,
)
generation: Optional[GenerationChunk] = None
try:
generation: Optional[GenerationChunk] = None
for chunk in self._stream(
prompt, stop=stop, run_manager=run_manager, **kwargs
):
Expand All @@ -396,7 +396,12 @@ def stream(
generation += chunk
assert generation is not None
except BaseException as e:
run_manager.on_llm_error(e)
run_manager.on_llm_error(
e,
response=LLMResult(
generations=[[generation]] if generation else []
),
)
raise e
else:
run_manager.on_llm_end(LLMResult(generations=[[generation]]))
Expand Down Expand Up @@ -436,8 +441,8 @@ async def astream(
name=config.get("run_name"),
batch_size=1,
)
generation: Optional[GenerationChunk] = None
try:
generation: Optional[GenerationChunk] = None
async for chunk in self._astream(
prompt, stop=stop, run_manager=run_manager, **kwargs
):
Expand All @@ -448,7 +453,12 @@ async def astream(
generation += chunk
assert generation is not None
except BaseException as e:
await run_manager.on_llm_error(e)
await run_manager.on_llm_error(
e,
response=LLMResult(
generations=[[generation]] if generation else []
),
)
raise e
else:
await run_manager.on_llm_end(LLMResult(generations=[[generation]]))
Expand Down Expand Up @@ -539,7 +549,7 @@ def _generate_helper(
)
except BaseException as e:
for run_manager in run_managers:
run_manager.on_llm_error(e)
run_manager.on_llm_error(e, response=LLMResult(generations=[]))
raise e
flattened_outputs = output.flatten()
for manager, flattened_output in zip(run_managers, flattened_outputs):
Expand Down Expand Up @@ -707,7 +717,10 @@ async def _agenerate_helper(
)
except BaseException as e:
await asyncio.gather(
*[run_manager.on_llm_error(e) for run_manager in run_managers]
*[
run_manager.on_llm_error(e, response=LLMResult(generations=[]))
for run_manager in run_managers
]
)
raise e
flattened_outputs = output.flatten()
Expand Down
8 changes: 5 additions & 3 deletions libs/core/tests/unit_tests/fake/callbacks.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ class BaseFakeCallbackHandler(BaseModel):
starts: int = 0
ends: int = 0
errors: int = 0
errors_args: List[Any] = []
text: int = 0
ignore_llm_: bool = False
ignore_chain_: bool = False
Expand Down Expand Up @@ -52,8 +53,9 @@ def on_llm_end_common(self) -> None:
self.llm_ends += 1
self.ends += 1

def on_llm_error_common(self) -> None:
def on_llm_error_common(self, *args: Any, **kwargs: Any) -> None:
self.errors += 1
self.errors_args.append({"args": args, "kwargs": kwargs})

def on_llm_new_token_common(self) -> None:
self.llm_streams += 1
Expand Down Expand Up @@ -160,7 +162,7 @@ def on_llm_error(
*args: Any,
**kwargs: Any,
) -> Any:
self.on_llm_error_common()
self.on_llm_error_common(*args, **kwargs)

def on_retry(
self,
Expand Down Expand Up @@ -322,7 +324,7 @@ async def on_llm_error(
*args: Any,
**kwargs: Any,
) -> None:
self.on_llm_error_common()
self.on_llm_error_common(*args, **kwargs)

async def on_chain_start(
self,
Expand Down
16 changes: 14 additions & 2 deletions libs/core/tests/unit_tests/fake/chat_model.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ class FakeListChatModel(SimpleChatModel):
responses: List
sleep: Optional[float] = None
i: int = 0
error_on_chunk_number: Optional[int] = None

@property
def _llm_type(self) -> str:
Expand Down Expand Up @@ -77,9 +78,15 @@ def _stream(
self.i += 1
else:
self.i = 0
for c in response:
for i_c, c in enumerate(response):
if self.sleep is not None:
time.sleep(self.sleep)
if (
self.error_on_chunk_number is not None
and i_c == self.error_on_chunk_number
):
raise Exception("Fake error")

yield ChatGenerationChunk(message=AIMessageChunk(content=c))

async def _astream(
Expand All @@ -94,9 +101,14 @@ async def _astream(
self.i += 1
else:
self.i = 0
for c in response:
for i_c, c in enumerate(response):
if self.sleep is not None:
await asyncio.sleep(self.sleep)
if (
self.error_on_chunk_number is not None
and i_c == self.error_on_chunk_number
):
raise Exception("Fake error")
yield ChatGenerationChunk(message=AIMessageChunk(content=c))

@property
Expand Down
18 changes: 16 additions & 2 deletions libs/core/tests/unit_tests/fake/llm.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,8 @@ def _identifying_params(self) -> Mapping[str, Any]:
class FakeStreamingListLLM(FakeListLLM):
"""Fake streaming list LLM for testing purposes."""

error_on_chunk_number: Optional[int] = None

def stream(
self,
input: LanguageModelInput,
Expand All @@ -69,9 +71,15 @@ def stream(
**kwargs: Any,
) -> Iterator[str]:
result = self.invoke(input, config)
for c in result:
for i_c, c in enumerate(result):
if self.sleep is not None:
time.sleep(self.sleep)

if (
self.error_on_chunk_number is not None
and i_c == self.error_on_chunk_number
):
raise Exception("Fake error")
yield c

async def astream(
Expand All @@ -83,7 +91,13 @@ async def astream(
**kwargs: Any,
) -> AsyncIterator[str]:
result = await self.ainvoke(input, config)
for c in result:
for i_c, c in enumerate(result):
if self.sleep is not None:
await asyncio.sleep(self.sleep)

if (
self.error_on_chunk_number is not None
and i_c == self.error_on_chunk_number
):
raise Exception("Fake error")
yield c
Original file line number Diff line number Diff line change
@@ -1,8 +1,15 @@
"""Test base chat model."""

import pytest

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.outputs.llm_result import LLMResult
from langchain_core.tracers.context import collect_runs
from tests.unit_tests.fake.callbacks import (
BaseFakeCallbackHandler,
FakeAsyncCallbackHandler,
FakeCallbackHandler,
)
from tests.unit_tests.fake.chat_model import FakeListChatModel


Expand Down Expand Up @@ -69,3 +76,33 @@ async def test_async_batch_size(messages: list, messages_2: list) -> None:
pass
assert len(cb.traced_runs) == 1
assert (cb.traced_runs[0].extra or {}).get("batch_size") == 1


async def test_stream_error_callback() -> None:
message = "test"

def eval_response(callback: BaseFakeCallbackHandler, i: int) -> None:
assert callback.errors == 1
assert len(callback.errors_args) == 1
llm_result: LLMResult = callback.errors_args[0]["kwargs"]["response"]
if i == 0:
assert llm_result.generations == []
else:
assert llm_result.generations[0][0].text == message[:i]

for i in range(0, 2):
llm = FakeListChatModel(
responses=[message],
error_on_chunk_number=i,
)
with pytest.raises(Exception):
cb_async = FakeAsyncCallbackHandler()
async for _ in llm.astream("Dummy message", callbacks=[cb_async]):
pass
eval_response(cb_async, i)

cb_sync = FakeCallbackHandler()
for _ in llm.stream("Dumy message", callbacks=[cb_sync]):
pass

eval_response(cb_sync, i)
Loading

0 comments on commit aa8ae31

Please sign in to comment.