self-constructing_graph/functions/call.py

# coding=utf-8
import requests
import json
import os
from dotenv import load_dotenv
from functions.basic_function import get_document_by_keyword,get_chunk_by_keyword,get_weather_by_city
# 加载环境变量
load_dotenv()
print(os.getenv("DEEPSEEK_API_KEY"))

DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
DEEPSEEK_API_URL = os.getenv("DEEPSEEK_API_URL")
def generate_response_with_function_call(functions, user_input):
    print(">>> generate_response_with_function_call")
    messages = []
    messages.append({"role": "system", "content": '''                     
                     你需要理解用户的意图，选择合适的功能，并给出参数。
                     如果用户的描述不明确，请要求用户提供必要信息'''})
    
    for text in user_input:
        messages.append({"role": "user", "content": text})
        
    headers = {
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}",
        "Content-Type": "application/json; charset=utf-8"
    }
    
    data = {
        "model": "Pro/deepseek-ai/DeepSeek-V3", #deepseek-ai/DeepSeek-V3",
        "messages": messages,
        "temperature": 0.7,
        "max_tokens": 2000,
        "tools":functions,
        "tool_choice": "auto",
        "stream": False
    }
    print(data)
    response = requests.post(DEEPSEEK_API_URL, json=data, headers=headers)
    response.raise_for_status()    
    response = response.json()
    print(">"*30)
    del headers
    del data
    return response
    
'''
{'id': '01951cb08af7038b211056325775cf0c', 
'object': 'chat.completion', 
'created': 1739943086, 
'model': 'Pro/deepseek-ai/DeepSeek-V3', 
'choices': [
    {'index': 0, 
    'message': {
        'role': 'assistant', 
        'content': '', 
        'tool_calls': [
            {'id': '01951cb097e5fe2f49765ac621ad6758', 
            'type': 'function', 
            'function': {'name': 'get_chunk_by_keyword', 
            'arguments': '{"keywords":"银行 销售 保险产品"}'}}]}, 
            'finish_reason': 'tool_calls'}], 
            'usage': {'prompt_tokens': 252, 'completion_tokens': 40, 'total_tokens': 292}, 
            'system_fingerprint': ''}
'''
def parse_function_call(model_response, messages):
    # 处理函数调用结果，根据模型返回参数，调用对应的函数。
    # 调用函数返回结果后构造tool message，再次调用模型，将函数结果输入模型
    # 模型会将函数调用结果以自然语言格式返回给用户。
    if 'tool_calls' in model_response['choices'][0]['message'].keys():
        tool_call = model_response['choices'][0]['message']['tool_calls'][0]
        args = tool_call['function']['arguments']
        function_result = {}
        function_name = tool_call['function']['name']
        print(f">>> call {function_name} with args: {args}")
        if function_name == "get_document_by_keyword":
            args_json = json.loads(args)
            function_result = get_document_by_keyword(args_json['keywords'])
        if function_name == "get_chunk_by_keyword":
            args_json = json.loads(args)
            function_result = get_chunk_by_keyword(args_json['keywords'])
        if function_name == "get_weather_by_city":
            args_json = json.loads(args)
            function_result = get_weather_by_city(args_json['keywords'])
        # messages.append({
        #     "role": "tool",
        #     "content": f"{json.dumps(function_result)}",
        #     "tool_call_id":tool_call['id']
        # })
        return {"result": function_result}
    
    return {"result": ""}
        #print(response.choices[0].message)
        #messages.append(response.choices[0].message.model_dump())