代码

可直接调用

from typing import Dict, Any
import requests
from langchain_ollama import ChatOllama
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import re
import json


class DeepSeekAgent:
    def __init__(self):
        self.llm = ChatOllama(
            model="deepseek-r1:1.5b",
            temperature=0.7,
            base_url="http://localhost:11434"
        )

        self.tools = {
            "get_weather": self.get_weather,
            "send_email": self.send_email,
        }
        # 修改模板，明确使用 {input} 变量
        self.prompt = ChatPromptTemplate.from_template("""
                你是一个人工智能助手，能够完全准确地根据根据用户需要的格式进行回答。

                用户输入: {input}

                """)

        self.chain = self.prompt | self.llm | StrOutputParser()

    def call_deepseek(self, prompt: str) -> str:
        # """调用DeepSeek API获取响应"""
        # headers = {"Authorization": f"Bearer {self.api_key}"}
        # data = {
        #     "model": "deepseek-chat",
        #     "messages": [{"role": "user", "content": prompt}]
        # }
        # response = requests.post(self.llm_api, json=data, headers=headers)
        # return response.json()["choices"][0]["message"]["content"]
        return self.chain.invoke({"input": prompt})

    def get_weather(self, location: str) -> str:
        """模拟天气查询工具"""
        # 实际可调用天气API（如OpenWeatherMap）
        return f"Weather in {location}: Sunny, 25°C"

    def send_email(self, recipient: str, subject: str) -> str:
        """模拟发邮件工具"""
        # 实际可调用SMTP或邮件API
        return f"Email sent to {recipient} with subject: '{subject}'"

    # def parse_instruction(self, user_input: str) -> Dict[str, Any]:
    #     """解析用户指令，决定调用哪个工具"""
    #     prompt = f"""
    #     用户指令: {user_input}
    #     请判断是否需要调用工具，并返回JSON格式，例如：
    #     {{"tool": "get_weather", "args": {{"location": "Beijing"}}}}
    #     如果无需工具，返回{{"tool": null}}。
    #     """
    #     llm_response = self.call_deepseek(prompt)
    #     return eval(llm_response)  # 实际应使用安全解析（如json.loads）

    def parse_instruction(self, user_input):
        # 保持原来的提示词模板不变
        prompt = f"""
        用户指令: {user_input}
        请严格遵循以下格式：
        请判断是否需要调用工具（你有如下工具可以使用：当用户询问天气如何，或者你需要知道天气如何的时候，查询天气工具（get_weather）），并返回JSON格式
        只返回一个JSON对象，不加任何额外解释。

        例如：
        {{"tool": "get_weather", "args": {{"location": "Beijing"}}}}

        如果无需工具，返回：
        {{"tool": null}}
        """

        # 获取大语言模型的响应
        llm_response = self.call_deepseek(prompt)

        # 添加调试信息：打印完整响应
        print(f"大模型完整响应: {llm_response}")
        """直接从含<think>的响应中提取JSON对象"""
        # 先移除<think>标签
        no_think = re.sub(r'<think>[\s\S]*?</think>', '', llm_response, flags=re.DOTALL)
        print(f"去除think标签响应: {no_think}")
        # 方法1：严格提取JSON部分
        # 查找第一个{和最后一个}的位置
        start = no_think.find('{')
        end = no_think.rfind('}') + 1
        if start == -1 or end == 0:
            raise ValueError("响应中未找到有效的JSON结构")

        json_match = no_think[start:end]
        print(f"大括号标签响应: {json_match}")

        # 验证大括号是否匹配
        if json_match.count('{') != json_match.count('}'):
            raise ValueError("大括号不匹配，JSON结构不完整")

        # 最终清理
        json_match = json_match.strip()
        print("最终提取的JSON:", repr(json_match))


        if not json_match:
            raise ValueError("无法从响应中提取JSON内容")

        # 关键修复：彻底清理字符串
        json_match = json_match.strip()  # 去除首尾空白
        json_match = re.sub(r'[\u200b-\u200f\u202a-\u202e]', '', json_match)  # 去除零宽字符
        json_match = json_match.encode('ascii', 'ignore').decode('ascii')  # 去除非ASCII字符

        # 调试：打印清理后的JSON字符串
        print("清理后的JSON字符串:", repr(json_match))
        try:
            result = json.loads(json_match)
            print("解析成功:", result)
            return result
        except json.JSONDecodeError as e:
            # 详细错误诊断
            print(f"JSON解析失败，错误位置：{e.lineno}行{e.colno}列")
            print("问题字符的ASCII码:", ord(json_match[e.pos]))
            raise ValueError(f"JSON解析失败: {str(e)}\n问题内容: {json_match[e.pos - 10:e.pos + 10]}")



        # # 再提取JSON部分
        # json_match = re.search(r'\{[\s\S]*?\}', no_think)
        # if not json_match:
        #     raise ValueError("未找到有效的JSON内容")
        #
        # print(f"调用工具响应: {json.loads(json_match.group())}")
        # return json.loads(json_match.group())
        # # 使用正则表达式提取JSON部分
        # json_match = re.search(r'\{.*\}', llm_response, re.DOTALL)
        #
        # if not json_match:
        #     raise ValueError(f"响应中未找到有效的JSON: {llm_response}")
        #
        # json_str = json_match.group()
        #
        # # 尝试处理可能存在的内部转义
        # json_str = re.sub(r'\\{', '{', json_str)
        # json_str = re.sub(r'\\}', '}', json_str)
        #
        # try:
        #     # 使用安全的json解析
        #     result = json.loads(json_str)
        #     print(f"成功解析JSON: {result}")  # 调试用
        #     return result
        # except json.JSONDecodeError as e:
        #     print(f"JSON解析错误: {e.msg}")
        #     print(f"尝试解析的内容: {json_str}")
        #     raise

        # # 定位并提取Action行
        # action_line = None
        # for line in llm_response.split('\n'):
        #     if line.startswith('Action:'):
        #         # 移除"Action:"前缀并清理空白
        #         action_line = line.replace('Action:', '').strip()
        #         break
        #
        # # 添加错误检查
        # if not action_line:
        #     raise ValueError("响应中未找到有效的Action行")
        #
        # # 只对函数调用部分执行eval
        # return eval(action_line)  # 现在只处理如"get_weather(...)"这样的内容

    def run(self, user_input: str) -> str:
        """Agent 主逻辑"""
        # 1. 解析指令
        action = self.parse_instruction(user_input)

        # 2. 执行工具或直接响应

        if action["tool"] in self.tools:
            print(f"调用工具：", {action["tool"]})
            tool_func = self.tools[action["tool"]]
            return tool_func(**action["args"])
        else:
            return self.call_deepseek(user_input)  # 直接LLM响应