Skip to main content

adapters.langchain_adapter

LangChain agent adapter.

Wraps a LangChain agent/chain to implement the BaseAgent interface, enabling use with protocol adapters.

LangChainAdapter Objects​

class LangChainAdapter(BaseAgent)

Adapter for LangChain agents.

Wraps a LangChain agent or chain to provide a consistent interface for protocol adapters.

Example:

from langchain.agents import AgentExecutor from langchain.tools import Tool

Create LangChain agent

langchain_agent = AgentExecutor(...)

agent = LangChainAgent( agent=langchain_agent, name="langchain_agent", description="A LangChain powered assistant", )

response = await agent.run("Hello!", context)

__init__​

def __init__(agent: Any,
             name: str = "langchain_agent",
             description: str = "A LangChain powered agent",
             version: str = "1.0.0")

Initialize the LangChain agent adapter.

Arguments:

  • agent - The LangChain agent/chain instance.
  • name - Agent name.
  • description - Agent description.
  • version - Agent version.

name​

@property
def name() -> str

Get the agent name.

description​

@property
def description() -> str

Get the agent description.

version​

@property
def version() -> str

Get the agent version.

tools​

@property
def tools() -> list[ToolDefinition]

Get available tools.

get_tools​

def get_tools() -> list[ToolDefinition]

Get the list of tools available to this agent.

Returns:

List of tool definitions.

initialize​

async def initialize() -> None

Initialize the agent.

run​

async def run(prompt: str, context: AgentContext) -> AgentResponse

Run the agent with a prompt.

Arguments:

  • prompt - User prompt.
  • context - Execution context.

Returns:

Agent response.

stream​

async def stream(prompt: str,
                 context: AgentContext) -> AsyncIterator[StreamEvent]

Stream agent responses.

Arguments:

  • prompt - User prompt.
  • context - Execution context.

Yields:

Stream events.

cancel​

async def cancel() -> None

Cancel the current execution.

cleanup​

async def cleanup() -> None

Clean up resources.