-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
101 lines (80 loc) · 3.03 KB
/
main.py
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
from dotenv import load_dotenv
from langchain.agents import tool
from langchain.tools import Tool
from langchain.prompts import PromptTemplate
from langchain.tools.render import render_text_description
from langchain_openai import ChatOpenAI
from langchain.agents.output_parsers import ReActSingleInputOutputParser
from langchain.agents.format_scratchpad import format_log_to_str
from langchain.schema import AgentAction, AgentFinish
from typing import Union, List
load_dotenv()
@tool
def get_text_length(text: str) -> int:
"""returns the length of a text by characters"""
print(f"get_text_legth enter with {text=}")
return len(text)
def find_tool_by_name(tools: List[Tool], tool_name: str) -> Tool:
for tool in tools:
if tool.name == tool_name:
return tool
raise ValueError(f"Tool with name {tool_name} not found")
if __name__ == "__main__":
print("Hello ReAct Langchain")
tools = [get_text_length]
template = """
Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought: {agent_scratchpad}
"""
prompt = PromptTemplate.from_template(template=template).partial(
tools=render_text_description(tools),
tool_names=", ".join([t.name for t in tools]),
)
llm = ChatOpenAI(temperature=0, model="gpt-4-0125-preview").bind(
stop=["\nObservation:"]
)
# llm = ChatOpenAI(temperature=0).bind(stop=["Observation:"])
intermediate_steps = []
agent = (
{
"input": lambda x: x["input"],
"agent_scratchpad": lambda x: format_log_to_str(x["agent_scratchpad"]),
}
| prompt
| llm
| ReActSingleInputOutputParser()
)
agent_step: Union[AgentAction, AgentFinish] = agent.invoke(
{
"input": "What is the legth of the text: elephant?",
"agent_scratchpad": intermediate_steps,
}
)
print(agent_step)
if isinstance(agent_step, AgentAction):
tool_name = agent_step.tool
tool_to_use = find_tool_by_name(tools, tool_name)
tool_input = agent_step.tool_input
observation = tool_to_use.func(str(tool_input))
print(f"{observation=}")
intermediate_steps.append((agent_step, str(observation)))
agent_step: Union[AgentAction, AgentFinish] = agent.invoke(
{
"input": "What is the legth of the text: elephant?",
"agent_scratchpad": intermediate_steps,
}
)
if isinstance(agent_step, AgentFinish):
print(agent_step.return_values)