codekingpro/portable-devtools
114k
1"""Run managers."""2 3from __future__ import annotations4 5import asyncio6import atexit7import functools8import logging9from abc import ABC, abstractmethod10from collections.abc import Callable, Mapping11from concurrent.futures import ThreadPoolExecutor12from contextlib import asynccontextmanager, contextmanager13from contextvars import copy_context14from typing import TYPE_CHECKING, Any, TypeVar, cast15 16from typing_extensions import Self, override17 18from langchain_core.callbacks.base import (19 BaseCallbackHandler,20 BaseCallbackManager,21 Callbacks,22 ChainManagerMixin,23 LLMManagerMixin,24 RetrieverManagerMixin,25 RunManagerMixin,26 ToolManagerMixin,27)28from langchain_core.callbacks.stdout import StdOutCallbackHandler29from langchain_core.globals import get_debug30from langchain_core.messages import BaseMessage, get_buffer_string31from langchain_core.utils.env import env_var_is_set32from langchain_core.utils.uuid import uuid733 34if TYPE_CHECKING:35 from collections.abc import AsyncGenerator, Coroutine, Generator, Sequence36 from uuid import UUID37 38 from langchain_protocol.protocol import MessagesData39 from tenacity import RetryCallState40 41 from langchain_core.agents import AgentAction, AgentFinish42 from langchain_core.documents import Document43 from langchain_core.outputs import ChatGenerationChunk, GenerationChunk, LLMResult44 from langchain_core.runnables.config import RunnableConfig45 from langchain_core.tracers.schemas import Run46 47logger = logging.getLogger(__name__)48 49 50def _get_debug() -> bool:51 return get_debug()52 53 54@contextmanager55def trace_as_chain_group(56 group_name: str,57 callback_manager: CallbackManager | None = None,58 *,59 inputs: dict[str, Any] | None = None,60 project_name: str | None = None,61 example_id: str | UUID | None = None,62 run_id: UUID | None = None,63 tags: list[str] | None = None,64 metadata: dict[str, Any] | None = None,65) -> Generator[CallbackManagerForChainGroup, None, None]:66 """Get a callback manager for a chain group in a context manager.67 68 Useful for grouping different calls together as a single run even if they aren't69 composed in a single chain.70 71 Args:72 group_name: The name of the chain group.73 callback_manager: The callback manager to use.74 inputs: The inputs to the chain group.75 project_name: The name of the project.76 example_id: The ID of the example.77 run_id: The ID of the run.78 tags: The inheritable tags to apply to all runs.79 metadata: The metadata to apply to all runs.80 81 !!! note82 83 Must have `LANGCHAIN_TRACING_V2` env var set to true to see the trace in84 LangSmith.85 86 Yields:87 The callback manager for the chain group.88 89 Example:90 ```python91 llm_input = "Foo"92 with trace_as_chain_group("group_name", inputs={"input": llm_input}) as manager:93 # Use the callback manager for the chain group94 res = llm.invoke(llm_input, {"callbacks": manager})95 manager.on_chain_end({"output": res})96 ```97 """98 from langchain_core.tracers.context import ( # noqa: PLC0415 -- deferred to avoid importing langsmith at module level99 _get_trace_callbacks,100 )101 102 cb = _get_trace_callbacks(103 project_name, example_id, callback_manager=callback_manager104 )105 cm = CallbackManager.configure(106 inheritable_callbacks=cb,107 inheritable_tags=tags,108 inheritable_metadata=metadata,109 )110 111 run_manager = cm.on_chain_start({"name": group_name}, inputs or {}, run_id=run_id)112 child_cm = run_manager.get_child()113 group_cm = CallbackManagerForChainGroup(114 child_cm.handlers,115 child_cm.inheritable_handlers,116 child_cm.parent_run_id,117 parent_run_manager=run_manager,118 tags=child_cm.tags,119 inheritable_tags=child_cm.inheritable_tags,120 metadata=child_cm.metadata,121 inheritable_metadata=child_cm.inheritable_metadata,122 )123 try:124 yield group_cm125 except Exception as e:126 if not group_cm.ended:127 run_manager.on_chain_error(e)128 raise129 else:130 if not group_cm.ended:131 run_manager.on_chain_end({})132 133 134@asynccontextmanager135async def atrace_as_chain_group(136 group_name: str,137 callback_manager: AsyncCallbackManager | None = None,138 *,139 inputs: dict[str, Any] | None = None,140 project_name: str | None = None,141 example_id: str | UUID | None = None,142 run_id: UUID | None = None,143 tags: list[str] | None = None,144 metadata: dict[str, Any] | None = None,145) -> AsyncGenerator[AsyncCallbackManagerForChainGroup, None]:146 """Get an async callback manager for a chain group in a context manager.147 148 Useful for grouping different async calls together as a single run even if they149 aren't composed in a single chain.150 151 Args:152 group_name: The name of the chain group.153 callback_manager: The async callback manager to use, which manages tracing and154 other callback behavior.155 inputs: The inputs to the chain group.156 project_name: The name of the project.157 example_id: The ID of the example.158 run_id: The ID of the run.159 tags: The inheritable tags to apply to all runs.160 metadata: The metadata to apply to all runs.161 162 Yields:163 The async callback manager for the chain group.164 165 !!! note166 167 Must have `LANGCHAIN_TRACING_V2` env var set to true to see the trace in168 LangSmith.169 170 Example:171 ```python172 llm_input = "Foo"173 async with atrace_as_chain_group(174 "group_name", inputs={"input": llm_input}175 ) as manager:176 # Use the async callback manager for the chain group177 res = await llm.ainvoke(llm_input, {"callbacks": manager})178 await manager.on_chain_end({"output": res})179 ```180 """181 from langchain_core.tracers.context import ( # noqa: PLC0415 -- deferred to avoid importing langsmith at module level182 _get_trace_callbacks,183 )184 185 cb = _get_trace_callbacks(186 project_name, example_id, callback_manager=callback_manager187 )188 cm = AsyncCallbackManager.configure(189 inheritable_callbacks=cb, inheritable_tags=tags, inheritable_metadata=metadata190 )191 192 run_manager = await cm.on_chain_start(193 {"name": group_name}, inputs or {}, run_id=run_id194 )195 child_cm = run_manager.get_child()196 group_cm = AsyncCallbackManagerForChainGroup(197 child_cm.handlers,198 child_cm.inheritable_handlers,199 child_cm.parent_run_id,200 parent_run_manager=run_manager,201 tags=child_cm.tags,202 inheritable_tags=child_cm.inheritable_tags,203 metadata=child_cm.metadata,204 inheritable_metadata=child_cm.inheritable_metadata,205 )206 try:207 yield group_cm208 except Exception as e:209 if not group_cm.ended:210 await run_manager.on_chain_error(e)211 raise212 else:213 if not group_cm.ended:214 await run_manager.on_chain_end({})215 216 217Func = TypeVar("Func", bound=Callable)218 219 220def shielded(func: Func) -> Func:221 """Makes so an awaitable method is always shielded from cancellation.222 223 Args:224 func: The function to shield.225 226 Returns:227 The shielded function228 229 """230 231 @functools.wraps(func)232 async def wrapped(*args: Any, **kwargs: Any) -> Any:233 # Capture the current context to preserve context variables234 ctx = copy_context()235 236 # Create the coroutine237 coro = func(*args, **kwargs)238 239 # For Python 3.11+, create task with explicit context240 # For older versions, fallback to original behavior241 try:242 # Create a task with the captured context to preserve context variables243 task = asyncio.create_task(coro, context=ctx) # type: ignore[call-arg, unused-ignore]244 # `call-arg` used to not fail 3.9 or 3.10 tests245 return await asyncio.shield(task)246 except TypeError:247 # Python < 3.11 fallback - create task normally then shield248 # This won't preserve context perfectly but is better than nothing249 task = asyncio.create_task(coro)250 return await asyncio.shield(task)251 252 return cast("Func", wrapped)253 254 255def handle_event(256 handlers: list[BaseCallbackHandler],257 event_name: str,258 ignore_condition_name: str | None,259 *args: Any,260 **kwargs: Any,261) -> None:262 """Generic event handler for `CallbackManager`.263 264 Args:265 handlers: The list of handlers that will handle the event.266 event_name: The name of the event (e.g., `'on_llm_start'`).267 ignore_condition_name: Name of the attribute defined on handler that if `True`268 will cause the handler to be skipped for the given event.269 *args: The arguments to pass to the event handler.270 **kwargs: The keyword arguments to pass to the event handler271 272 """273 coros: list[Coroutine[Any, Any, Any]] = []274 275 try:276 message_strings: list[str] | None = None277 for handler in handlers:278 try:279 if ignore_condition_name is None or not getattr(280 handler, ignore_condition_name281 ):282 event = getattr(handler, event_name)(*args, **kwargs)283 if asyncio.iscoroutine(event):284 coros.append(event)285 except NotImplementedError as e:286 if event_name == "on_chat_model_start":287 if message_strings is None:288 message_strings = [get_buffer_string(m) for m in args[1]]289 handle_event(290 [handler],291 "on_llm_start",292 "ignore_llm",293 args[0],294 message_strings,295 *args[2:],296 **kwargs,297 )298 else:299 handler_name = handler.__class__.__name__300 logger.warning(301 "NotImplementedError in %s.%s callback: %s",302 handler_name,303 event_name,304 repr(e),305 )306 except Exception as e:307 logger.warning(308 "Error in %s.%s callback: %s",309 handler.__class__.__name__,310 event_name,311 repr(e),312 )313 if handler.raise_error:314 raise315 finally:316 if coros:317 try:318 # Raises RuntimeError if there is no current event loop.319 asyncio.get_running_loop()320 loop_running = True321 except RuntimeError:322 loop_running = False323 324 if loop_running:325 # If we try to submit this coroutine to the running loop326 # we end up in a deadlock, as we'd have gotten here from a327 # running coroutine, which we cannot interrupt to run this one.328 # The solution is to run the synchronous function on the globally shared329 # thread pool executor to avoid blocking the main event loop.330 _executor().submit(331 cast("Callable", copy_context().run), _run_coros, coros332 ).result()333 else:334 # If there's no running loop, we can run the coroutines directly.335 _run_coros(coros)336 337 338def _run_coros(coros: list[Coroutine[Any, Any, Any]]) -> None:339 if hasattr(asyncio, "Runner"):340 # Python 3.11+341 # Run the coroutines in a new event loop, taking care to342 # - install signal handlers343 # - run pending tasks scheduled by `coros`344 # - close asyncgens and executors345 # - close the loop346 with asyncio.Runner() as runner:347 # Run the coroutine, get the result348 for coro in coros:349 try:350 runner.run(coro)351 except Exception as e:352 logger.warning("Error in callback coroutine: %s", repr(e))353 354 # Run pending tasks scheduled by coros until they are all done355 while pending := asyncio.all_tasks(runner.get_loop()):356 runner.run(asyncio.wait(pending))357 else:358 # Before Python 3.11 we need to run each coroutine in a new event loop359 # as the Runner api is not available.360 for coro in coros:361 try:362 asyncio.run(coro)363 except Exception as e:364 logger.warning("Error in callback coroutine: %s", repr(e))365 366 367async def _ahandle_event_for_handler(368 handler: BaseCallbackHandler,369 event_name: str,370 ignore_condition_name: str | None,371 *args: Any,372 **kwargs: Any,373) -> None:374 try:375 if ignore_condition_name is None or not getattr(handler, ignore_condition_name):376 event = getattr(handler, event_name)377 if asyncio.iscoroutinefunction(event):378 await event(*args, **kwargs)379 elif handler.run_inline:380 event(*args, **kwargs)381 else:382 await asyncio.get_event_loop().run_in_executor(383 None,384 cast(385 "Callable",386 functools.partial(copy_context().run, event, *args, **kwargs),387 ),388 )389 except NotImplementedError as e:390 if event_name == "on_chat_model_start":391 message_strings = [get_buffer_string(m) for m in args[1]]392 await _ahandle_event_for_handler(393 handler,394 "on_llm_start",395 "ignore_llm",396 args[0],397 message_strings,398 *args[2:],399 **kwargs,400 )401 else:402 logger.warning(403 "NotImplementedError in %s.%s callback: %s",404 handler.__class__.__name__,405 event_name,406 repr(e),407 )408 except Exception as e:409 logger.warning(410 "Error in %s.%s callback: %s",411 handler.__class__.__name__,412 event_name,413 repr(e),414 )415 if handler.raise_error:416 raise417 418 419async def ahandle_event(420 handlers: list[BaseCallbackHandler],421 event_name: str,422 ignore_condition_name: str | None,423 *args: Any,424 **kwargs: Any,425) -> None:426 """Async generic event handler for `AsyncCallbackManager`.427 428 Args:429 handlers: The list of handlers that will handle the event.430 event_name: The name of the event (e.g., `'on_llm_start'`).431 ignore_condition_name: Name of the attribute defined on handler that if `True`432 will cause the handler to be skipped for the given event.433 *args: The arguments to pass to the event handler.434 **kwargs: The keyword arguments to pass to the event handler.435 436 """437 for handler in [h for h in handlers if h.run_inline]:438 await _ahandle_event_for_handler(439 handler, event_name, ignore_condition_name, *args, **kwargs440 )441 await asyncio.gather(442 *(443 _ahandle_event_for_handler(444 handler,445 event_name,446 ignore_condition_name,447 *args,448 **kwargs,449 )450 for handler in handlers451 if not handler.run_inline452 )453 )454 455 456class BaseRunManager(RunManagerMixin):457 """Base class for run manager (a bound callback manager)."""458 459 def __init__(460 self,461 *,462 run_id: UUID,463 handlers: list[BaseCallbackHandler],464 inheritable_handlers: list[BaseCallbackHandler],465 parent_run_id: UUID | None = None,466 tags: list[str] | None = None,467 inheritable_tags: list[str] | None = None,468 metadata: dict[str, Any] | None = None,469 inheritable_metadata: dict[str, Any] | None = None,470 ) -> None:471 """Initialize the run manager.472 473 Args:474 run_id: The ID of the run.475 handlers: The list of handlers.476 inheritable_handlers: The list of inheritable handlers.477 parent_run_id: The ID of the parent run.478 tags: The list of tags.479 inheritable_tags: The list of inheritable tags.480 metadata: The metadata.481 inheritable_metadata: The inheritable metadata.482 483 """484 self.run_id = run_id485 self.handlers = handlers486 self.inheritable_handlers = inheritable_handlers487 self.parent_run_id = parent_run_id488 self.tags = tags or []489 self.inheritable_tags = inheritable_tags or []490 self.metadata = metadata or {}491 self.inheritable_metadata = inheritable_metadata or {}492 493 @classmethod494 def get_noop_manager(cls) -> Self:495 """Return a manager that doesn't perform any operations.496 497 Returns:498 The noop manager.499 500 """501 return cls(502 run_id=uuid7(),503 handlers=[],504 inheritable_handlers=[],505 tags=[],506 inheritable_tags=[],507 metadata={},508 inheritable_metadata={},509 )510 511 512class RunManager(BaseRunManager):513 """Synchronous run manager."""514 515 def on_text(516 self,517 text: str,518 **kwargs: Any,519 ) -> None:520 """Run when a text is received.521 522 Args:523 text: The received text.524 **kwargs: Additional keyword arguments.525 """526 if not self.handlers:527 return528 handle_event(529 self.handlers,530 "on_text",531 None,532 text,533 run_id=self.run_id,534 parent_run_id=self.parent_run_id,535 tags=self.tags,536 **kwargs,537 )538 539 def on_retry(540 self,541 retry_state: RetryCallState,542 **kwargs: Any,543 ) -> None:544 """Run when a retry is received.545 546 Args:547 retry_state: The retry state.548 **kwargs: Additional keyword arguments.549 550 """551 if not self.handlers:552 return553 handle_event(554 self.handlers,555 "on_retry",556 "ignore_retry",557 retry_state,558 run_id=self.run_id,559 parent_run_id=self.parent_run_id,560 tags=self.tags,561 **kwargs,562 )563 564 565class ParentRunManager(RunManager):566 """Synchronous parent run manager."""567 568 def get_child(self, tag: str | None = None) -> CallbackManager:569 """Get a child callback manager.570 571 Args:572 tag: The tag for the child callback manager.573 574 Returns:575 The child callback manager.576 577 """578 manager = CallbackManager(handlers=[], parent_run_id=self.run_id)579 manager.set_handlers(self.inheritable_handlers)580 manager.add_tags(self.inheritable_tags)581 manager.add_metadata(self.inheritable_metadata)582 if tag is not None:583 manager.add_tags([tag], inherit=False)584 return manager585 586 587class AsyncRunManager(BaseRunManager, ABC):588 """Async run manager."""589 590 @abstractmethod591 def get_sync(self) -> RunManager:592 """Get the equivalent sync `RunManager`.593 594 Returns:595 The sync `RunManager`.596 597 """598 599 async def on_text(600 self,601 text: str,602 **kwargs: Any,603 ) -> None:604 """Run when a text is received.605 606 Args:607 text: The received text.608 **kwargs: Additional keyword arguments.609 """610 if not self.handlers:611 return612 await ahandle_event(613 self.handlers,614 "on_text",615 None,616 text,617 run_id=self.run_id,618 parent_run_id=self.parent_run_id,619 tags=self.tags,620 **kwargs,621 )622 623 async def on_retry(624 self,625 retry_state: RetryCallState,626 **kwargs: Any,627 ) -> None:628 """Async run when a retry is received.629 630 Args:631 retry_state: The retry state.632 **kwargs: Additional keyword arguments.633 634 """635 if not self.handlers:636 return637 await ahandle_event(638 self.handlers,639 "on_retry",640 "ignore_retry",641 retry_state,642 run_id=self.run_id,643 parent_run_id=self.parent_run_id,644 tags=self.tags,645 **kwargs,646 )647 648 649class AsyncParentRunManager(AsyncRunManager):650 """Async parent run manager."""651 652 def get_child(self, tag: str | None = None) -> AsyncCallbackManager:653 """Get a child callback manager.654 655 Args:656 tag: The tag for the child callback manager.657 658 Returns:659 The child callback manager.660 661 """662 manager = AsyncCallbackManager(handlers=[], parent_run_id=self.run_id)663 manager.set_handlers(self.inheritable_handlers)664 manager.add_tags(self.inheritable_tags)665 manager.add_metadata(self.inheritable_metadata)666 if tag is not None:667 manager.add_tags([tag], inherit=False)668 return manager669 670 671class CallbackManagerForLLMRun(RunManager, LLMManagerMixin):672 """Callback manager for LLM run."""673 674 def on_llm_new_token(675 self,676 token: str,677 *,678 chunk: GenerationChunk | ChatGenerationChunk | None = None,679 **kwargs: Any,680 ) -> None:681 """Run when LLM generates a new token.682 683 Args:684 token: The new token.685 chunk: The chunk.686 **kwargs: Additional keyword arguments.687 688 """689 if not self.handlers:690 return691 handle_event(692 self.handlers,693 "on_llm_new_token",694 "ignore_llm",695 token=token,696 run_id=self.run_id,697 parent_run_id=self.parent_run_id,698 tags=self.tags,699 chunk=chunk,700 **kwargs,701 )702 703 def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:704 """Run when LLM ends running.705 706 Args:707 response: The LLM result.708 **kwargs: Additional keyword arguments.709 710 """711 if not self.handlers:712 return713 handle_event(714 self.handlers,715 "on_llm_end",716 "ignore_llm",717 response,718 run_id=self.run_id,719 parent_run_id=self.parent_run_id,720 tags=self.tags,721 **kwargs,722 )723 724 def on_llm_error(725 self,726 error: BaseException,727 **kwargs: Any,728 ) -> None:729 """Run when LLM errors.730 731 Args:732 error: The error.733 **kwargs: Additional keyword arguments.734 735 - response (LLMResult): The response which was generated before736 the error occurred.737 """738 if not self.handlers:739 return740 handle_event(741 self.handlers,742 "on_llm_error",743 "ignore_llm",744 error,745 run_id=self.run_id,746 parent_run_id=self.parent_run_id,747 tags=self.tags,748 **kwargs,749 )750 751 def on_stream_event(self, event: MessagesData, **kwargs: Any) -> None:752 """Run on each protocol event from `stream_events(version="v3")`.753 754 Args:755 event: The protocol event.756 **kwargs: Additional keyword arguments.757 """758 if not self.handlers:759 return760 handle_event(761 self.handlers,762 "on_stream_event",763 "ignore_llm",764 event,765 run_id=self.run_id,766 parent_run_id=self.parent_run_id,767 tags=self.tags,768 **kwargs,769 )770 771 772class AsyncCallbackManagerForLLMRun(AsyncRunManager, LLMManagerMixin):773 """Async callback manager for LLM run."""774 775 def get_sync(self) -> CallbackManagerForLLMRun:776 """Get the equivalent sync `RunManager`.777 778 Returns:779 The sync `RunManager`.780 781 """782 return CallbackManagerForLLMRun(783 run_id=self.run_id,784 handlers=self.handlers,785 inheritable_handlers=self.inheritable_handlers,786 parent_run_id=self.parent_run_id,787 tags=self.tags,788 inheritable_tags=self.inheritable_tags,789 metadata=self.metadata,790 inheritable_metadata=self.inheritable_metadata,791 )792 793 async def on_llm_new_token(794 self,795 token: str,796 *,797 chunk: GenerationChunk | ChatGenerationChunk | None = None,798 **kwargs: Any,799 ) -> None:800 """Run when LLM generates a new token.801 802 Args:803 token: The new token.804 chunk: The chunk.805 **kwargs: Additional keyword arguments.806 807 """808 if not self.handlers:809 return810 await ahandle_event(811 self.handlers,812 "on_llm_new_token",813 "ignore_llm",814 token,815 chunk=chunk,816 run_id=self.run_id,817 parent_run_id=self.parent_run_id,818 tags=self.tags,819 **kwargs,820 )821 822 @shielded823 async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:824 """Run when LLM ends running.825 826 Args:827 response: The LLM result.828 **kwargs: Additional keyword arguments.829 830 """831 if not self.handlers:832 return833 await ahandle_event(834 self.handlers,835 "on_llm_end",836 "ignore_llm",837 response,838 run_id=self.run_id,839 parent_run_id=self.parent_run_id,840 tags=self.tags,841 **kwargs,842 )843 844 @shielded845 async def on_llm_error(846 self,847 error: BaseException,848 **kwargs: Any,849 ) -> None:850 """Run when LLM errors.851 852 Args:853 error: The error.854 **kwargs: Additional keyword arguments.855 856 - response (LLMResult): The response which was generated before857 the error occurred.858 859 """860 if not self.handlers:861 return862 await ahandle_event(863 self.handlers,864 "on_llm_error",865 "ignore_llm",866 error,867 run_id=self.run_id,868 parent_run_id=self.parent_run_id,869 tags=self.tags,870 **kwargs,871 )872 873 async def on_stream_event(self, event: MessagesData, **kwargs: Any) -> None:874 """Run on each protocol event from `astream_events(version="v3")`.875 876 Args:877 event: The protocol event.878 **kwargs: Additional keyword arguments.879 """880 if not self.handlers:881 return882 await ahandle_event(883 self.handlers,884 "on_stream_event",885 "ignore_llm",886 event,887 run_id=self.run_id,888 parent_run_id=self.parent_run_id,889 tags=self.tags,890 **kwargs,891 )892 893 894class CallbackManagerForChainRun(ParentRunManager, ChainManagerMixin):895 """Callback manager for chain run."""896 897 def on_chain_end(self, outputs: dict[str, Any] | Any, **kwargs: Any) -> None:898 """Run when chain ends running.899 900 Args:901 outputs: The outputs of the chain.902 **kwargs: Additional keyword arguments.903 904 """905 if not self.handlers:906 return907 handle_event(908 self.handlers,909 "on_chain_end",910 "ignore_chain",911 outputs,912 run_id=self.run_id,913 parent_run_id=self.parent_run_id,914 tags=self.tags,915 **kwargs,916 )917 918 def on_chain_error(919 self,920 error: BaseException,921 **kwargs: Any,922 ) -> None:923 """Run when chain errors.924 925 Args:926 error: The error.927 **kwargs: Additional keyword arguments.928 929 """930 if not self.handlers:931 return932 handle_event(933 self.handlers,934 "on_chain_error",935 "ignore_chain",936 error,937 run_id=self.run_id,938 parent_run_id=self.parent_run_id,939 tags=self.tags,940 **kwargs,941 )942 943 def on_agent_action(self, action: AgentAction, **kwargs: Any) -> None:944 """Run when agent action is received.945 946 Args:947 action: The agent action.948 **kwargs: Additional keyword arguments.949 """950 if not self.handlers:951 return952 handle_event(953 self.handlers,954 "on_agent_action",955 "ignore_agent",956 action,957 run_id=self.run_id,958 parent_run_id=self.parent_run_id,959 tags=self.tags,960 **kwargs,961 )962 963 def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> None:964 """Run when agent finish is received.965 966 Args:967 finish: The agent finish.968 **kwargs: Additional keyword arguments.969 """970 if not self.handlers:971 return972 handle_event(973 self.handlers,974 "on_agent_finish",975 "ignore_agent",976 finish,977 run_id=self.run_id,978 parent_run_id=self.parent_run_id,979 tags=self.tags,980 **kwargs,981 )982 983 984class AsyncCallbackManagerForChainRun(AsyncParentRunManager, ChainManagerMixin):985 """Async callback manager for chain run."""986 987 def get_sync(self) -> CallbackManagerForChainRun:988 """Get the equivalent sync `RunManager`.989 990 Returns:991 The sync `RunManager`.992 """993 return CallbackManagerForChainRun(994 run_id=self.run_id,995 handlers=self.handlers,996 inheritable_handlers=self.inheritable_handlers,997 parent_run_id=self.parent_run_id,998 tags=self.tags,999 inheritable_tags=self.inheritable_tags,1000 metadata=self.metadata,1001 inheritable_metadata=self.inheritable_metadata,1002 )1003 1004 @shielded1005 async def on_chain_end(self, outputs: dict[str, Any] | Any, **kwargs: Any) -> None:1006 """Run when a chain ends running.1007 1008 Args:1009 outputs: The outputs of the chain.1010 **kwargs: Additional keyword arguments.1011 1012 """1013 if not self.handlers:1014 return1015 await ahandle_event(1016 self.handlers,1017 "on_chain_end",1018 "ignore_chain",1019 outputs,1020 run_id=self.run_id,1021 parent_run_id=self.parent_run_id,1022 tags=self.tags,1023 **kwargs,1024 )1025 1026 @shielded1027 async def on_chain_error(1028 self,1029 error: BaseException,1030 **kwargs: Any,1031 ) -> None:1032 """Run when chain errors.1033 1034 Args:1035 error: The error.1036 **kwargs: Additional keyword arguments.1037 1038 """1039 if not self.handlers:1040 return1041 await ahandle_event(1042 self.handlers,1043 "on_chain_error",1044 "ignore_chain",1045 error,1046 run_id=self.run_id,1047 parent_run_id=self.parent_run_id,1048 tags=self.tags,1049 **kwargs,1050 )1051 1052 async def on_agent_action(self, action: AgentAction, **kwargs: Any) -> None:1053 """Run when agent action is received.1054 1055 Args:1056 action: The agent action.1057 **kwargs: Additional keyword arguments.1058 """1059 if not self.handlers:1060 return1061 await ahandle_event(1062 self.handlers,1063 "on_agent_action",1064 "ignore_agent",1065 action,1066 run_id=self.run_id,1067 parent_run_id=self.parent_run_id,1068 tags=self.tags,1069 **kwargs,1070 )1071 1072 async def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> None:1073 """Run when agent finish is received.1074 1075 Args:1076 finish: The agent finish.1077 **kwargs: Additional keyword arguments.1078 """1079 if not self.handlers:1080 return1081 await ahandle_event(1082 self.handlers,1083 "on_agent_finish",1084 "ignore_agent",1085 finish,1086 run_id=self.run_id,1087 parent_run_id=self.parent_run_id,1088 tags=self.tags,1089 **kwargs,1090 )1091 1092 1093class CallbackManagerForToolRun(ParentRunManager, ToolManagerMixin):1094 """Callback manager for tool run."""1095 1096 def on_tool_end(1097 self,1098 output: Any,1099 **kwargs: Any,1100 ) -> None:1101 """Run when the tool ends running.1102 1103 Args:1104 output: The output of the tool.1105 **kwargs: The keyword arguments to pass to the event handler1106 1107 """1108 if not self.handlers:1109 return1110 handle_event(1111 self.handlers,1112 "on_tool_end",1113 "ignore_agent",1114 output,1115 run_id=self.run_id,1116 parent_run_id=self.parent_run_id,1117 tags=self.tags,1118 **kwargs,1119 )1120 1121 def on_tool_error(1122 self,1123 error: BaseException,1124 **kwargs: Any,1125 ) -> None:1126 """Run when tool errors.1127 1128 Args:1129 error: The error.1130 **kwargs: Additional keyword arguments.1131 1132 """1133 if not self.handlers:1134 return1135 handle_event(1136 self.handlers,1137 "on_tool_error",1138 "ignore_agent",1139 error,1140 run_id=self.run_id,1141 parent_run_id=self.parent_run_id,1142 tags=self.tags,1143 **kwargs,1144 )1145 1146 1147class AsyncCallbackManagerForToolRun(AsyncParentRunManager, ToolManagerMixin):1148 """Async callback manager for tool run."""1149 1150 def get_sync(self) -> CallbackManagerForToolRun:1151 """Get the equivalent sync `RunManager`.1152 1153 Returns:1154 The sync `RunManager`.1155 """1156 return CallbackManagerForToolRun(1157 run_id=self.run_id,1158 handlers=self.handlers,1159 inheritable_handlers=self.inheritable_handlers,1160 parent_run_id=self.parent_run_id,1161 tags=self.tags,1162 inheritable_tags=self.inheritable_tags,1163 metadata=self.metadata,1164 inheritable_metadata=self.inheritable_metadata,1165 )1166 1167 async def on_tool_end(self, output: Any, **kwargs: Any) -> None:1168 """Async run when the tool ends running.1169 1170 Args:1171 output: The output of the tool.1172 **kwargs: Additional keyword arguments.1173 1174 """1175 if not self.handlers:1176 return1177 await ahandle_event(1178 self.handlers,1179 "on_tool_end",1180 "ignore_agent",1181 output,1182 run_id=self.run_id,1183 parent_run_id=self.parent_run_id,1184 tags=self.tags,1185 **kwargs,1186 )1187 1188 async def on_tool_error(1189 self,1190 error: BaseException,1191 **kwargs: Any,1192 ) -> None:1193 """Run when tool errors.1194 1195 Args:1196 error: The error.1197 **kwargs: Additional keyword arguments.1198 1199 """1200 if not self.handlers: