init() — langchain Function Reference

Architecture documentation for the __init__() function in streaming_aiter_final_only.py from the langchain codebase.

Function python LangChainCore MessageInterface

Entity Profile

LangChainCore→ MessageInterface→ __init__() — langchain Function Reference

Dependency Diagram

graph TD
  4fbcb1c2_72be_1d3f_eacb_3e3407d2d46b["__init__()"]
  8ff14092_eef3_abb2_55fb_ca2a28f79a83["AsyncFinalIteratorCallbackHandler"]
  4fbcb1c2_72be_1d3f_eacb_3e3407d2d46b -->|defined in| 8ff14092_eef3_abb2_55fb_ca2a28f79a83
  style 4fbcb1c2_72be_1d3f_eacb_3e3407d2d46b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain/langchain_classic/callbacks/streaming_aiter_final_only.py lines 33–65

    def __init__(
        self,
        *,
        answer_prefix_tokens: list[str] | None = None,
        strip_tokens: bool = True,
        stream_prefix: bool = False,
    ) -> None:
        """Instantiate AsyncFinalIteratorCallbackHandler.

        Args:
            answer_prefix_tokens: Token sequence that prefixes the answer.
                Default is ["Final", "Answer", ":"]
            strip_tokens: Ignore white spaces and new lines when comparing
                answer_prefix_tokens to last tokens? (to determine if answer has been
                reached)
            stream_prefix: Should answer prefix itself also be streamed?
        """
        super().__init__()
        if answer_prefix_tokens is None:
            self.answer_prefix_tokens = DEFAULT_ANSWER_PREFIX_TOKENS
        else:
            self.answer_prefix_tokens = answer_prefix_tokens
        if strip_tokens:
            self.answer_prefix_tokens_stripped = [
                token.strip() for token in self.answer_prefix_tokens
            ]
        else:
            self.answer_prefix_tokens_stripped = self.answer_prefix_tokens
        self.last_tokens = [""] * len(self.answer_prefix_tokens)
        self.last_tokens_stripped = [""] * len(self.answer_prefix_tokens)
        self.strip_tokens = strip_tokens
        self.stream_prefix = stream_prefix
        self.answer_reached = False