chatomon.connectors.langchain ¶

Build an atom into a native LangChain agent (a LangGraph compiled graph).

Install with pip install 'chatomon[langchain]'.

Classes:

Name	Description
`LangChainConnector`	Builds an atom into a native LangChain agent (a LangGraph `CompiledStateGraph`).

LangChainConnector ¶

LangChainConnector(
    model: str | BaseChatModel,
    *,
    resolve_model: (
        Callable[[str], BaseChatModel] | None
    ) = None,
    parent_span: Callable[[], Span | None] | None = None,
    **agent_kwargs: Any
)

Bases: Connector['CompiledStateGraph[Any, Any, Any, Any]']

Builds an atom into a native LangChain agent (a LangGraph CompiledStateGraph).

chatomon's middleware is the outermost layer; your own middleware run inside it. Extra keyword arguments go to langchain.agents.create_agent unchanged (tools, checkpointer, store, ...). With a checkpointer, the thread_id of the run config becomes the conversation id. When the atom has output orbitals the model is tagged nostream, so streaming emits only checked replies.

Parameters:

Name	Type	Description	Default
`model`	`str \| BaseChatModel`	The nucleus model: a chat model or a `"provider:model"` string.	required
`resolve_model`	`Callable[[str], BaseChatModel] \| None`	Maps the model id of a JudgeOrbital to a chat model; by default `init_chat_model(model_id)`.	`None`
`parent_span`	`Callable[[], Span \| None] \| None`	Returns the span chatomon's spans should nest under; LangChain does not propagate an OpenTelemetry context itself.	`None`
`**agent_kwargs`	`Any`	Passed to `create_agent`.	`{}`

Raises:

Type	Description
`ChatomonConfigError`	If `agent_kwargs` sets `system_prompt`, `response_format` or `name` (chatomon sets them from the atom).

Examples:

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, HumanMessage

from chatomon import Atom
from chatomon.connectors.langchain import LangChainConnector

model = GenericFakeChatModel(messages=iter([AIMessage(content="Hi!")]))
agent = Atom(
    name="speedypack",
    instructions="You are the customer assistant of SpeedyPack.",
    connector=LangChainConnector(model),
).build()
state = agent.invoke({"messages": [HumanMessage(content="hello")]})
assert state["messages"][-1].content == "Hi!"

Methods:

Name	Description
`build`	Build the native agent with chatomon's middleware installed.
`generate_structured`	Make one structured call with `with_structured_output`.

Source code in src/chatomon/connectors/langchain.py

def __init__(
    self,
    model: str | BaseChatModel,
    *,
    resolve_model: Callable[[str], BaseChatModel] | None = None,
    parent_span: Callable[[], trace.Span | None] | None = None,
    **agent_kwargs: Any,
) -> None:
    reserved = sorted(_RESERVED & set(agent_kwargs))
    if reserved:
        raise ChatomonConfigError(
            f"{', '.join(reserved)} is set by chatomon from the atom; "
            "remove it from LangChainConnector(...)"
        )
    self.model = model
    self.resolve_model = resolve_model
    self.parent_span = parent_span
    self.agent_kwargs = agent_kwargs

build ¶

build(
    atom: Atom[Any], engine: Engine
) -> CompiledStateGraph[Any, Any, Any, Any]

Build the native agent with chatomon's middleware installed.

Parameters:

Name	Type	Description	Default
`atom`	`Atom[Any]`	The validated atom.	required
`engine`	`Engine`	The engine that runs the atom's orbitals.	required

Returns:

Type	Description
`CompiledStateGraph[Any, Any, Any, Any]`	The compiled LangGraph agent.

Source code in src/chatomon/connectors/langchain.py

def build(self, atom: Atom[Any], engine: Engine) -> CompiledStateGraph[Any, Any, Any, Any]:
    """Build the native agent with chatomon's middleware installed.

    Args:
        atom: The validated atom.
        engine: The engine that runs the atom's orbitals.

    Returns:
        The compiled LangGraph agent.
    """
    model: Any = self.model
    if engine.output_orbitals:
        if isinstance(model, str):
            model = init_chat_model(model)
        model = _with_nostream(model)
    kwargs = dict(self.agent_kwargs)
    middleware = [
        _ChatomonMiddleware(engine, atom.blocked_placeholder, self.parent_span),
        *kwargs.pop("middleware", ()),
    ]
    return create_agent(
        model,
        system_prompt=atom.system_prompt,
        middleware=middleware,
        name=atom.name,
        **kwargs,
    )

generate_structured `async` ¶

generate_structured(
    model: str,
    *,
    system: str,
    prompt: str,
    schema: type[ModelT]
) -> ModelT

Make one structured call with with_structured_output.

Parameters:

Name	Type	Description	Default
`model`	`str`	Model id; resolved with `resolve_model` or `init_chat_model`.	required
`system`	`str`	System prompt.	required
`prompt`	`str`	User prompt.	required
`schema`	`type[ModelT]`	Output type.	required

Returns:

Type	Description
`ModelT`	The validated output.

Source code in src/chatomon/connectors/langchain.py

async def generate_structured(
    self, model: str, *, system: str, prompt: str, schema: type[ModelT]
) -> ModelT:
    """Make one structured call with `with_structured_output`.

    Args:
        model: Model id; resolved with `resolve_model` or `init_chat_model`.
        system: System prompt.
        prompt: User prompt.
        schema: Output type.

    Returns:
        The validated output.
    """
    llm = (
        self.resolve_model(model) if self.resolve_model is not None else init_chat_model(model)
    )
    result = await llm.with_structured_output(schema).ainvoke(
        [SystemMessage(content=system), HumanMessage(content=prompt)]
    )
    return result if isinstance(result, schema) else schema.model_validate(result)

chatomon.connectors.langchain ¶

LangChainConnector ¶

build ¶

generate_structured async ¶

generate_structured `async` ¶