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)
|
