Source code for langchain_cohere.csv_agent.agent

from datetime import datetime
from io import IOBase
from typing import List, Optional, Union

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.language_models import BaseLanguageModel
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_core.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
)
from langchain_core.prompts.chat import (
    BaseMessagePromptTemplate,
    HumanMessagePromptTemplate,
)
from langchain_core.tools import BaseTool

from langchain_cohere.chat_models import ChatCohere
from langchain_cohere.csv_agent.prompts import (
    CSV_PREAMBLE,
)

# lets define a set of tools for the Agent
from langchain_cohere.csv_agent.tools import (
    get_file_peek_tool,
    get_file_read_tool,
    get_python_tool,
)


[docs] def create_prompt( system_message: Optional[BaseMessage] = SystemMessage( content="You are a helpful AI assistant." ), extra_prompt_messages: Optional[List[BaseMessagePromptTemplate]] = None, ) -> ChatPromptTemplate: """Create prompt for this agent. Args: system_message: Message to use as the system message that will be the first in the prompt. extra_prompt_messages: Prompt messages that will be placed between the system message and the new human input. Returns: A prompt template to pass into this agent. """ _prompts = extra_prompt_messages or [] messages: List[Union[BaseMessagePromptTemplate, BaseMessage]] if system_message: messages = [system_message] else: messages = [] messages.extend( [ *_prompts, HumanMessagePromptTemplate.from_template("{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ] ) return ChatPromptTemplate(messages=messages)
def _get_csv_head_str(path: str, number_of_head_rows: int) -> str: with open(path, "r") as file: lines = [] for _ in range(number_of_head_rows): lines.append(file.readline().strip("\n")) # validate that the head contents are well formatted csv return " ".join(lines) def _get_prompt( path: Union[str, List[str]], number_of_head_rows: int ) -> ChatPromptTemplate: if isinstance(path, str): lines = _get_csv_head_str(path, number_of_head_rows) prompt_message = f"The user uploaded the following attachments:\nFilename: {path}\nWord Count: {count_words_in_file(path)}\nPreview: {lines}" # noqa: E501 elif isinstance(path, list): prompt_messages = [] for file_path in path: lines = _get_csv_head_str(file_path, number_of_head_rows) prompt_messages.append( f"The user uploaded the following attachments:\nFilename: {file_path}\nWord Count: {count_words_in_file(file_path)}\nPreview: {lines}" # noqa: E501 ) prompt_message = " ".join(prompt_messages) prompt = create_prompt(system_message=HumanMessage(prompt_message)) return prompt
[docs] def count_words_in_file(file_path: str) -> int: try: with open(file_path, "r") as file: content = file.readlines() words = [len(sentence.split()) for sentence in content] return sum(words) except FileNotFoundError: print("File not found.") return 0 except Exception as e: print("An error occurred:", str(e)) return 0
[docs] def create_csv_agent( llm: BaseLanguageModel, path: Union[str, List[str]], extra_tools: List[BaseTool] = [], pandas_kwargs: Optional[dict] = None, prompt: Optional[ChatPromptTemplate] = None, number_of_head_rows: int = 5, verbose: bool = True, return_intermediate_steps: bool = True, temp_path_dir: Optional[str] = None, temp_path_prefix: Optional[str] = "langchain", temp_path_suffix: Optional[str] = "csv_agent", ) -> AgentExecutor: """Create csv agent with the specified language model. Args: llm: Language model to use for the agent. path: A string path, or a list of string paths that can be read in as pandas DataFrames with pd.read_csv(). number_of_head_rows: Number of rows to display in the prompt for sample data include_df_in_prompt: Display the DataFrame sample values in the prompt. pandas_kwargs: Named arguments to pass to pd.read_csv(). prefix: Prompt prefix string. suffix: Prompt suffix string. prompt: Prompt to use for the agent. This takes precedence over the other prompt arguments, such as suffix and prefix. temp_path_dir: Temporary directory to store the csv files in for the python repl. delete_temp_path: Whether to delete the temporary directory after the agent is done. This only works if temp_path_dir is not provided. Returns: An AgentExecutor with the specified agent_type agent and access to a PythonREPL and any user-provided extra_tools. Example: .. code-block:: python from langchain_cohere import ChatCohere, create_csv_agent llm = ChatCohere(model="command-r-plus", temperature=0) agent_executor = create_csv_agent( llm, "titanic.csv" ) resp = agent_executor.invoke({"input":"How many people were on the titanic?"}) print(resp.get("output")) """ # noqa: E501 try: import pandas as pd except ImportError: raise ImportError( "pandas package not found, please install with `pip install pandas`." ) _kwargs = pandas_kwargs or {} if isinstance(path, (str)): df = pd.read_csv(path, **_kwargs) elif isinstance(path, list): df = [] for item in path: if not isinstance(item, (str, IOBase)): raise ValueError(f"Expected str or file-like object, got {type(path)}") df.append(pd.read_csv(item, **_kwargs)) else: raise ValueError(f"Expected str, list, or file-like object, got {type(path)}") if not prompt: prompt = _get_prompt(path, number_of_head_rows) final_tools = [ get_file_read_tool(), get_file_peek_tool(), get_python_tool(), ] + extra_tools if "preamble" in llm.__dict__ and not llm.__dict__.get("preamble"): llm = ChatCohere(**llm.__dict__) llm.preamble = CSV_PREAMBLE.format( current_date=datetime.now().strftime("%A, %B %d, %Y %H:%M:%S") ) agent = create_tool_calling_agent(llm=llm, tools=final_tools, prompt=prompt) agent_executor = AgentExecutor( agent=agent, tools=final_tools, verbose=verbose, return_intermediate_steps=return_intermediate_steps, ) return agent_executor