Source code for langchain_core.outputs.generation

from __future__ import annotations

from typing import Any, Dict, List, Literal, Optional

from langchain_core.load import Serializable
from langchain_core.utils._merge import merge_dicts


class Generation(Serializable):
    """一个单独的文本生成输出。"""

    text: str
    """生成的文本输出。"""

    generation_info: Optional[Dict[str, Any]] = None
    """来自提供者的原始响应。可能包括完成原因或令牌日志概率等内容。"""
    type: Literal["Generation"] = "Generation"
    """Type 仅用于序列化目的。"""
    # TODO: add log probs as separate attribute

    @classmethod
    def is_lc_serializable(cls) -> bool:
        """返回此类是否可序列化。"""
        return True

    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        """获取langchain对象的命名空间。"""
        return ["langchain", "schema", "output"]


class GenerationChunk(Generation):
    """生成块，可以与其他生成块连接。"""

    @classmethod
    def get_lc_namespace(cls) -> List[str]:
        """获取langchain对象的命名空间。"""
        return ["langchain", "schema", "output"]

    def __add__(self, other: GenerationChunk) -> GenerationChunk:
        if isinstance(other, GenerationChunk):
            generation_info = merge_dicts(
                self.generation_info or {},
                other.generation_info or {},
            )
            return GenerationChunk(
                text=self.text + other.text,
                generation_info=generation_info or None,
            )
        else:
            raise TypeError(
                f"unsupported operand type(s) for +: '{type(self)}' and '{type(other)}'"
            )