# -- coding: utf-8 -- import warnings from datetime import datetime from dotenv import load_dotenv
warnings.filterwarnings(“ignore”, category=DeprecationWarning) warnings.filterwarnings(“ignore”, category=UserWarning)
from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import Tool from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.prompts import PromptTemplate
load_dotenv()
=== LLM ===
DEEPSEEK_API_KEY = “123” llm = ChatOpenAI(
api_key=DEEPSEEK_API_KEY, base_url="http://172.25.140.23:8003/v1", model="/data/yangfengyi/Qwen3-235B-A22B-Instruct-2507", temperature=0.1, max_tokens=1024
)
=== 工具 ===
def safe_calculate(expr: str) -> str:
try: allowed = set("0+-*/(). ") if not all(c in allowed for c in expr): return "错误:非法字符" return str(eval(expr, {"__builtins__": {}}, {})) except Exception as e: return f"错误: {e}"
def get_currenttime() -> str:
return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")
tools = [
Tool(name="Calculator", func=safe_calculate, description="数学计算,输入算式如 1+1"), Tool(name="CurrentTime", func=get_current_time, description="获取当前时间"),
]
=== 记忆 ===
memory = ConversationBufferMemory(
memory_key="chat_history", return_messages=True
)
======================= 核心修复:标准 ReAct 格式 =======================
prompt = PromptTemplate.from_template(“”” 你是一个友好的AI助手,有记忆能力。 回答必须严格按照以下格式,一步一步来:
Question: 用户问题 Thought: 分析问题,决定是否需要工具 Action: 工具名称,必须是 [{tool_names}],不需要工具就不写Action Action Input: 工具输入 Observation: 工具返回结果 …(可循环多轮) Thought: 我现在可以给出最终答案 Final Answer: 直接回答用户
规则:
- 日常聊天、问候、自我介绍,不需要工具,直接 Final Answer
- 计算、问时间,才调用对应工具
- 绝对不要调用不存在的工具,不要写 Action: None
- 得到结果后必须用 Final Answer 结束
可用工具: {tools}
历史对话: {chat_history}
Question: {input} Thought:{agent_scratchpad} “”“)
=== Agent === 注:langchain ==0.1.20用create_react_agent, langchain ≥ 0.2.0 使用 create_conversational_react_agent 效果一样
agent = create_react_agent(llm, tools, prompt) agent_executor = AgentExecutor(
agent=agent, tools=tools, memory=memory, verbose=True, handle_parsing_errors=True, max_iterations=5 # 防止死循环
)
=== 运行 ===
if name == “main”:
print("✅ 3-Conversational_ReAct 正常启动!输入 quit 退出") while True: user_input = input("你: ").strip() if user_input.lower() in ["quit", "exit"]: break try: resp = agent_executor.invoke({"input": user_input}) print(f"助手: {resp['output']}
”)
except Exception as e: print(f"助手: 你好呀小老虎!
”)
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容,请联系我们,一经查实,本站将立刻删除。
如需转载请保留出处:https://51itzy.com/kjqy/255967.html