Cohere Tools on LangChain (Integration Guide)

import os
from langchain.agents import create_agent
from langchain_cohere import ChatCohere
from langchain_tavily import TavilySearch
# Internet search tool. Replace the placeholder with your Tavily API key.
os.environ["TAVILY_API_KEY"] = "TAVILY_API_KEY"
internet_search = TavilySearch()
# Define the Cohere LLM
llm = ChatCohere(
    cohere_api_key="COHERE_API_KEY",
    model="command-a-03-2025",
    temperature=0,
)
# System instruction for the agent
system_prompt = """
You are an expert who answers the user's question by searching the internet for the most relevant, up-to-date information.
"""
# Create a multi-step agent, passing the instruction via `system_prompt`
agent = create_agent(
    llm, tools=[internet_search], system_prompt=system_prompt
)
# The agent can search multiple times to answer the question
result = agent.invoke(
    {
        "messages": [
            ("user", "Who is the mayor of the capital of Ontario?")
        ]
    }
)
print(result["messages"][-1].content)

from langchain_cohere import ChatCohere
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel, Field
# Data model
class web_search(BaseModel):
    """
    The internet. Use web_search for questions that are related to anything else than agents, prompt engineering, and adversarial attacks.
    """
    query: str = Field(
        description="The query to use when searching the internet."
    )
class vectorstore(BaseModel):
    """
    A vectorstore containing documents related to agents, prompt engineering, and adversarial attacks. Use the vectorstore for questions on these topics.
    """
    query: str = Field(
        description="The query to use when searching the vectorstore."
    )
# System instruction that tells the model how to route
system_message = SystemMessage(
    content="""You are an expert at routing a user question to a vectorstore or web search.
The vectorstore contains documents related to agents, prompt engineering, and adversarial attacks.
Use the vectorstore for questions on these topics. Otherwise, use web-search."""
)
# Define the Cohere LLM
llm = ChatCohere(
    cohere_api_key="COHERE_API_KEY", model="command-a-03-2025"
)
# Bind the tools to the model
llm_with_tools = llm.bind_tools(tools=[web_search, vectorstore])
# The model routes this question to web search
messages = [
    system_message,
    HumanMessage("Who will the Bears draft first in the NFL draft?"),
]
response = llm_with_tools.invoke(messages)
print(response.tool_calls)
# The model routes this question to the vectorstore
messages = [
    system_message,
    HumanMessage("What are the types of agent memory?"),
]
response = llm_with_tools.invoke(messages)
print(response.tool_calls)
# When no tool is needed, `.tool_calls` is an empty list
messages = [system_message, HumanMessage("Hi, how are you?")]
response = llm_with_tools.invoke(messages)
print(response.tool_calls)

from langchain.agents import create_agent
from langchain_cohere import ChatCohere
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
import urllib.request
# Download the Chinook SQLite database
url = "https://github.com/lerocha/chinook-database/raw/master/ChinookDatabase/DataSources/Chinook_Sqlite.sqlite"
urllib.request.urlretrieve(url, "Chinook.db")
print("Chinook database downloaded successfully.")
db = SQLDatabase.from_uri("sqlite:///Chinook.db")
print(db.dialect)
print(db.get_usable_table_names())
db.run("SELECT * FROM Artist LIMIT 10;")
# Define the Cohere LLM
llm = ChatCohere(
    cohere_api_key="COHERE_API_KEY",
    model="command-a-03-2025",
    temperature=0,
)
# Build a SQL agent from the database toolkit's tools
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
agent_executor = create_agent(llm, tools=toolkit.get_tools())
result = agent_executor.invoke(
    {
        "messages": [
            ("user", "Show me the first 5 rows of the Album table.")
        ]
    }
)
print(result["messages"][-1].content)

from langchain.agents import create_agent
from langchain_cohere import ChatCohere
from langchain_experimental.tools import PythonAstREPLTool
import pandas as pd
import urllib.request
# Download the Titanic CSV and load it into a dataframe
url = "https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv"
urllib.request.urlretrieve(url, "titanic.csv")
df = pd.read_csv("titanic.csv")
# Give the agent a Python REPL with the dataframe (`df`) in scope so it can
# answer arbitrary questions about the CSV by writing pandas code.
python_tool = PythonAstREPLTool(locals={"df": df})
# Define the Cohere LLM
llm = ChatCohere(
    cohere_api_key="COHERE_API_KEY",
    model="command-a-03-2025",
    temperature=0,
)
# Give the model the dataframe's columns and a preview so it knows the schema
# before it writes any pandas code.
system_prompt = (
    "You are a data analyst working with a pandas dataframe named `df`.\n"
    f"The dataframe columns are: {list(df.columns)}.\n"
    f"Here is `df.head()`:\n{df.head().to_string()}\n\n"
    "Answer the user's question by writing pandas code against `df` and running "
    "it with the Python tool, then report the result."
)
agent_executor = create_agent(
    llm, tools=[python_tool], system_prompt=system_prompt
)
result = agent_executor.invoke(
    {"messages": [("user", "How many people were on the titanic?")]}
)
print(result["messages"][-1].content)

from langchain_core.tools import tool
from langchain_cohere import ChatCohere
@tool
def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b
@tool
def multiply(a: int, b: int) -> int:
    """Multiplies a and b."""
    return a * b
tools = [add, multiply]
# Define the Cohere LLM
llm = ChatCohere(
    cohere_api_key="COHERE_API_KEY",
    model="command-a-03-2025",
    temperature=0,
)
llm_with_tools = llm.bind_tools(tools)
query = "What is 3 * 12? Also, what is 11 + 49?"
for chunk in llm_with_tools.stream(query):
    if chunk.tool_call_chunks:
        print(chunk.tool_call_chunks)

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_cohere import ChatCohere
# Create a state graph
class State(TypedDict):
    messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
# Define the Cohere LLM
llm = ChatCohere(
    cohere_api_key="COHERE_API_KEY", model="command-a-03-2025"
)
# Add nodes
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
# Compile the graph
graph = graph_builder.compile()
# Run the chatbot
while True:
    user_input = input("User: ")
    print("User: " + user_input)
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break
    for event in graph.stream({"messages": ("user", user_input)}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

from langchain_tavily import TavilySearch
from langchain_cohere import ChatCohere
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langchain_core.messages import ToolMessage
from langchain_core.messages import BaseMessage
from typing import Annotated, Literal
from typing_extensions import TypedDict
import json
# Create a tool
tool = TavilySearch(max_results=2)
tools = [tool]
# Create a state graph
class State(TypedDict):
    messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
# Define the LLM
llm = ChatCohere(
    cohere_api_key="COHERE_API_KEY", model="command-a-03-2025"
)
# Bind the tools to the LLM
llm_with_tools = llm.bind_tools(tools)
# Add nodes
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
class BasicToolNode:
    """A node that runs the tools requested in the last AIMessage."""
    def __init__(self, tools: list) -> None:
        self.tools_by_name = {tool.name: tool for tool in tools}
    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No message found in input")
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[
                tool_call["name"]
            ].invoke(tool_call["args"])
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}
tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)
def route_tools(
    state: State,
) -> Literal["tools", "__end__"]:
    """
    Use in the conditional_edge to route to the ToolNode if the last message
    has tool calls. Otherwise, route to the end.
    """
    if isinstance(state, list):
        ai_message = state[-1]
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError(
            f"No messages found in input state to tool_edge: {state}"
        )
    if (
        hasattr(ai_message, "tool_calls")
        and len(ai_message.tool_calls) > 0
    ):
        return "tools"
    return "__end__"
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", "__end__": "__end__"},
)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
# Compile the graph
graph = graph_builder.compile()
# Run the chatbot
while True:
    user_input = input("User: ")
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break
    for event in graph.stream({"messages": [("user", user_input)]}):
        for value in event.values():
            if isinstance(value["messages"][-1], BaseMessage):
                print("Assistant:", value["messages"][-1].content)