nat.plugins.langchain.agent.base#

Attributes#

`logger`
`TOOL_NOT_FOUND_ERROR_MESSAGE`
`INPUT_SCHEMA_MESSAGE`
`NO_INPUT_ERROR_MESSAGE`
`AGENT_LOG_PREFIX`
`AGENT_CALL_LOG_MESSAGE`
`TOOL_CALL_LOG_MESSAGE`

Classes#

`AgentDecision`	Create a collection of name/value pairs.
`BaseAgent`	Helper class that provides a standard way to create an ABC using

Functions#

_chunk_to_message(→ langchain_core.messages.AIMessage)

Convert an accumulated AIMessageChunk into an AIMessage, preserving tool_calls.

Module Contents#

logger#

_chunk_to_message( chunk: langchain_core.messages.AIMessageChunk, ) → langchain_core.messages.AIMessage#

Convert an accumulated AIMessageChunk into an AIMessage, preserving tool_calls.

When streaming chunks are accumulated via +, the result has tool_calls but additional_kwargs["tool_calls"] (the OpenAI wire format) is left empty. LLM providers read the wire format when the message is sent back in conversation history, so we reconstruct it here using convert_to_openai_messages.

TOOL_NOT_FOUND_ERROR_MESSAGE = 'There is no tool named {tool_name}. Tool must be one of {tools}.'#

INPUT_SCHEMA_MESSAGE = '. Arguments must be provided as a valid JSON object following this format: {schema}'#

NO_INPUT_ERROR_MESSAGE = 'No human input received to the agent, Please ask a valid question.'#

AGENT_LOG_PREFIX = '[AGENT]'#

AGENT_CALL_LOG_MESSAGE#

TOOL_CALL_LOG_MESSAGE#

class AgentDecision(*args, \*\*kwds)#

Bases: enum.Enum

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

attribute access:
```
>>> Color.RED
<Color.RED: 1>
```
value lookup:
```
>>> Color(1)
<Color.RED: 1>
```
name lookup:
```
>>> Color['RED']
<Color.RED: 1>
```

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

TOOL = 'tool'#

END = 'finished'#

class BaseAgent( llm: langchain_core.language_models.BaseChatModel, tools: list[langchain_core.tools.BaseTool], callbacks: list[langchain_core.callbacks.AsyncCallbackHandler] | None = None, detailed_logs: bool = False, log_response_max_chars: int = 1000, )#

Bases: abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

llm#

tools#

callbacks = []#

detailed_logs = False#

log_response_max_chars = 1000#

graph = None#

_runnable_config#

async _stream_llm( runnable: Any, inputs: dict[str, Any], config: langchain_core.runnables.RunnableConfig | None = None, ) → langchain_core.messages.AIMessage#

Stream from LLM runnable. Retry logic is handled automatically by the underlying LLM client.

Accumulates streamed chunks using LangChain’s + operator which preserves tool_calls and tool_call_chunks, then converts the result to an AIMessage via _chunk_to_message. This ensures that native tool calling (use_native_tool_calling=True) works correctly with the ReAct agent.

When a LangGraph runtime config is provided (for example, from an injected node config), it is merged with the local runnable config so that both LangGraph’s streaming callbacks and the profiler callbacks fire together.

Parameters#

runnableAny: The LLM runnable (prompt | llm or similar)
inputsDict[str, Any]: The inputs to pass to the runnable
configRunnableConfig | None: Optional LangGraph runtime config to merge with the local runnable config.

Returns#

AIMessage: The LLM response, including any tool_calls from native tool calling.

async _call_llm( llm: langchain_core.runnables.Runnable, inputs: dict[str, Any], ) → langchain_core.messages.AIMessage#

Call the LLM directly. Retry logic is handled automatically by the underlying LLM client.

Parameters#

llmRunnable: The LLM runnable (prompt | llm or similar)
inputsdict[str, Any]: The inputs to pass to the runnable

Returns#

AIMessage: The LLM response

async _call_tool( tool: langchain_core.tools.BaseTool, tool_input: dict[str, Any] | str, max_retries: int = 3, ) → langchain_core.messages.ToolMessage#

Call a tool with retry logic and error handling.

Parameters#

toolBaseTool: The tool to call
tool_inputUnion[Dict[str, Any], str]: The input to pass to the tool
max_retriesint: Maximum number of retry attempts (default: 3)

Returns#

ToolMessage: The tool response

_log_tool_response( tool_name: str, tool_input: Any, tool_response: str, ) → None#

Log tool response with consistent formatting and length limits.

Parameters#

tool_namestr: The name of the tool that was called
tool_inputAny: The input that was passed to the tool
tool_responsestr: The response from the tool

_parse_json(json_string: str) → dict[str, Any]#

Safely parse JSON with graceful error handling. If JSON parsing fails, returns an empty dict or error info.

Parameters#

json_stringstr: The JSON string to parse

Returns#

Dict[str, Any]: The parsed JSON or error information

_get_chat_history( messages: list[langchain_core.messages.BaseMessage], ) → str#

Get the chat history excluding the last message.

Parameters#

messageslist[BaseMessage]: The messages to get the chat history from

Returns#

str: The chat history excluding the last message

abstractmethod _build_graph( state_schema: type, ) → langgraph.graph.state.CompiledStateGraph#

Async: