LangChain에서 Cohere Tools 사용하기
LangChain에서 Cohere Tools 사용하기 (통합 가이드)
챗봇에서 다단계 및 단일 단계 도구 사용을 위한 코드 예시를 알아볼 거예요. 인터넷 검색과 벡터 저장소를 활용한답니다.
Cohere는 Cohere의 모델을 기반으로 애플리케이션을 빠르게 만들 수 있게 해 주는 대규모 언어 모델(LLM) 프레임워크인 LangChain과 다양한 통합을 지원해요. 이 문서는 LangChain과 함께 Cohere tools를 활용하는 방법을 안내할 거예요.
출처: 문서
사전 요구 사항 (Prerequisites)
LangChain으로 Cohere tools를 실행하는 데는 사전 요구 사항이 많지 않아요. 자세한 내용은 최상위 문서를 참조하세요.
다단계 도구 사용 (Multi-Step Tool Use)
LangChain v1에서 다단계 에이전트를 구축하는 관용적인 방법은 langchain.agents의 create_agent예요 (langchain에 포함되어 있으므로 추가 설치가 필요 없어요). 에이전트는 도구를 반복 호출할 수 있고, 최종 답변을 반환하기 전에 여러 단계에 걸쳐 추론해요. 여기서는 에이전트에게 인터넷 검색 도구(Tavily)를 줍니다. 실행하려면 pip install langchain-tavily로 설치하고 TAVILY_API_KEY 환경 변수를 설정하세요. 원하는 다른 LangChain 도구로 바꿀 수도 있어요. create_agent에 system_prompt 인자를 통해 시스템 지시를 전달하면 에이전트의 동작을 조정할 수 있어요.
PYTHON
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)
단일 단계 도구 사용 (Single-Step Tool Use)
단일 단계 도구 사용은 모델이 쿼리에 대해 어떤 도구를 호출할지 실행하지 않고 결정하게 해 줘요. 도구를 모델에 바인딩하고, 응답의 .tool_calls 속성에서 선택된 도구 호출을 읽으면 돼요. 대화 시작 부분에 SystemMessage로 라우팅 지시를 제공하세요.
PYTHON
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)
SQL 에이전트 (SQL Agent)
LangChain의 SQLDatabaseToolkit의 도구를 create_agent에 주면 SQL 데이터베이스와 상호작용하는 에이전트를 구축할 수 있어요.
PYTHON
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)
CSV 에이전트 (CSV Agent)
CSV 파일을 pandas 데이터프레임으로 로드하고, 데이터프레임이 범위에 있는 Python REPL 도구를 create_agent에 주면 CSV 파일에 대한 질문에 답하는 에이전트를 구축할 수 있어요. 그러면 에이전트가 pandas 코드를 작성·실행해 데이터에 대한 임의의 질문에 답할 수 있어요 (pip install langchain-experimental pandas 설치 필요).
참고 (Note)
Python REPL 도구는 모델이 생성한 코드를 실행하므로, 신뢰할 수 있는 데이터와 쿼리에만 사용하세요.
PYTHON
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)
도구 호출을 위한 스트리밍 (Streaming for Tool Calling)
스트리밍 컨텍스트에서 도구가 호출되면 받은 메시지 청크에 .tool_call_chunks 속성을 통해 도구 호출 청크 객체 목록이 채워져요.
PYTHON
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)
LangGraph 에이전트 (LangGraph Agents)
LangGraph는 에이전트 워크플로에 추가 제어를 제공하는 상태 저장(stateful) 오케스트레이션 프레임워크예요.
Cohere와 LangGraph를 사용하려면 LangGraph 패키지를 설치해야 해요. 설치하려면 pip install langgraph를 실행하세요.
기본 챗봇 (Basic Chatbot)
이 간단한 챗봇 예시는 LangGraph로 구축하는 핵심 개념을 보여줄 거예요.
PYTHON
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)
도구로 챗봇 강화하기 (Enhancing the Chatbot with Tools)
챗봇이 "기억(memory)"으로 답할 수 없는 쿼리를 처리하기 위해 웹 검색 도구를 통합할 거예요. 우리 봇은 이 도구를 사용해 관련 정보를 찾고 더 나은 응답을 제공할 수 있어요.
PYTHON
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)