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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
| import json
import re
from chat_bot import ChatBot
from typing import Callable, Dict
def calculator(expression: str) -> str:
try:
return str(eval(expression))
except Exception as e:
return f"Calculator Error: {str(e)}"
def reverse_text(text: str) -> str:
return text[::-1]
class Agent:
def __init__(
self,
chatbot: ChatBot,
tools: Dict[str, Callable] = None,
):
self.chatbot = chatbot
self.tools = tools or {
"calculator": calculator,
"reverse_text": reverse_text,
}
self.chatbot.clear_chat_history()
self.chatbot.messages.append({
"role": "system",
"content": self._build_agent_prompt()
})
def _build_agent_prompt(self):
tool_descriptions = "\n".join([
"- calculator: evaluates math expressions. Params: {'expression': str}",
"- reverse_text: reverses a string. Params: {'text': str}",
])
return f"""You are a structured reasoning agent. You can use the following tools:
{tool_descriptions}
You run in a loop of Reasoning, Action, and Execution. If you have enough information to answer the question, you should do so. Otherwise, you will carefully oberve, reason, and use one of the tools to get the information you need.
1. Reason and Action: In this step, you will reason about the question and decide which tool to use and call one of the tools with the required parameters. You must reply using this format:
<thought>
your current reasoning
</thought>
<action>
{{"function_name": "tool_name", "function_params": {{"param1": "...", "param2": "..."}}}}
</action>
After you call a tool, you will receive an observation.
2. Observation: It is a passive step where the output of the tool will be returned in the role of User. The observation will be in the format:
<observation>
tool execution output here
</observation>
Only use one tool at a time. Always use valid JSON for <action>. After you receive the observation, you will reason again and decide. If you can answer the question based on existing information or you sense it is a insolvable question, then go to step 3. Otherwise, if you cannot answer the question based on existing information and you still think it is a olvable question, you will go to step 1.
3. Reason and Answer: After you receive the observation, you will reason again and decide if you can answer the question. If you can, reply with:
<thought>
your current reasoning
</thought>
<answer>
your final answer
</answer>
"""
def __call__(self, user_input: str):
self.chatbot.messages.append({"role": "user", "content": f"User Input: {user_input}"})
while True:
response = self.chatbot.execute()
final_match = re.search(r"<answer>(.*?)</answer>", response, re.DOTALL)
if final_match:
print(f'---\n[AI final response]:\n{response}\n---')
self.chatbot.messages.append({"role": "assistant", "content": response})
return final_match.group(1).strip()
action_match = re.search(r"<action>\n\s*(\{.*?\})\s*\n</action>", response, re.DOTALL)
if not action_match:
return "Invalid format: <action> block not found."
try:
action_dict = json.loads(action_match.group(1))
func_name = action_dict["function_name"]
params = action_dict["function_params"]
assert isinstance(params, dict)
except Exception as e:
return f"Error parsing <action>: {e}"
if func_name not in self.tools:
observation = f"Unknown tool: {func_name}"
else:
try:
observation = self.tools[func_name](**params)
observation = f"tool execution results: \n{observation}"
except Exception as e:
observation = f"Error during tool execution: {e}"
appending_new_ai_message = {
"role": "assistant",
"content": response
}
appending_new_tool_message = {
"role": "user",
"content": f"<observation>\n{observation}\n</observation>"
}
print(f'---\n[AI intermediate response]:\n{appending_new_ai_message["content"]}\n---')
print(f'---\n[tool intermediate response]:\n{appending_new_tool_message["content"]}\n---')
self.chatbot.messages.append(appending_new_ai_message)
self.chatbot.messages.append(appending_new_tool_message)
if __name__ == "__main__":
agent = Agent(ChatBot())
print(agent("What is the result of 12 * 7 + 3?"))
|
