LangChain

sieves supports LangChain's Chat Models (e.g., ChatOpenAI, ChatAnthropic) for structured output.

Usage

from langchain_openai import ChatOpenAI
from sieves import tasks

# Initialize a LangChain Chat Model
model = ChatOpenAI(model="gpt-4o-mini", api_key="dummy")

# Pass it to a task
task = tasks.SentimentAnalysis(model=model)

---

Bases: PydanticModelWrapper[PromptSignature, Result, Model, InferenceMode]

ModelWrapper for LangChain.

Source code in sieves/model_wrappers/langchain_.py

class LangChain(PydanticModelWrapper[PromptSignature, Result, Model, InferenceMode]):
    """ModelWrapper for LangChain."""

    @override
    @property
    def inference_modes(self) -> type[InferenceMode]:
        return InferenceMode

    @override
    def build_executable(
        self,
        inference_mode: InferenceMode,
        prompt_template: str | None,  # noqa: UP007
        prompt_signature: type[PromptSignature] | PromptSignature,
        fewshot_examples: Sequence[pydantic.BaseModel] = tuple(),
    ) -> Executable[Result | None]:
        assert isinstance(prompt_signature, type)
        cls_name = self.__class__.__name__
        template = self._create_template(prompt_template)
        model = self._model.with_structured_output(prompt_signature, include_raw=True)

        def execute(values: Sequence[dict[str, Any]]) -> Sequence[tuple[Result | None, Any, TokenUsage]]:
            """Execute prompts with model wrapper for given values.

            :param values: Values to inject into prompts.
            :return: Sequence of tuples containing results, raw outputs, and token usage. Results are None if
                corresponding prompt failed.
            """
            match inference_mode:
                case InferenceMode.structured:

                    def generate(prompts: list[str]) -> Iterable[tuple[Result, Any, TokenUsage]]:
                        try:
                            results = asyncio.run(model.abatch(prompts, **self._inference_kwargs))
                            for res in results:
                                usage = TokenUsage()
                                raw = res["raw"]
                                if hasattr(raw, "usage_metadata") and raw.usage_metadata:
                                    usage.input_tokens = raw.usage_metadata.get("input_tokens")
                                    usage.output_tokens = raw.usage_metadata.get("output_tokens")

                                yield res["parsed"], raw, usage

                        except Exception as err:
                            raise RuntimeError(
                                f"Encountered problem in parsing {cls_name} output. Double-check your prompts and "
                                f"examples."
                            ) from err

                    generator = generate
                case _:
                    raise ValueError(f"Inference mode {inference_mode} not supported by {cls_name} model wrapper.")

            return self._infer(generator, template, values)

        return execute

model property

Return model instance.

Returns:

Type	Description
ModelWrapperModel	Model instance.

model_settings property

Return model settings.

Returns:

Type	Description
ModelSettings	Model settings.

__init__(model, model_settings)

Initialize model wrapper with model and model settings.

Parameters:

Name	Type	Description	Default
model	ModelWrapperModel	Instantiated model instance.	required
model_settings	ModelSettings	Model settings.	required

Source code in sieves/model_wrappers/core.py

def __init__(self, model: ModelWrapperModel, model_settings: ModelSettings):
    """Initialize model wrapper with model and model settings.

    :param model: Instantiated model instance.
    :param model_settings: Model settings.
    """
    self._model = model
    self._model_settings = model_settings
    self._inference_kwargs = model_settings.inference_kwargs or {}
    self._init_kwargs = model_settings.init_kwargs or {}
    self._strict = model_settings.strict

convert_fewshot_examples(fewshot_examples) staticmethod

Convert few‑shot examples to dicts.

Parameters:

Name	Type	Description	Default
fewshot_examples	Sequence[BaseModel]	Fewshot examples to convert.	required

Returns:

Type	Description
list[dict[str, Any]]	Fewshot examples as dicts.

Source code in sieves/model_wrappers/core.py

@staticmethod
def convert_fewshot_examples(fewshot_examples: Sequence[pydantic.BaseModel]) -> list[dict[str, Any]]:
    """Convert few‑shot examples to dicts.

    :param fewshot_examples: Fewshot examples to convert.
    :return: Fewshot examples as dicts.
    """
    return [fs_example.model_dump(serialize_as_any=True) for fs_example in fewshot_examples]