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