Langchain框架的设计哲学

"""LangChain **Runnable** and the **LangChain Expression Language (LCEL)**.
 
The LangChain Expression Language (LCEL) offers a declarative method to build
production-grade programs that harness the power of LLMs.
 
Programs created using LCEL and LangChain `Runnable` objects inherently support
synchronous asynchronous, batch, and streaming operations.
 
Support for **async** allows servers hosting LCEL based programs to scale bette for
higher concurrent loads.
 
**Batch** operations allow for processing multiple inputs in parallel.
 
**Streaming** of intermediate outputs, as they're being generated, allows for creating
more responsive UX.
 
This module contains schema and implementation of LangChain `Runnable` object
primitives.
"""

class Chain(RunnableSerializable[dict[str, Any], dict[str, Any]], ABC):
    """Abstract base class for creating structured sequences of calls to components.
 
    Chains should be used to encode a sequence of calls to components like
    models, document retrievers, other chains, etc., and provide a simple interface
    to this sequence.
 
    The Chain interface makes it easy to create apps that are:
        - Stateful: add Memory to any Chain to give it state,
        - Observable: pass Callbacks to a Chain to execute additional functionality,
            like logging, outside the main sequence of component calls,
        - Composable: the Chain API is flexible enough that it is easy to combine
            Chains with other components, including other Chains.
 
    The main methods exposed by chains are:
        - `__call__`: Chains are callable. The `__call__` method is the primary way to
            execute a Chain. This takes inputs as a dictionary and returns a
            dictionary output.
        - `run`: A convenience method that takes inputs as args/kwargs and returns the
            output as a string or object. This method can only be used for a subset of
            chains and cannot return as rich of an output as `__call__`.
    """

 
When deserializing, the class path from the JSON `'id'` field is checked against an
allowlist. If the class is not in the allowlist, deserialization raises a `ValueError`.
 
## Threat model
 
A serialized LangChain payload crosses a trust boundary because the manifest
may contain serialized objects and configuration that affect runtime behavior.
For example, a payload can configure a chat model with a custom `base_url`,
custom headers, a different model name, or other constructor arguments. These
are supported features, but they also mean the payload contents should be
treated as executable configuration rather than plain text.
 
Concretely, deserialization instantiates Python objects, so any constructor
(`__init__`) or validator on an allowed class can run during `load()`. A
crafted payload that is allowed to reach an unintended class — or an intended
class with attacker-controlled kwargs — could cause network calls, file
operations, or environment-variable access while the object is being built.
 
!!! warning "Do not use with untrusted input"
 
    If the source is untrusted, avoid calling `load()` / `loads()` on it. If
    you must, restrict `allowed_objects` to types that do not execute logic
    during init — `allowed_objects='messages'` (or an explicit list of
    message classes) is the safe choice. Keep `secrets_from_env=False`

[project.optional-dependencies]
community = ["langchain-community"]
anthropic = ["langchain-anthropic"]
openai = ["langchain-openai"]
azure-ai = ["langchain-azure-ai"]
#cohere = ["langchain-cohere"]
google-vertexai = ["langchain-google-vertexai"]
google-genai = ["langchain-google-genai"]
fireworks = ["langchain-fireworks"]
ollama = ["langchain-ollama"]
together = ["langchain-together"]
mistralai = ["langchain-mistralai"]
huggingface = ["langchain-huggingface"]
groq = ["langchain-groq"]
aws = ["langchain-aws"]
baseten = ["langchain-baseten>=0.2.0"]
deepseek = ["langchain-deepseek"]
xai = ["langchain-xai"]
perplexity = ["langchain-perplexity"]
meta = ["langchain-meta"]

 
- `uv` – Fast Python package installer and resolver (replaces pip/poetry)
- `make` – Task runner for common development commands. Feel free to look at the `Makefile` for available commands and usage patterns.
- `ruff` – Fast Python linter and formatter
- `mypy` – Static type checking
- `pytest`

    def __or__(
        self,
        other: Runnable[Output, Other]
        | Callable[[Iterator[Output]], Iterator[Other]]
        | Callable[[AsyncIterator[Output]], AsyncIterator[Other]]
        | Callable[[Output], Other]
        | Mapping[str, Runnable[Output, Any] | Callable[[Output], Any] | Any],
    ) -> RunnableSerializable[Input, Any]:
        """Runnable "or" operator.
 
        Compose this `Runnable` with another object to create a
        `RunnableSequence`.
 
        Args:
            other: Another `Runnable` or a `Runnable`-like object.
 
        Returns:
            A new `Runnable`.
        """
        return RunnableSequence(self, coerce_to_runnable(other))

    @overload
    def __ror__(


        | Callable[[Other], Input],
    ) -> RunnableSerializable[Other, Output]: ...
 
    def __ror__(
        self,
        other: Runnable[Other, Input]
        | Callable[[Iterator[Other]], Iterator[Input]]
        | Callable[[AsyncIterator[Other]], AsyncIterator[Input]]
        | Callable[[Other], Any]
        | Mapping[str, Runnable[Other, Input] | Callable[[Other], Any] | Any],
    ) -> RunnableSerializable[Any, Output]:
        """Runnable "reverse-or" operator.
 
        Compose this `Runnable` with another object to create a
        `RunnableSequence`.
 
        Args:
            other: Another `Runnable` or a `Runnable`-like object.
 
        Returns:
            A new `Runnable`.
        """
        return RunnableSequence(coerce_to_runnable(other), self

sequence = runnable_1.pipe(runnable_2)  # 等价于 
sequence = runnable_1 | runnable_2  # 等价于 
sequence = RunnableSequence(first=runnable_1, last=runnable_2)

def pipe(
        self,
        *others: Runnable[Any, Other] | Callable[[Any], Other],
        name: str | None = None,
    ) -> RunnableSerializable[Input, Other]:
        """Pipe `Runnable` objects.
 
        Compose this `Runnable` with `Runnable`-like objects to make a
        `RunnableSequence`.
 
        Equivalent to `RunnableSequence(self, *others)` or `self | others[0] | ...`
 
        Example:
            ```python
            from langchain_core.runnables import RunnableLambda
 
 
            def add_one(x: int) -> int:
                return x + 1
 
 
            def mul_two(x: int) -> int:
                return x * 2
 
 
            runnable_1 = RunnableLambda(add_one)
            runnable_2 = RunnableLambda(mul_two)
            sequence = runnable_1.pipe(runnable_2)
            # Or equivalently:
            # sequence = runnable_1 | runnable_2
            # sequence = RunnableSequence(first=runnable_1, last=runnable_2)
            sequence.invoke(1)
            await sequence.ainvoke(1)
            # -> 4
 
            sequence.batch([1, 2, 3])
            await sequence.abatch([1, 2, 3])
            # -> [4, 6, 8]
            ```
 
        Args:
            *others: Other `Runnable` or `Runnable`-like objects to compose
            name: An optional name for the resulting `RunnableSequence`.
 
        Returns:
            A new `Runnable`.
        """
        return RunnableSequence(self, *others, name=name)

 class RunnableSequence(RunnableSerializable[Input, Output]):
    """Sequence of `Runnable` objects, where the output of one is the input of the next.
 
    **`RunnableSequence`** is the most important composition operator in LangChain
    as it is used in virtually every chain.
 
    A `RunnableSequence` can be instantiated directly or more commonly by using the
    `|` operator where either the left or right operands (or both) must be a
    `Runnable`.
 
    Any `RunnableSequence` automatically supports sync, async, batch.
 
    The default implementations of `batch` and `abatch` utilize threadpools and
    asyncio gather and will be faster than naive invocation of `invoke` or `ainvoke`
    for IO bound `Runnable`s.
 
    Batching is implemented by invoking the batch method on each component of the
    `RunnableSequence` in order.
 
    A `RunnableSequence` preserves the streaming properties of its components, so if
    all components of the sequence implement a `transform` method -- which
    is the method that implements the logic to map a streaming input to a streaming
    output -- then the sequence will be able to stream input to output!
 
    If any component of the sequence does not implement transform then the
    streaming will only begin after this component is run. If there are
    multiple blocking components, streaming begins after the last one.
 
    !!! note
        `RunnableLambdas` do not support `transform` by default! So if you need to
        use a `RunnableLambdas` be careful about where you place them in a
        `RunnableSequence` (if you need to use the `stream`/`astream` methods).
 
        If you need arbitrary logic and need streaming, you can subclass
        Runnable, and implement `transform` for whatever logic you need.
 
    Here is a simple example that uses simple functions to illustrate the use of
    `RunnableSequence`:
 
        ```python
        from langchain_core.runnables import RunnableLambda
 
 
        def add_one(x: int) -> int:
            return x + 1
 
 
        def mul_two(x: int) -> int:
            return x * 2
 
 
        runnable_1 = RunnableLambda(add_one)
        runnable_2 = RunnableLambda(mul_two)
        sequence = runnable_1 | runnable_2
        # Or equivalently:
        # sequence = RunnableSequence(first=runnable_1, last=runnable_2)
        sequence.invoke(1)
        await sequence.ainvoke(1)
 
        sequence.batch([1, 2, 3])
        await sequence.abatch([1, 2, 3])
        ```
 
    Here's an example that uses streams JSON output generated by an LLM:
 
        ```python
        from langchain_core.output_parsers.json import SimpleJsonOutputParser
        from langchain_openai import ChatOpenAI
 
        prompt = PromptTemplate.from_template(
            "In JSON format, give me a list of {topic} and their "
            "corresponding names in French, Spanish and in a "
            "Cat Language."
        )
 
        model = ChatOpenAI()
        chain = prompt | model | SimpleJsonOutputParser()
 
        async for chunk in chain.astream({"topic": "colors"}):
            print("-")  # noqa: T201
            print(chunk, sep="", flush=True)  # noqa: T201
        ```
    """
 
    # The steps are broken into first, middle and last, solely for type checking
    # purposes. It allows specifying the `Input` on the first type, the `Output` of
    # the last type.
    first: Runnable[Input, Any]
    """The first `Runnable` in the sequence."""
    middle: list[Runnable[Any, Any]] = Field(default_factory=list)
    """The middle `Runnable` in the sequence."""
    last: Runnable[Any, Output]
    """The last `Runnable` in the sequence."""
 
    def __init__(
        self,
        *steps: RunnableLike[Any, Any],
        name: str | None = None,
        first: Runnable[Any, Any] | None = None,
        middle: list[Runnable[Any, Any]] | None = None,
        last: Runnable[Any, Any] | None = None,
    ) -> None:
        """Create a new `RunnableSequence`.
 
        Args:
            steps: The steps to include in the sequence.
            name: The name of the `Runnable`.
            first: The first `Runnable` in the sequence.
            middle: The middle `Runnable` objects in the sequence.
            last: The last `Runnable` in the sequence.
 
        Raises:
            ValueError: If the sequence has less than 2 steps.
        """
        steps_flat: list[Runnable[Any, Any]] = []
        if not steps and first is not None and last is not None:
            steps_flat = [first] + (middle or []) + [last]
        for step in steps:
            if isinstance(step, RunnableSequence):
                steps_flat.extend(step.steps)
            else:
                steps_flat.append(coerce_to_runnable(step))
        if len(steps_flat) < _RUNNABLE_SEQUENCE_MIN_STEPS:
            msg = (
                f"RunnableSequence must have at least {_RUNNABLE_SEQUENCE_MIN_STEPS} "
                f"steps, got {len(steps_flat)}"
            )
            raise ValueError(msg)
        super().__init__(
            first=steps_flat[0],
            middle=list(steps_flat[1:-1]),
            last=steps_flat[-1],
            name=name,
        )

    llm = FakeListLLM(responses=["i'm a textbot"])
 
    chain = (
        prompt
        | passthrough
        | {
            "chat": chat,
            "llm": llm,
        }
    )
 
    assert repr(chain) == snapshot
    assert isinstance(chain, RunnableSequence)
    assert chain.first == prompt
    assert chain.middle == [RunnableLambda(passthrough)]
    assert isinstance(chain.last, RunnableParallel)
    assert dumps(chain, pretty=True) == snapshot

    @classmethod
    def is_lc_serializable(cls) -> bool:
        """Is this class serializable?
 
        By design, even if a class inherits from `Serializable`, it is not serializable
        by default. This is to prevent accidental serialization of objects that should
        not be serialized.
 
        Returns:
            Whether the class is serializable. Default is `False`.
        """
        return False

    @classmethod
    def get_lc_namespace(cls) -> list[str]:
        """Get the namespace of the LangChain object.
 
        The default implementation splits `cls.__module__` on `'.'`, e.g.
        `langchain_openai.chat_models` becomes
        `["langchain_openai", "chat_models"]`. This value is used by `lc_id` to
        build the serialization identifier.
 
        New partner packages should **not** override this method. The default
        behavior is correct for any class whose module path already reflects
        its package name. Some older packages (e.g. `langchain-openai`,
        `langchain-anthropic`) override it to return a legacy-style namespace
        like `["langchain", "chat_models", "openai"]`, matching the module
        paths that existed before those integrations were split out of the
        main `langchain` package. Those overrides are kept for
        backwards-compatible deserialization; new packages should not copy them.
 
        Deserialization mapping is handled separately by
        `SERIALIZABLE_MAPPING` in `langchain_core.load.mapping`.
 
        Returns:
            The namespace.
        """
        return cls.__module__.split(".")

    @property
    def lc_secrets(self) -> dict[str, str]:
        """Mapping of secret keys to environment variables."""
        return {"openai_api_key": "OPENAI_API_KEY"}

    @property
    def lc_attributes(self) -> dict[str, Any]:
        """LangChain attributes for this class."""
        attributes: dict[str, Any] = {}
        if self.openai_api_base:
            attributes["openai_api_base"] = self.openai_api_base
 
        if self.openai_organization:
            attributes["openai_organization"] = self.openai_organization
 
        if self.openai_proxy:
            attributes["openai_proxy"] = self.openai_proxy
 
        return attributes

    def to_json(self) -> SerializedConstructor | SerializedNotImplemented:
        """Serialize the object to JSON.
 
        Raises:
            ValueError: If the class has deprecated attributes.
 
        Returns:
            A JSON serializable object or a `SerializedNotImplemented` object.
        """
        if not self.is_lc_serializable():
            return self.to_json_not_implemented()
 
        model_fields = type(self).model_fields
        secrets = {}
        # Get latest values for kwargs if there is an attribute with same name
        lc_kwargs = {}
        for k, v in self:
            if not _is_field_useful(self, k, v):
                continue
            # Do nothing if the field is excluded
            if k in model_fields and model_fields[k].exclude:
                continue
 
            lc_kwargs[k] = getattr(self, k, v)
 
        # Merge the lc_secrets and lc_attributes from every class in the MRO
        for cls in [None, *self.__class__.mro()]:
            # Once we get to Serializable, we're done
            if cls is Serializable:
                break
 
            if cls:
                deprecated_attributes = [
                    "lc_namespace",
                    "lc_serializable",
                ]
 
                for attr in deprecated_attributes:
                    if hasattr(cls, attr):
                        msg = (
                            f"Class {self.__class__} has a deprecated "
                            f"attribute {attr}. Please use the corresponding "
                            f"classmethod instead."
                        )
                        raise ValueError(msg)
 
            # Get a reference to self bound to each class in the MRO
            this = cast("Serializable", self if cls is None else super(cls, self))
 
            secrets.update(this.lc_secrets)
            # Now also add the aliases for the secrets
            # This ensures known secret aliases are hidden.
            # Note: this does NOT hide any other extra kwargs
            # that are not present in the fields.
            for key in list(secrets):
                value = secrets[key]
                if (key in model_fields) and (
                    alias := model_fields[key].alias
                ) is not None:
                    secrets[alias] = value
            lc_kwargs.update(this.lc_attributes)
 
        # include all secrets, even if not specified in kwargs
        # as these secrets may be passed as an environment variable instead

    @abstractmethod
    def invoke(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> Output:
        """Transform a single input into an output.
 
        Args:
            input: The input to the `Runnable`.
            config: A config to use when invoking the `Runnable`.
 
                The config supports standard keys like `'tags'`, `'metadata'` for
                tracing purposes, `'max_concurrency'` for controlling how much work to
                do in parallel, and other keys.
 
                Please refer to `RunnableConfig` for more details.
 
        Returns:
            The output of the `Runnable`.
        """

    async def ainvoke(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> Output:
        """Transform a single input into an output.
 
        Args:
            input: The input to the `Runnable`.
            config: A config to use when invoking the `Runnable`.
 
                The config supports standard keys like `'tags'`, `'metadata'` for
                tracing purposes, `'max_concurrency'` for controlling how much work to
                do in parallel, and other keys.
 
                Please refer to `RunnableConfig` for more details.
 
        Returns:
            The output of the `Runnable`.
        """
        return await run_in_executor(config, self.invoke, input, config, **kwargs)

class Serializable(BaseModel, ABC):
    """Serializable base class.
 

    def __init__(
        self,
        *steps: RunnableLike[Any, Any],

        Raises:
            ValueError: If the sequence has less than 2 steps.
        """
        steps_flat: list[Runnable[Any, Any]] = []
        if not steps and first is not None and last is not None:
            steps_flat = [first] + (middle or []) + [last]
        for step in steps:
            if isinstance(step, RunnableSequence):
                steps_flat.extend(step.steps)
            else:
                steps_flat.append(coerce_to_runnable(step))
        if len(steps_flat) < _RUNNABLE_SEQUENCE_MIN_STEPS:
            msg = (
                f"RunnableSequence must have at least {_RUNNABLE_SEQUENCE_MIN_STEPS} "
                f"steps, got {len(steps_flat)}"
            )
            raise ValueError(msg)
        super().__init__(
            first=steps_flat[0],
            middle=list(steps_flat[1:-1]),
            last=steps_flat[-1],
            name=name,
       

    @property
    def steps(self) -> list[Runnable[Any, Any]]:
        """All the `Runnable`s that make up the sequence in order.
 
        Returns:
            A list of `Runnable`s.
        """
        return [self.first, *self.middle, self.last]

class BaseMessage(Serializable):
    """Base abstract message class.
 
    Messages are the inputs and outputs of a chat model.
 
    Examples include [`HumanMessage`][langchain.messages.HumanMessage],
    [`AIMessage`][langchain.messages.AIMessage], and
    [`SystemMessage`][langchain.messages.SystemMessage].
    """
 
    content: str | list[str | dict[Any, Any]]
    """The contents of the message."""
 
    additional_kwargs: dict[Any, Any] = Field(default_factory=dict)
    """Reserved for additional payload data associated with the message.
 
    For example, for a message from an AI, this could include tool calls as
    encoded by the model provider.
 
    """
 
    response_metadata: dict[Any, Any] = Field(default_factory=dict)
    """Examples: response headers, logprobs, token counts, model name."""
 
    type: str
    """The type of the message. Must be a string that is unique to the message type.
 
    The purpose of this field is to allow for easy identification of the message type
    when deserializing messages.
 
    """
 
    name: str | None = None
    """An optional name for the message.
 
    This can be used to provide a human-readable name for the message.
 
    Usage of this field is optional, and whether it's used or not is up to the
    model implementation.
 
    """
 
    id: str | None = Field(default=None, coerce_numbers_to_str=True)
    """An optional unique identifier for the message.
 
    This should ideally be provided by the provider/model which created the message.
 
    """
 
    model_config = ConfigDict(
        extra="allow",
    )

class SystemMessageChunk(SystemMessage, BaseMessageChunk):
    """System Message chunk."""
 
    # Ignoring mypy re-assignment here since we're overriding the value
    # to make sure that the chunk variant can be discriminated from the
    # non-chunk variant.
    type: Literal["SystemMessageChunk"] = "SystemMessageChunk"  # type: ignore[assignment]
    """The type of the message (used for serialization)."""

class HumanMessageChunk(HumanMessage, BaseMessageChunk):
    """Human Message chunk."""
 
    # Ignoring mypy re-assignment here since we're overriding the value
    # to make sure that the chunk variant can be discriminated from the
    # non-chunk variant.
    type: Literal["HumanMessageChunk"] = "HumanMessageChunk"  # type: ignore[assignment]
    """The type of the message (used for serialization)."""

LanguageModelInput = PromptValue | str | Sequence[MessageLikeRepresentation]
"""Input to a language model."""
 
LanguageModelOutput = BaseMessage | str
"""Output from a language model."""
 
LanguageModelLike = Runnable[LanguageModelInput, LanguageModelOutput]
"""Input/output interface for a language model."""
 
LanguageModelOutputVar = TypeVar("LanguageModelOutputVar", AIMessage, str)
"""Type variable for the output of a language model."""
 

class BaseLLM(BaseLanguageModel[str], ABC):
    """Base LLM abstract interface.
 
    It should take in a prompt and return a string.
    """
 
    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )
 
    @functools.cached_property
    def _serialized(self) -> builtins.dict[str, Any]:
        # self is always a Serializable object in this case, thus the result is
        # guaranteed to be a dict since dumpd uses the default callback, which uses
        # obj.to_json which always returns TypedDict subclasses
        return cast("builtins.dict[str, Any]", dumpd(self))
 
    # --- Runnable methods ---
 
    @property
    @override
    def OutputType(self) -> type[str]:
        """Get the output type for this `Runnable`."""
        return str

    def _convert_input(self, model_input: LanguageModelInput) -> PromptValue:
        if isinstance(model_input, PromptValue):
            return model_input
        if isinstance(model_input, str):
            return StringPromptValue(text=model_input)
        if isinstance(model_input, Sequence):
            return ChatPromptValue(messages=convert_to_messages(model_input))
        msg = (  # type: ignore[unreachable]
            f"Invalid input type {type(model_input)}. "
            "Must be a PromptValue, str, or list of BaseMessages."
        )
        raise ValueError(msg)

""Output classes.
 
Used to represent the output of a language model call and the output of a chat.
 
The top container for information is the `LLMResult` object. `LLMResult` is used by both
chat models and LLMs. This object contains the output of the language model and any
additional information that the model provider wants to return.
 
When invoking models via the standard runnable methods (e.g. invoke, batch, etc.):
 
- Chat models will return `AIMessage` objects.
- LLMs will return regular text strings.
 
In addition, users can access the raw output of either LLMs or chat models via
callbacks. The `on_chat_model_end` and `on_llm_end` callbacks will return an `LLMResult`
object containing the generated outputs and any additional information returned by the
model provider.
 
In general, if information is already available in the AIMessage object, it is
recommended to access it from there rather than from the `LLMResult` object.
"""
 
from typing import TYPE_CHECKING
 
from langchain_core._import_utils import import_attr
 
if TYPE_CHECKING:
    from langchain_core.outputs.chat_generation import (
        ChatGeneration,
        ChatGenerationChunk,
    )
    from langchain_core.outputs.chat_result import ChatResult
    from langchain_core.outputs.generation import Generation, GenerationChunk
    from langchain_core.outputs.llm_result import LLMResult
    from langchain_core.outputs.run_info import RunInfo
 
__all__ = (
    "ChatGeneration",
    "ChatGenerationChunk",
    "ChatResult",
    "Generation",
    "GenerationChunk",
    "LLMResult",
    "RunInfo",
)
 
_dynamic_imports = {
    "ChatGeneration": "chat_generation",
    "ChatGenerationChunk": "chat_generation",
    "ChatResult": "chat_result",
    "Generation": "generation",
    "GenerationChunk": "generation",
    "LLMResult": "llm_result",
    "RunInfo": "run_info",
}