Cohere Tools on LangChain (Integration Guide)

1
import os
2
3
from langchain.agents import create_agent
4
from langchain_cohere import ChatCohere
5
from langchain_tavily import TavilySearch
6
7
# Internet search tool. Replace the placeholder with your Tavily API key.
8
os.environ["TAVILY_API_KEY"] = "TAVILY_API_KEY"
9
10
internet_search = TavilySearch()
11
12
# Define the Cohere LLM
13
llm = ChatCohere(
14
    cohere_api_key="COHERE_API_KEY",
15
    model="command-a-03-2025",
16
    temperature=0,
17
)
18
19
# System instruction for the agent
20
system_prompt = """
21
You are an expert who answers the user's question by searching the internet for the most relevant, up-to-date information.
22
"""
23
24
# Create a multi-step agent, passing the instruction via `system_prompt`
25
agent = create_agent(
26
    llm, tools=[internet_search], system_prompt=system_prompt
27
)
28
29
# The agent can search multiple times to answer the question
30
result = agent.invoke(
31
    {
32
        "messages": [
33
            ("user", "Who is the mayor of the capital of Ontario?")
34
        ]
35
    }
36
)
37
38
print(result["messages"][-1].content)

1
from langchain_cohere import ChatCohere
2
from langchain_core.messages import HumanMessage, SystemMessage
3
from pydantic import BaseModel, Field
4
5
6
# Data model
7
class web_search(BaseModel):
8
    """
9
    The internet. Use web_search for questions that are related to anything else than agents, prompt engineering, and adversarial attacks.
10
    """
11
12
    query: str = Field(
13
        description="The query to use when searching the internet."
14
    )
15
16
17
class vectorstore(BaseModel):
18
    """
19
    A vectorstore containing documents related to agents, prompt engineering, and adversarial attacks. Use the vectorstore for questions on these topics.
20
    """
21
22
    query: str = Field(
23
        description="The query to use when searching the vectorstore."
24
    )
25
26
27
# System instruction that tells the model how to route
28
system_message = SystemMessage(
29
    content="""You are an expert at routing a user question to a vectorstore or web search.
30
The vectorstore contains documents related to agents, prompt engineering, and adversarial attacks.
31
Use the vectorstore for questions on these topics. Otherwise, use web-search."""
32
)
33
34
# Define the Cohere LLM
35
llm = ChatCohere(
36
    cohere_api_key="COHERE_API_KEY", model="command-a-03-2025"
37
)
38
39
# Bind the tools to the model
40
llm_with_tools = llm.bind_tools(tools=[web_search, vectorstore])
41
42
# The model routes this question to web search
43
messages = [
44
    system_message,
45
    HumanMessage("Who will the Bears draft first in the NFL draft?"),
46
]
47
response = llm_with_tools.invoke(messages)
48
print(response.tool_calls)
49
50
# The model routes this question to the vectorstore
51
messages = [
52
    system_message,
53
    HumanMessage("What are the types of agent memory?"),
54
]
55
response = llm_with_tools.invoke(messages)
56
print(response.tool_calls)
57
58
# When no tool is needed, `.tool_calls` is an empty list
59
messages = [system_message, HumanMessage("Hi, how are you?")]
60
response = llm_with_tools.invoke(messages)
61
print(response.tool_calls)

1
from langchain.agents import create_agent
2
from langchain_cohere import ChatCohere
3
from langchain_community.agent_toolkits import SQLDatabaseToolkit
4
from langchain_community.utilities import SQLDatabase
5
import urllib.request
6
7
# Download the Chinook SQLite database
8
url = "https://github.com/lerocha/chinook-database/raw/master/ChinookDatabase/DataSources/Chinook_Sqlite.sqlite"
9
urllib.request.urlretrieve(url, "Chinook.db")
10
print("Chinook database downloaded successfully.")
11
12
db = SQLDatabase.from_uri("sqlite:///Chinook.db")
13
print(db.dialect)
14
print(db.get_usable_table_names())
15
db.run("SELECT * FROM Artist LIMIT 10;")
16
17
# Define the Cohere LLM
18
llm = ChatCohere(
19
    cohere_api_key="COHERE_API_KEY",
20
    model="command-a-03-2025",
21
    temperature=0,
22
)
23
24
# Build a SQL agent from the database toolkit's tools
25
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
26
agent_executor = create_agent(llm, tools=toolkit.get_tools())
27
28
result = agent_executor.invoke(
29
    {
30
        "messages": [
31
            ("user", "Show me the first 5 rows of the Album table.")
32
        ]
33
    }
34
)
35
print(result["messages"][-1].content)

1
from langchain.agents import create_agent
2
from langchain_cohere import ChatCohere
3
from langchain_experimental.tools import PythonAstREPLTool
4
import pandas as pd
5
import urllib.request
6
7
# Download the Titanic CSV and load it into a dataframe
8
url = "https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv"
9
urllib.request.urlretrieve(url, "titanic.csv")
10
df = pd.read_csv("titanic.csv")
11
12
# Give the agent a Python REPL with the dataframe (`df`) in scope so it can
13
# answer arbitrary questions about the CSV by writing pandas code.
14
python_tool = PythonAstREPLTool(locals={"df": df})
15
16
# Define the Cohere LLM
17
llm = ChatCohere(
18
    cohere_api_key="COHERE_API_KEY",
19
    model="command-a-03-2025",
20
    temperature=0,
21
)
22
23
# Give the model the dataframe's columns and a preview so it knows the schema
24
# before it writes any pandas code.
25
system_prompt = (
26
    "You are a data analyst working with a pandas dataframe named `df`.\n"
27
    f"The dataframe columns are: {list(df.columns)}.\n"
28
    f"Here is `df.head()`:\n{df.head().to_string()}\n\n"
29
    "Answer the user's question by writing pandas code against `df` and running "
30
    "it with the Python tool, then report the result."
31
)
32
33
agent_executor = create_agent(
34
    llm, tools=[python_tool], system_prompt=system_prompt
35
)
36
37
result = agent_executor.invoke(
38
    {"messages": [("user", "How many people were on the titanic?")]}
39
)
40
print(result["messages"][-1].content)

1
from langchain_core.tools import tool
2
from langchain_cohere import ChatCohere
3
4
5
@tool
6
def add(a: int, b: int) -> int:
7
    """Adds a and b."""
8
    return a + b
9
10
11
@tool
12
def multiply(a: int, b: int) -> int:
13
    """Multiplies a and b."""
14
    return a * b
15
16
17
tools = [add, multiply]
18
19
# Define the Cohere LLM
20
llm = ChatCohere(
21
    cohere_api_key="COHERE_API_KEY",
22
    model="command-a-03-2025",
23
    temperature=0,
24
)
25
26
llm_with_tools = llm.bind_tools(tools)
27
28
query = "What is 3 * 12? Also, what is 11 + 49?"
29
30
for chunk in llm_with_tools.stream(query):
31
    if chunk.tool_call_chunks:
32
        print(chunk.tool_call_chunks)

1
from typing import Annotated
2
from typing_extensions import TypedDict
3
from langgraph.graph import StateGraph, START, END
4
from langgraph.graph.message import add_messages
5
from langchain_cohere import ChatCohere
6
7
8
# Create a state graph
9
class State(TypedDict):
10
    messages: Annotated[list, add_messages]
11
12
13
graph_builder = StateGraph(State)
14
15
# Define the Cohere LLM
16
llm = ChatCohere(
17
    cohere_api_key="COHERE_API_KEY", model="command-a-03-2025"
18
)
19
20
21
# Add nodes
22
def chatbot(state: State):
23
    return {"messages": [llm.invoke(state["messages"])]}
24
25
26
graph_builder.add_node("chatbot", chatbot)
27
graph_builder.add_edge(START, "chatbot")
28
graph_builder.add_edge("chatbot", END)
29
30
# Compile the graph
31
graph = graph_builder.compile()
32
33
# Run the chatbot
34
while True:
35
    user_input = input("User: ")
36
    print("User: " + user_input)
37
    if user_input.lower() in ["quit", "exit", "q"]:
38
        print("Goodbye!")
39
        break
40
    for event in graph.stream({"messages": ("user", user_input)}):
41
        for value in event.values():
42
            print("Assistant:", value["messages"][-1].content)

1
from langchain_tavily import TavilySearch
2
from langchain_cohere import ChatCohere
3
from langgraph.graph import StateGraph, START
4
from langgraph.graph.message import add_messages
5
from langchain_core.messages import ToolMessage
6
from langchain_core.messages import BaseMessage
7
from typing import Annotated, Literal
8
from typing_extensions import TypedDict
9
import json
10
11
# Create a tool
12
tool = TavilySearch(max_results=2)
13
tools = [tool]
14
15
16
# Create a state graph
17
class State(TypedDict):
18
    messages: Annotated[list, add_messages]
19
20
21
graph_builder = StateGraph(State)
22
23
# Define the LLM
24
llm = ChatCohere(
25
    cohere_api_key="COHERE_API_KEY", model="command-a-03-2025"
26
)
27
28
# Bind the tools to the LLM
29
llm_with_tools = llm.bind_tools(tools)
30
31
32
# Add nodes
33
def chatbot(state: State):
34
    return {"messages": [llm_with_tools.invoke(state["messages"])]}
35
36
37
graph_builder.add_node("chatbot", chatbot)
38
39
40
class BasicToolNode:
41
    """A node that runs the tools requested in the last AIMessage."""
42
43
    def __init__(self, tools: list) -> None:
44
        self.tools_by_name = {tool.name: tool for tool in tools}
45
46
    def __call__(self, inputs: dict):
47
        if messages := inputs.get("messages", []):
48
            message = messages[-1]
49
        else:
50
            raise ValueError("No message found in input")
51
        outputs = []
52
        for tool_call in message.tool_calls:
53
            tool_result = self.tools_by_name[
54
                tool_call["name"]
55
            ].invoke(tool_call["args"])
56
            outputs.append(
57
                ToolMessage(
58
                    content=json.dumps(tool_result),
59
                    name=tool_call["name"],
60
                    tool_call_id=tool_call["id"],
61
                )
62
            )
63
        return {"messages": outputs}
64
65
66
tool_node = BasicToolNode(tools=[tool])
67
graph_builder.add_node("tools", tool_node)
68
69
70
def route_tools(
71
    state: State,
72
) -> Literal["tools", "__end__"]:
73
    """
74
    Use in the conditional_edge to route to the ToolNode if the last message
75
    has tool calls. Otherwise, route to the end.
76
    """
77
    if isinstance(state, list):
78
        ai_message = state[-1]
79
    elif messages := state.get("messages", []):
80
        ai_message = messages[-1]
81
    else:
82
        raise ValueError(
83
            f"No messages found in input state to tool_edge: {state}"
84
        )
85
    if (
86
        hasattr(ai_message, "tool_calls")
87
        and len(ai_message.tool_calls) > 0
88
    ):
89
        return "tools"
90
    return "__end__"
91
92
93
graph_builder.add_conditional_edges(
94
    "chatbot",
95
    route_tools,
96
    {"tools": "tools", "__end__": "__end__"},
97
)
98
graph_builder.add_edge("tools", "chatbot")
99
graph_builder.add_edge(START, "chatbot")
100
101
# Compile the graph
102
graph = graph_builder.compile()
103
104
# Run the chatbot
105
while True:
106
    user_input = input("User: ")
107
    if user_input.lower() in ["quit", "exit", "q"]:
108
        print("Goodbye!")
109
        break
110
    for event in graph.stream({"messages": [("user", user_input)]}):
111
        for value in event.values():
112
            if isinstance(value["messages"][-1], BaseMessage):
113
                print("Assistant:", value["messages"][-1].content)