借助CodeBuddy对MCP从理解到实践

'''
@Time    :   2024/01/24 15:18:43
@Author  :   zoyxiong 
@File    :   rag_tool.py
@Desc    :   调用RAG，获取内部数据库用户问题相关内容
'''
from langchain.tools.base import ToolException
from langchain.agents.tools import Tool
from functools import lru_cache

class RAGTool:
    name = "RAGTool"
    description = """知识库工具：此工具可以从本地知识库中检索用户问题相关内容。"""

    def __init__(self, query):
        self.query = query
    
    @lru_cache(maxsize=128)
    def get_rag_content(self, info):
        # 调用RAG接口，这里需要替换为实际的API调用
        content = your_rag_api(info)
        return content
        
    def get_tool(self):
        return Tool(
            name=self.name,
            func=self._run,
            description=self.description,
        )
    
    def _run(self, query) -> str:
        # 修正参数，应该使用self.query而不是query
        result = self.get_rag_content(query) 
        return result

def test_agent(query):
    rag_tool = RAGTool(query)
    web_tool = WebTool(query)
    tools = [rag_tool.get_tool(), web_tool.get_tool()]
    planner = load_chat_planner(llm, system_prompt=PLAN_PROMPT)
    executor = load_agent_executor(llm, tools, verbose=True)
    agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)
    result = agent.run(query)
    return result

PREFIX = """Respond to the human as helpfully and accurately as possible. You have access to the following tools:"""
FORMAT_INSTRUCTIONS = """Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).

Valid "action" values: "Final Answer" or {tool_names}
Provide only ONE action per $JSON_BLOB, as shown:

```
{{{{
  "action": $TOOL_NAME,
  "action_input": $INPUT
}}}}
```

Follow this format:

Question: input question to answer
Thought: consider previous and subsequent steps
Action:
```
$JSON_BLOB
```
Observation: action result
... (repeat Thought/Action/Observation N times)
Thought: I know what to respond
Action:
```
{{{{
  "action": "Final Answer",
  "action_input": "Final response to human"
}}}}
```"""
SUFFIX = """Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation:.
Thought:"""

# 定义函数的结构和参数
functions = [
    {
        "name": "get_current_weather",
        "description": "获取指定位置的当前天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "城市名称，如'北京'、'上海'等"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位"
                }
            },
            "required": ["location"]
        }
    }
]
def chat_with_function_call(user_message: str) -> None:
    """
    Chat with ChatGPT and handle potential function calls.
    
    Args:
        user_message (str): The user's input message to send to ChatGPT.
    
    Returns:
        None: Prints ChatGPT's response or function call results directly.
    
    Process:
        1. Sends user message to ChatGPT API with function definitions
        2. Checks if response requires function call
        3. If function call needed:
            - Executes the requested function
            - Sends function result back to ChatGPT
            - Prints final response
        4. If no function call needed, prints direct response
        5. Handles and reports any errors during the process
    
    Note:
        Requires properly configured 'client' and 'functions' to be available.
        Uses 'gpt-3.5-turbo-1106' model which supports function calling.
    """
    """
    与ChatGPT进行对话，并处理可能的函数调用
    """
    # 创建对话消息
    messages = [{"role": "user", "content": user_message}]
    
    try:
        # 调用ChatGPT API
        response = client.chat.completions.create(
            model="gpt-3.5-turbo-1106",  # 使用支持函数调用的模型
            messages=messages,
            functions=functions,
            function_call="auto"
        )
        
        # 获取助手的回应
        assistant_message = response.choices[0].message
        
        # 检查是否需要调用函数
        if assistant_message.function_call:
            # 获取函数调用信息
            function_name = assistant_message.function_call.name
            function_args = eval(assistant_message.function_call.arguments)
            
            # 调用相应的函数
            if function_name == "get_current_weather":
                function_response = get_current_weather(**function_args)
                
                # 将函数调用结果添加到消息历史
                messages.append({
                    "role": "function",
                    "name": function_name,
                    "content": str(function_response)
                })
                
                # 再次调用API让模型处理函数返回的结果
                second_response = client.chat.completions.create(
                    model="gpt-3.5-turbo-1106",
                    messages=messages
                )
                
                # 打印最终响应
                print("ChatGPT的回答:", second_response.choices[0].message.content)
        else:
            # 如果不需要调用函数，直接打印回答
            print("ChatGPT的回答:", assistant_message.content)
            
    except Exception as e:
        print(f"发生错误: {str(e)}")

{
  "mcpServers": {
    "xdt-charge-station": {     
       "command":"npx",     
         "args": [       
           "-y",        
           "@longshine-ai/xdt"      
          ]
    }
  }
}