python_api.py · xpeter80/aicheckdoc

【升级通知】Gitee 将于 7 月 11 日晚 23 点进行全站升级维护

代码拉取完成,页面将自动刷新

richgiteeai

Watch

不关注关注所有动态仅关注版本发行动态关注但不提醒动态

1 Star 0 Fork 0

xpeter80/aicheckdoc

代码 Issues 0 Pull Requests 0 Wiki 统计流水线

服务

加入 Gitee

与超过 1400万开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)

免费加入

已有帐号? 立即登录

文件

master

分支 (1)

管理

master

克隆/下载

HTTPS SSH SVN SVN+SSH 下载ZIP

提示

下载代码请复制以下命令到终端执行

为确保你提交的代码身份被 Gitee 正确识别,请执行以下命令完成配置

git config --global user.name userName 
git config --global user.email userEmail

初次使用 SSH 协议进行代码克隆、推送等操作时,需按下述提示完成 SSH 配置

1 生成 RSA 密钥

2 获取 RSA 公钥内容,并配置到 SSH公钥中

在 Gitee 上使用 SVN,请访问使用指南

使用 HTTPS 协议时,命令行会出现如下账号密码验证步骤。基于安全考虑,Gitee 建议配置并使用私人令牌替代登录密码进行克隆、推送等操作

Username for 'https://gitee.com': userName

Password for 'https://userName@gitee.com': # 私人令牌

分支 1

标签 0

aicheckdoc

python_api.py

python_api.py 4.97 KB

import sys
import json
import logging
from dotenv import load_dotenv
from openai import OpenAI
import os

# u914du7f6eu65e5u5fd7
logging.basicConfig(
 level=logging.INFO,
 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
 handlers=[
 logging.FileHandler("python_api.log", encoding="utf-8"),
 logging.StreamHandler()
 ]
)
logger = logging.getLogger(__name__)

# u52a0u8f7du73afu5883u53d8u91cf
load_dotenv()

# u83b7u53d6APIu5bc6u94a5
DEEPSEEK_API_KEY = os.getenv('DEEPSEEK_API_KEY')
DEEPSEEK_API_URL = os.getenv('DEEPSEEK_API_URL', 'https://api.deepseek.com/v1')

def call_reasoner_api(input_data):
 """u8c03u7528DeepSeek Reasoner API
 
 Args:
 input_data (dict): u5305u542bu6a21u578bu3001u6d88u606fu548cu5176u4ed6u53c2u6570u7684u5b57u5178
 """
 try:
 model = input_data.get('model', 'deepseek')
 messages = input_data.get('messages', [])
 temperature = input_data.get('temperature', 0.7)
 max_tokens = input_data.get('max_tokens', 2048)
 
 logger.info(f"u5f00u59cbu4f7fu7528{model}u6a21u578bu8c03u7528API")
 
 # u68c0u67e5APIu5bc6u94a5
 if not DEEPSEEK_API_KEY or DEEPSEEK_API_KEY == 'your_deepseek_key_here':
 logger.warning("u672au8bbeu7f6eDeepSeek APIu5bc6u94a5uff0cu4f7fu7528u6a21u62dfu54cdu5e94")
 return simulate_response(messages)
 
 # u521bu5efaOpenAIu5ba2u6237u7aef
 client = OpenAI(
 api_key=DEEPSEEK_API_KEY,
 base_url=DEEPSEEK_API_URL
 )
 
 # u8c03u7528API
 response = client.chat.completions.create(
 model="deepseek-reasoner",
 messages=messages,
 stream=True,
 max_tokens=max_tokens
 )
 
 # u6d41u5f0fu8f93u51fau7ed3u679c
 for chunk in response:
 # u68c0u67e5u662fu5426u6709u601du7ef4u94feu5185u5bb9
 if hasattr(chunk.choices[0].delta, 'reasoning_content') and chunk.choices[0].delta.reasoning_content:
 content = chunk.choices[0].delta.reasoning_content
 sys.stdout.write(content)
 sys.stdout.flush()
 # u68c0u67e5u662fu5426u6709u6700u7ec8u56deu7b54
 elif hasattr(chunk.choices[0].delta, 'content') and chunk.choices[0].delta.content:
 content = chunk.choices[0].delta.content
 sys.stdout.write(content)
 sys.stdout.flush()
 
 logger.info("u6210u529fu5b8cu6210APIu8c03u7528")
 except Exception as e:
 logger.error(f"u8c03u7528APIu65f6u51fau9519: {str(e)}")
 # u51fau9519u65f6u8fd4u56deu6a21u62dfu54cdu5e94
 return simulate_response(messages)

def simulate_response(messages):
 """u751fu6210u6a21u62dfu7684u54cdu5e94"""
 user_message = messages[-1]['content'] if messages and messages[-1]['role'] == 'user' else ''
 
 # u6a21u62dfu601du7ef4u94feu8fc7u7a0b
 thinking_parts = [
 f"u5bf9u4e8eu95eeu9898 '{user_message}'\n",
 "u6211u9700u8981u8fdbu884cu601du8003\n\n",
 "u9996u5148uff0cu6211u8981u7406u89e3u95eeu9898u7684u5185u5bb9u548cu8981u6c42\n",
 "u8fd9u4e2au95eeu9898u7684u5173u952eu662f...\n",
 "u6839u636eu6211u7684u77e5u8bc6uff0cu6211u8ba4u4e3a...\n",
 "u7ecfu8fc7u5206u6790uff0cu6211u5f97u51fau4ee5u4e0bu7ed3u8bba\n"
 ]
 
 # u6a21u62dfu6700u7ec8u56deu7b54
 response_parts = [
 "u60a8u597duff01",
 f"u5173u4e8eu60a8u63d0u51fau7684'{user_message}'u7684u95eeu9898uff0c",
 "u6211u7684u56deu7b54u662fuff1a\n",
 "u2026u2026u8fd9u91ccu662fu8be6u7ec6u89e3u91cau2026u2026\n",
 "u5e0cu671bu6211u7684u56deu7b54u5bf9u60a8u6709u6240u5e2eu52a9uff01u5982u679cu8fd8u6709u5176u4ed6u95eeu9898uff0cu8bf7u968fu65f6u63d0u95eeu3002"
 ]
 
 # u6a21u62dfu601du7ef4u94feu8f93u51fa
 for part in thinking_parts:
 sys.stdout.write(part)
 sys.stdout.flush()
 # u6a21u62dfu4e00u4e9bu5ef6u8fdf
 # time.sleep(0.5)
 
 # u6a21u62dfu6700u7ec8u56deu7b54u8f93u51fa
 for part in response_parts:
 sys.stdout.write(part)
 sys.stdout.flush()
 # u6a21u62dfu4e00u4e9bu5ef6u8fdf
 # time.sleep(0.5)

def main():
 """u4e3bu51fdu6570uff0cu5904u7406u547du4ee4u884cu53c2u6570u5e76u8c03u7528API"""
 try:
 if len(sys.argv) < 2:
 logger.error("u7f3au5c11u8f93u5165u6587u4ef6u8defu5f84u53c2u6570")
 sys.exit(1)
 
 input_file = sys.argv[1]
 
 # u68c0u67e5u6587u4ef6u662fu5426u5b58u5728
 if not os.path.exists(input_file):
 logger.error(f"u6307u5b9au7684u6587u4ef6u4e0du5b58u5728: {input_file}")
 sys.exit(1)
 
 # u8bfbu53d6u548cu89e3u6790u5165u53c3u6570
 with open(input_file, 'r', encoding='utf-8') as f:
 input_data = json.load(f)
 
 # u8c03u7528API
 call_reasoner_api(input_data)
 except Exception as e:
 logger.error(f"u6267u884cu811au672cu65f6u51fau9519: {str(e)}")
 sys.exit(1)

if __name__ == "__main__":
 main()

一键复制编辑原始数据按行查看历史

Administrator 提交于 2025年03月10日 18:13 +08:00 . 初始提交:添加项目文件

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140

import sys
import json
import logging
from dotenv import load_dotenv
from openai import OpenAI
import os

# u914du7f6eu65e5u5fd7
logging.basicConfig(
 level=logging.INFO,
 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
 handlers=[
 logging.FileHandler("python_api.log", encoding="utf-8"),
 logging.StreamHandler()
 ]
)
logger = logging.getLogger(__name__)

# u52a0u8f7du73afu5883u53d8u91cf
load_dotenv()

# u83b7u53d6APIu5bc6u94a5
DEEPSEEK_API_KEY = os.getenv('DEEPSEEK_API_KEY')
DEEPSEEK_API_URL = os.getenv('DEEPSEEK_API_URL', 'https://api.deepseek.com/v1')

def call_reasoner_api(input_data):
 """u8c03u7528DeepSeek Reasoner API

 Args:
 input_data (dict): u5305u542bu6a21u578bu3001u6d88u606fu548cu5176u4ed6u53c2u6570u7684u5b57u5178
 """
 try:
 model = input_data.get('model', 'deepseek')
 messages = input_data.get('messages', [])
 temperature = input_data.get('temperature', 0.7)
 max_tokens = input_data.get('max_tokens', 2048)
 
 logger.info(f"u5f00u59cbu4f7fu7528{model}u6a21u578bu8c03u7528API")
 
 # u68c0u67e5APIu5bc6u94a5
 if not DEEPSEEK_API_KEY or DEEPSEEK_API_KEY == 'your_deepseek_key_here':
 logger.warning("u672au8bbeu7f6eDeepSeek APIu5bc6u94a5uff0cu4f7fu7528u6a21u62dfu54cdu5e94")
 return simulate_response(messages)
 
 # u521bu5efaOpenAIu5ba2u6237u7aef
 client = OpenAI(
 api_key=DEEPSEEK_API_KEY,
 base_url=DEEPSEEK_API_URL
 )
 
 # u8c03u7528API
 response = client.chat.completions.create(
 model="deepseek-reasoner",
 messages=messages,
 stream=True,
 max_tokens=max_tokens
 )
 
 # u6d41u5f0fu8f93u51fau7ed3u679c
 for chunk in response:
 # u68c0u67e5u662fu5426u6709u601du7ef4u94feu5185u5bb9
 if hasattr(chunk.choices[0].delta, 'reasoning_content') and chunk.choices[0].delta.reasoning_content:
 content = chunk.choices[0].delta.reasoning_content
 sys.stdout.write(content)
 sys.stdout.flush()
 # u68c0u67e5u662fu5426u6709u6700u7ec8u56deu7b54
 elif hasattr(chunk.choices[0].delta, 'content') and chunk.choices[0].delta.content:
 content = chunk.choices[0].delta.content
 sys.stdout.write(content)
 sys.stdout.flush()
 
 logger.info("u6210u529fu5b8cu6210APIu8c03u7528")
 except Exception as e:
 logger.error(f"u8c03u7528APIu65f6u51fau9519: {str(e)}")
 # u51fau9519u65f6u8fd4u56deu6a21u62dfu54cdu5e94
 return simulate_response(messages)

def simulate_response(messages):
 """u751fu6210u6a21u62dfu7684u54cdu5e94"""
 user_message = messages[-1]['content'] if messages and messages[-1]['role'] == 'user' else ''
 
 # u6a21u62dfu601du7ef4u94feu8fc7u7a0b
 thinking_parts = [
 f"u5bf9u4e8eu95eeu9898 '{user_message}'\n",
 "u6211u9700u8981u8fdbu884cu601du8003\n\n",
 "u9996u5148uff0cu6211u8981u7406u89e3u95eeu9898u7684u5185u5bb9u548cu8981u6c42\n",
 "u8fd9u4e2au95eeu9898u7684u5173u952eu662f...\n",
 "u6839u636eu6211u7684u77e5u8bc6uff0cu6211u8ba4u4e3a...\n",
 "u7ecfu8fc7u5206u6790uff0cu6211u5f97u51fau4ee5u4e0bu7ed3u8bba\n"
 ]
 
 # u6a21u62dfu6700u7ec8u56deu7b54
 response_parts = [
 "u60a8u597duff01",
 f"u5173u4e8eu60a8u63d0u51fau7684'{user_message}'u7684u95eeu9898uff0c",
 "u6211u7684u56deu7b54u662fuff1a\n",
 "u2026u2026u8fd9u91ccu662fu8be6u7ec6u89e3u91cau2026u2026\n",
 "u5e0cu671bu6211u7684u56deu7b54u5bf9u60a8u6709u6240u5e2eu52a9uff01u5982u679cu8fd8u6709u5176u4ed6u95eeu9898uff0cu8bf7u968fu65f6u63d0u95eeu3002"
 ]
 
 # u6a21u62dfu601du7ef4u94feu8f93u51fa
 for part in thinking_parts:
 sys.stdout.write(part)
 sys.stdout.flush()
 # u6a21u62dfu4e00u4e9bu5ef6u8fdf
 # time.sleep(0.5)
 
 # u6a21u62dfu6700u7ec8u56deu7b54u8f93u51fa
 for part in response_parts:
 sys.stdout.write(part)
 sys.stdout.flush()
 # u6a21u62dfu4e00u4e9bu5ef6u8fdf
 # time.sleep(0.5)

def main():
 """u4e3bu51fdu6570uff0cu5904u7406u547du4ee4u884cu53c2u6570u5e76u8c03u7528API"""
 try:
 if len(sys.argv) < 2:
 logger.error("u7f3au5c11u8f93u5165u6587u4ef6u8defu5f84u53c2u6570")
 sys.exit(1)
 
 input_file = sys.argv[1]
 
 # u68c0u67e5u6587u4ef6u662fu5426u5b58u5728
 if not os.path.exists(input_file):
 logger.error(f"u6307u5b9au7684u6587u4ef6u4e0du5b58u5728: {input_file}")
 sys.exit(1)
 
 # u8bfbu53d6u548cu89e3u6790u5165u53c3u6570
 with open(input_file, 'r', encoding='utf-8') as f:
 input_data = json.load(f)
 
 # u8c03u7528API
 call_reasoner_api(input_data)
 except Exception as e:
 logger.error(f"u6267u884cu811au672cu65f6u51fau9519: {str(e)}")
 sys.exit(1)

if __name__ == "__main__":
 main()