nat.plugins.langchain.agent.tool_calling_agent.agent#

Attributes#

logger

Classes#

`ToolCallAgentGraphState`	State schema for the Tool Calling Agent Graph
`ToolCallAgentGraph`	Configurable LangGraph Tool Calling Agent. A Tool Calling Agent requires an LLM which supports tool calling.

Functions#

`_chunk_to_message`(→ langchain_core.messages.AIMessage)	Convert an accumulated AIMessageChunk into an AIMessage.
`create_tool_calling_agent_prompt`(→ str \| None)	Create a Tool Calling Agent prompt from the config.

Module Contents#

logger#

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

Convert an accumulated AIMessageChunk into an AIMessage.

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.

class ToolCallAgentGraphState(/, **data: Any)#

Bases: pydantic.BaseModel

State schema for the Tool Calling Agent Graph

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

messages: list[langchain_core.messages.base.BaseMessage] = None#

class ToolCallAgentGraph( llm: langchain_core.language_models.BaseChatModel, tools: list[langchain_core.tools.BaseTool], prompt: str | None = None, callbacks: list[langchain_core.callbacks.base.AsyncCallbackHandler] = None, detailed_logs: bool = False, log_response_max_chars: int = 1000, handle_tool_errors: bool = True, return_direct: list[langchain_core.tools.BaseTool] | None = None, max_truncation_retries: int = 0, truncation_scaling_fn: Callable[[int], int] | None = None, max_empty_response_retries: int = 0, )#

Bases: nat.plugins.langchain.agent.dual_node.DualNodeAgent

Configurable LangGraph Tool Calling Agent. A Tool Calling Agent requires an LLM which supports tool calling. A tool Calling Agent utilizes the tool input parameters to select the optimal tool. Supports handling tool errors. Argument “detailed_logs” toggles logging of inputs, outputs, and intermediate steps.

agent#

tool_caller#

return_direct#

_max_truncation_retries: int = 0#

_truncation_retries_remaining: int = 0#

_truncation_scaling_fn: Callable[[int], int]#

_current_max_tokens: int | None#

_max_empty_response_retries: int = 0#

async _invoke_llm(state: ToolCallAgentGraphState)#

Stream the LLM and return the accumulated AIMessage response.

Args:: state: Current agent graph state containing the conversation messages.
Returns:: The fully accumulated AIMessage from the LLM.
Raises:: RuntimeError: If the LLM returns no response.

async agent_node(state: ToolCallAgentGraphState)#

_get_token_usage( response: langchain_core.messages.AIMessage, ) → langchain_core.messages.ai.UsageMetadata#

Extract token usage from an AIMessage, checking both LangChain and OpenAI formats.

Returns:: UsageMetadata with input_tokens, output_tokens, total_tokens (values default to 0 if unavailable).

async _validate_llm_response( response: langchain_core.messages.AIMessage, state: ToolCallAgentGraphState, ) → langchain_core.messages.AIMessage#

Validate the LLM response and attempt recovery if configured.

Args:: response: The accumulated AIMessage from the LLM. state: Current agent graph state.
Returns:: A validated (possibly retried) response.

async _retry_on_truncation( first_response: langchain_core.messages.AIMessage, state: ToolCallAgentGraphState, ) → langchain_core.messages.AIMessage#

Retry the LLM call with a higher max_tokens when the output is truncated.

Without recovery, a truncated response typically lacks valid tool calls or a complete answer, causing the agent to loop until it hits a GraphRecursionError. Each retry increases max_tokens via the configured scaling callable, giving the LLM room to finish its output.

Args:: first_response: The truncated AIMessage from the LLM. state: Current agent graph state.
Returns:: The response from the successful retry.
Raises:: RuntimeError: If all retries are exhausted without a non-truncated response.

async _retry_on_empty_response( state: ToolCallAgentGraphState, first_metadata: dict, ) → langchain_core.messages.AIMessage#

Retry the LLM call when it returns an empty response.

Args:: state: Current agent graph state. first_metadata: The response_metadata from the empty response.
Returns:: The response from the successful retry.
Raises:: RuntimeError: If all retries are exhausted without a non-empty response.

async conditional_edge(state: ToolCallAgentGraphState)#

async tool_node(state: ToolCallAgentGraphState)#

async tool_conditional_edge( state: ToolCallAgentGraphState, ) → nat.plugins.langchain.agent.base.AgentDecision#

Determines whether to continue to the agent or end graph execution after a tool call.

Args:: state: The current state of the Tool Calling Agent graph containing messages and tool responses.
Returns:: AgentDecision: TOOL to continue to agent for processing, or END to terminate graph execution. Returns END if the tool is in return_direct list, otherwise returns TOOL to continue processing.

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

async build_graph() → langgraph.graph.state.CompiledStateGraph#

create_tool_calling_agent_prompt( config: nat.plugins.langchain.agent.tool_calling_agent.register.ToolCallAgentWorkflowConfig, ) → str | None#

Create a Tool Calling Agent prompt from the config.

Args:: config (ToolCallAgentWorkflowConfig): The config to use for the prompt.
Returns:: ChatPromptTemplate: The Tool Calling Agent prompt.