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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
| import ollama
import re
import logging
from sympy import sympify
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
client = ollama.Client(host='http://192.168.100.135:11434')
def lookup_fact(keyword):
"""
根据关键词检索预定义的事实。
"""
logger.info(f"正在查找事实: {keyword}")
facts = {
"光速": "光速约为299,792,458米每秒。",
"普朗克常数": "普朗克常数约为6.62607015 × 10^-34 Js。",
"地球质量": "地球的质量约为5.972 × 10^24千克。",
"月球质量": "月球的质量约为7.342 × 10^22千克。",
"speed of light": "光速约为299,792,458米每秒。",
"planck's constant": "普朗克常数约为6.62607015 × 10^-34 Js。",
"mass of earth": "地球的质量约为5.972 × 10^24千克。",
"mass of moon": "月球的质量约为7.342 × 10^22千克。"
}
result = facts.get(keyword.lower(), "未找到相关信息")
logger.info(f"查找结果: {result}")
return result
def calculate(expression):
"""
安全地计算数学表达式。
"""
logger.info(f"正在计算: {expression}")
try:
result = float(sympify(expression))
logger.info(f"计算结果: {result}")
return result
except Exception as e:
logger.error(f"计算错误: {e}")
return f"计算错误: {e}"
class ToolRegistry:
def __init__(self):
self.tools = {}
def register(self, name, description, handler):
logger.debug(f"Registering tool: {name}, Description: {description}")
self.tools[name] = {"description": description, "handler": handler}
def get_handler(self, name):
handler = self.tools.get(name, {}).get("handler")
if handler:
logger.debug(f"Found handler for tool: {name}")
else:
logger.warning(f"Handler for tool not found: {name}")
return handler
tool_registry = ToolRegistry()
tool_registry.register("lookup_fact", "根据关键词检索预定义事实（有效键值：光速、普朗克常数、地球质量、月球质量）", lookup_fact)
tool_registry.register("calculate", "执行数学表达式计算", calculate)
def generate_prompt(tools):
tool_descriptions = "\n".join(
[f"{i+1}. {name}: 功能: {desc['description']} 使用格式: {name}: <参数>"
for i, (name, desc) in enumerate(tools.items())]
)
return (
"你将扮演一个ReAct代理，逐步解决用户的问题，严格按照以下步骤执行:\n"
"工作流程:\n"
"1. 思考: 描述你对给定问题的思考过程。\n"
"2. 行动: 选择一个可用工具并执行操作。每次只执行一个操作，等待观察结果后再继续。\n"
"3. 观察: 记录工具返回的结果。\n"
"4. 答案: 问题解决后输出最终答案。\n"
"可用工具:\n"
f"{tool_descriptions}\n"
"对话示例:\n"
"问题: 查找地球和月球的质量，并计算它们的质量比。\n"
"思考: 我需要查找地球和月球的质量。\n"
"行动: lookup_fact 地球质量\n"
"观察: 地球的质量约为5.972 × 10^24千克。\n"
"思考: 接下来，我需要查找月球的质量。\n"
"行动: lookup_fact 月球质量\n"
"观察: 月球的质量约为7.342 × 10^22千克。\n"
"思考: 现在我已经得到了地球和月球的质量，可以计算它们的质量比。\n"
"行动: calculate 5.972e24 / 7.342e22\n"
"观察: 结果是81.35215091343292。\n"
"答案: 地球的质量大约是月球质量的81.35倍。\n"
"\n重要: 每次响应只能提供单行内容(思考或行动或观察)。如果没有需要执行的操作，则继续推理。"
).strip()
prompt = generate_prompt(tool_registry.tools)
action_re = re.compile(r'^Action:\s*(\w+)\s+(.*)$', re.IGNORECASE | re.MULTILINE)
class ChatBot:
def __init__(self, system=""):
self.messages = [{"role": "system", "content": system}] if system else []
def __call__(self, message):
self.messages.append({"role": "user", "content": message})
try:
logger.debug(f"发送消息到模型: {message}")
response = client.chat(model="llama3.2:1b", messages=self.messages)
content = response['message']['content'].splitlines()[0]
logger.debug(f"模型响应: {content}")
self.messages.append({"role": "assistant", "content": content})
return content
except Exception as e:
logger.error(f"API调用失败: {e}")
return "发生错误，无法继续对话。"
def trim_messages(self, max_messages=10):
logger.debug(f"修剪消息历史以保留最后 {max_messages} 条消息")
self.messages = self.messages[-max_messages:]
def query(question, max_turns=5):
bot = ChatBot(prompt)
next_prompt = question
for turn in range(max_turns):
logger.info(f"---- 第 {turn + 1} 轮 ----")
logger.debug(f"当前提示: {next_prompt}")
result = bot(next_prompt)
logger.info(f"AI响应: {result}")
actions = [action_re.match(line) for line in result.splitlines() if action_re.match(line)]
logger.debug(f"找到的行动匹配: {actions}")
if actions:
action, action_input = actions[0].groups()
logger.debug(f"提取的行动: {action}, 参数: {action_input}")
handler = tool_registry.get_handler(action)
if handler:
logger.info(f" -- 执行行动: {action} 参数: {action_input}")
observation = handler(action_input)
observation_message = f"观察: {observation}"
logger.info(f"观察结果: {observation_message}")
next_prompt = f"{result}\n{observation_message}\n思考:"
bot.trim_messages()
bot.messages.append({"role": "assistant", "content": observation_message})
else:
logger.error(f"未知行动: {action}")
break
else:
next_prompt = f"{result}\n思考:"
bot.trim_messages()
if __name__ == "__main__":
question = "查找光速和普朗克常数的值，并计算它们的乘积。"
query(question)
|
