File size: 2,617 Bytes
bfb5d08 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
from typing import Sequence
from dotenv import load_dotenv
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts.prompt import PromptTemplate
from langchain.tools.render import render_text_description
from langchain.agents.output_parsers import ReActSingleInputOutputParser
from langchain.agents import AgentExecutor
from langchain.agents.format_scratchpad import format_log_to_str
from schemas.model_llm import ModelLLM
from langchain.globals import set_verbose
import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)
set_verbose(True)
load_dotenv()
class ModelAI:
def __init__(self, model: ModelLLM):
self.model = model.name_model
self.base_url = model.base_url
self.api_key = model.api_key
def agent_executer(self, tools: Sequence[BaseTool]) -> AgentExecutor:
"""
Create an agent executor with the given tools and the model.
Args:
tools: A sequence of tools to be used by the agent.
Returns:
An agent executor with the given tools and the model.
"""
llm = ChatOpenAI(
model=self.model,
api_key=self.api_key,
base_url=self.base_url,
temperature=0.5,
)
memory = ConversationBufferMemory(
memory_key="chat_history", return_messages=False
)
prompt = self._load_prompt("prompt_system_agent.txt")
agent_prompt = PromptTemplate.from_template(prompt)
prompt = agent_prompt.partial(
tools=render_text_description(tools),
tool_names=", ".join([t.name for t in tools]),
)
agent = self._create_agent(llm, prompt)
return AgentExecutor(agent=agent, tools=tools, memory=memory)
@staticmethod
def _load_prompt(filepath: str) -> str:
with open(filepath, "r") as file:
return file.read()
@staticmethod
def _create_agent(llm: ChatOpenAI, prompt: PromptTemplate) -> dict:
llm_with_stop = llm.bind(stop=["\nObservation"])
return (
{
"input": lambda x: x["input"],
"agent_scratchpad": lambda x: format_log_to_str(
x["intermediate_steps"]
),
"chat_history": lambda x: x["chat_history"],
}
| prompt
| llm_with_stop
| ReActSingleInputOutputParser()
)
|