基于deepseek的日程助手
·
没时间了,上代码
from a import chat,chat2,chat3
import time
#print(i,end="",flush=True)
def extract_content(text, tag):
import re
pattern = re.compile(rf'<{tag}>(.*?)</{tag}>', re.DOTALL)
match = pattern.search(text)
if match:
return match.group(1).strip()
memory=open("memory.txt","r",encoding="utf-8").read()
make_prompt = lambda memory,word: f"""
你是一个助手,每次对话时,你会知道当前基本信息和记忆。
你需要输出建议(使用<output>和</output>包裹)和修改后的记忆(使用<memory>和</memory>包裹)。
记忆如下:
<memory>{memory}</memory>
输入如下:
“{word}”
时间 {time.strftime(r"%Y-%m-%d %H:%M:%S", time.localtime())}
"""
while True:
try:
prompt = make_prompt(memory,input("输入:"))
except KeyboardInterrupt:
break
words=""
import tqdm
while 1:
for i in tqdm.tqdm(chat3(prompt=prompt)):
#print(i,end="",flush=True)
words+=i
output = extract_content(words, "output")
memory_unsafe = extract_content(words, "memory")
if words.count("<output>")>=1 and memory_unsafe and len(memory_unsafe)>30:
break
else:
print("重新尝试")
with open("error.txt","a",encoding="utf-8") as f:
f.write(words)
with open("memory.txt","w",encoding="utf-8") as f:
if memory_unsafe and len(memory_unsafe)>30:
memory=memory_unsafe
f.write(memory_unsafe)
print(output)
import requests
ip="71.211.164.135"
# 生成文本
def chat(prompt):
response = requests.post(
f"http://{ip}:11434/api/generate",
json={
"model": "deepseek-r1:latest",
"prompt": prompt,
"stream": True # 启用流式处理
},
stream=True # 启用流式处理
)
# 逐行读取响应内容
for line in response.iter_lines():
if line:
import json
#print(type(line.decode('utf-8')))
yield json.loads(line.decode('utf-8'))["response"]
def chat2(prompt):
ip="127.0.0.1"
response = requests.post(
f"http://{ip}:11434/api/generate",
json={
"model": "qwen2:0.5b",
"prompt": prompt,
"stream": True # 启用流式处理
},
stream=True # 启用流式处理
)
# 逐行读取响应内容
for line in response.iter_lines():
if line:
import json
#print(type(line.decode('utf-8')))
yield json.loads(line.decode('utf-8'))["response"]
def chat3(prompt):
#连接阿里云百炼智能
from openai import OpenAI
import os
# 初始化OpenAI客户端
client = OpenAI(
# 如果没有配置环境变量,请用百炼API Key替换:api_key="sk-xxx"
api_key = os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")
reasoning_content = "" # 定义完整思考过程
answer_content = "" # 定义完整回复
is_answering = False # 判断是否结束思考过程并开始回复
# 创建聊天完成请求
completion = client.chat.completions.create(
model="deepseek-r1", # 此处以 deepseek-r1 为例,可按需更换模型名称
messages=[
{"role": "user", "content": prompt}
],
stream=True,
# 解除以下注释会在最后一个chunk返回Token使用量
# stream_options={
# "include_usage": True
# }
)
#print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")
for chunk in completion:
# 如果chunk.choices为空,则打印usage
if not chunk.choices:
print("\nUsage:")
print(chunk.usage)
else:
delta = chunk.choices[0].delta
# 打印思考过程
if hasattr(delta, 'reasoning_content') and delta.reasoning_content != None:
#print(delta.reasoning_content, end='', flush=True)
yield "..."# delta.reasoning_content
reasoning_content += delta.reasoning_content
else:
# 开始回复
if delta.content != "" and is_answering == False:
#print("\n" + "=" * 20 + "完整回复" + "=" * 20 + "\n")
is_answering = True
# 打印回复过程
yield delta.content
answer_content += delta.content
# print("=" * 20 + "完整思考过程" + "=" * 20 + "\n")
# print(reasoning_content)
# print("=" * 20 + "完整回复" + "=" * 20 + "\n")
# print(answer_content)
#chat3("你好")
更多推荐


所有评论(0)