13. LangChain Tools Integration

LangChain provides a mature framework for building LLM applications with tool support. Integrating local models like Ollama with LangChain's tool system requires understanding the adapter patterns and configuration options available.

LangChain Tool Schema

LangChain uses Pydantic models to define tool schemas:

from langchain.tools import BaseTool
from langchain.pydantic_v1 import BaseModel, Field
from typing import Type

class WeatherInput(BaseModel):
    location: str = Field(description="City name to get weather for")
    unit: str = Field(default="celsius", description="Temperature unit: celsius or fahrenheit")

class WeatherTool(BaseTool):
    name = "get_weather"
    description = "Get current weather for a location"
    args_schema: Type[BaseModel] = WeatherInput
    
    def _run(self, location: str, unit: str = "celsius"):
        # Actual weather API call here
        return {"location": location, "temp": 22, "unit": unit}
    
    async def _arun(self, location: str, unit: str = "celsius"):
        # Async version for streaming pipelines
        return self._run(location, unit)

Ollama with LangChain

LangChain supports Ollama as a chat model backend:

from langchain_ollama import ChatOllama
from langchain.callbacks.manager import CallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# Basic configuration
llm = ChatOllama(
    model="llama3.2",
    base_url="http://localhost:11434",
    temperature=0.7,
    # Streaming callbacks for real-time output
    callback_manager=CallbackManager([
        StreamingStdOutCallbackHandler()
    ])
)

# With additional Ollama-specific settings
llm = ChatOllama(
    model="llama3.2",
    base_url="http://localhost:11434",
    # Ollama-specific parameters
    options={
        "num_gpu": 1,
        "num_ctx": 4096,
        "temperature": 0.7,
    }
)

Binding Tools to the Model

LangChain provides the bind_tools method to attach tools to any chat model:

from langchain_core.messages import HumanMessage, SystemMessage

# Define tools
tools = [WeatherTool(), SearchTool(), CalculatorTool()]

# Bind tools to model
llm_with_tools = llm.bind_tools(tools)

# Invoke with tool call
messages = [
    SystemMessage(content="You have access to weather, search, and calculator tools."),
    HumanMessage(content="What's the weather in San Francisco?")
]

response = llm_with_tools.invoke(messages)

# response.additional_kwargs contains tool calls
if hasattr(response, 'additional_kwargs') and 'tool_calls' in response.additional_kwargs:
    tool_calls = response.additional_kwargs['tool_calls']
    for call in tool_calls:
        print(f"Tool: {call['function']['name']}")
        print(f"Args: {call['function']['arguments']}")

Tool Execution with LangChain

LangChain offers several patterns for executing tools:

from langchain_core.output_parser import JsonOutputParser
from langchain.schema import AgentFinish, AgentAction

class ToolExecutor:
    def __init__(self, tools: list[BaseTool]):
        self.tools = {tool.name: tool for tool in tools}
    
    def execute(self, tool_call: dict) -> str:
        tool_name = tool_call['function']['name']
        tool_args = json.loads(tool_call['function']['arguments'])
        
        if tool_name not in self.tools:
            return f"Error: Unknown tool {tool_name}"
        
        tool = self.tools[tool_name]
        try:
            result = tool.run(tool_args)
            return json.dumps(result)
        except Exception as e:
            return f"Error executing {tool_name}: {str(e)}"

# Using LangChain's built-in agent
from langchain.agents import AgentType, initialize_agent

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    verbose=True,
    handle_parsing_errors=True
)

result = agent.invoke("What's the weather in Tokyo and then calculate 15% of 200?")

Streaming with LangChain Agents

LangChain supports streaming at the agent level:

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain.callbacks.manager import CallbackManager

# Streaming agent
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.OPENAI_FUNCTIONS,
    callback_manager=CallbackManager([StreamingStdOutCallbackHandler()])
)

# Stream the entire agent run
async for event in agent.astream_events(
    {"input": "What's the weather in Paris?"},
    version="v1"
):
    if event["event"] == "on_llm_stream":
        print(event["data"]["chunk"].content, end="", flush=True)