技术栈

完整代码

# app.py import os import json from flask import Flask, request, jsonify, render_template from openai import OpenAI app = Flask(__name__) client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") # 加载知识库 with open("knowledge_base.json", "r", encoding="utf-8") as f: KNOWLEDGE_BASE = json.load(f) # 会话存储 (生产环境建议用 Redis) sessions = {} SYSTEM_PROMPT = """你是一个专业的客服助手。请根据以下知识库内容回答用户问题。 规则： 1. 如果知识库中有相关信息，请准确引用 2. 如果知识库中没有，请礼貌告知用户需要转接人工 3. 保持友好、专业、简洁的回答风格 知识库内容： {knowledge} 当前对话历史： {history}""" def build_prompt(session_id, user_message): session = sessions.get(session_id, {"history": []}) history_text = "\n".join([f"{'用户' if h['role']=='user' else '客服'}: {h['content']}" for h in session["history"][-6:]]) knowledge_text = json.dumps(KNOWLEDGE_BASE, ensure_ascii=False, indent=2) return SYSTEM_PROMPT.format(knowledge=knowledge_text, history=history_text) @app.route("/") def index(): return render_template("index.html") @app.route("/chat", methods=["POST"]) def chat(): data = request.json session_id = data.get("session_id", "default") user_message = data.get("message", "") if not user_message: return jsonify({"error": "消息不能为空"}), 400 if session_id not in sessions: sessions[session_id] = {"history": []} sessions[session_id]["history"].append({"role": "user", "content": user_message}) prompt = build_prompt(session_id, user_message) response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": prompt}, {"role": "user", "content": user_message} ], temperature=0.3, max_tokens=1024 ) reply = response.choices[0].message.content sessions[session_id]["history"].append({"role": "assistant", "content": reply}) return jsonify({"reply": reply, "session_id": session_id}) if __name__ == "__main__": app.run(debug=True, port=5000) # knowledge_base.json # { # "faq": [ # {"q": "如何退货？", "a": "在订单页面点击申请退货..."}, # {"q": "发货时间是多久？", "a": "下单后48小时内发货..."}, # {"q": "支持哪些支付方式？", "a": "支持微信、支付宝、银行卡..."} # ], # "policies": { # "return": "7天无理由退货，商品需保持原包装...", # "warranty": "所有商品享受1年质保服务..." # } # }

运行方式

# 安装依赖 pip install flask openai # 设置环境变量 export DEEPSEEK_API_KEY="sk-your-api-key" # 创建知识库文件 echo '{"faq":[{"q":"如何退货？","a":"在订单页面点击申请退货..."}]}' > knowledge_base.json # 运行 python app.py # 访问 http://localhost:5000

预期效果

• 用户通过网页聊天界面提问，机器人基于知识库实时回答
• 支持多轮对话，记住上下文（最近 6 轮）
• 知识库外的内容会引导用户转接人工客服
• 响应延迟 < 2 秒（取决于 API 网络延迟）

技术栈

完整代码

#!/usr/bin/env python3 # code_review.py import os import sys import subprocess import argparse from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") REVIEW_PROMPT = """你是一位资深代码审查专家。请对以下 git diff 进行深度审查，输出结构化报告。 审查维度： 1. 安全漏洞：SQL注入、XSS、敏感信息泄露、权限问题 2. 代码质量：命名规范、函数长度、复杂度、重复代码 3. 性能问题：N+1查询、不必要的循环、内存泄漏风险 4. 最佳实践：错误处理、日志记录、类型安全、测试覆盖 输出格式（Markdown）： ## 代码审查报告 ### 安全漏洞 - [严重程度] 问题描述 + 建议修复方案 ### 代码质量问题 - [严重程度] 问题描述 + 建议重构方案 ### 性能问题 - [严重程度] 问题描述 + 优化建议 ### 最佳实践建议 - [优先级] 建议内容 ### 总体评分 - 安全性: X/10 - 代码质量: X/10 - 性能: X/10 - 综合评分: X/10 Git Diff: {diff}""" def get_git_diff(staged_only=False): cmd = ["git", "diff", "--cached"] if staged_only else ["git", "diff"] try: result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print(f"错误: git diff 执行失败: {result.stderr}") sys.exit(1) return result.stdout.strip() except FileNotFoundError: print("错误: 未找到 git 命令，请确认在 git 仓库中执行") sys.exit(1) def review_code(diff_text, use_reasoner=False): model = "deepseek-reasoner" if use_reasoner else "deepseek-chat" prompt = REVIEW_PROMPT.format(diff=diff_text[:30000]) response = client.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=4096 ) return response.choices[0].message.content def main(): parser = argparse.ArgumentParser(description="DeepSeek 代码审查助手") parser.add_argument("--staged", action="store_true", help="仅审查已暂存的更改") parser.add_argument("--reasoner", action="store_true", help="使用 R1 推理模型（更深入但更慢）") parser.add_argument("--output", "-o", help="输出文件路径") parser.add_argument("--json", action="store_true", help="JSON 格式输出") args = parser.parse_args() diff = get_git_diff(args.staged) if not diff: print("没有需要审查的代码变更") return print(f"正在审查 {len(diff)} 字符的代码变更...") report = review_code(diff, args.reasoner) if args.output: with open(args.output, "w", encoding="utf-8") as f: f.write(report) print(f"审查报告已保存至: {args.output}") else: print("\n" + "=" * 60) print(report) print("=" * 60) if __name__ == "__main__": main()

运行方式

# 安装依赖 pip install openai # 在 git 仓库中运行 python code_review.py # 审查已暂存的更改 python code_review.py --staged # 使用 R1 推理模型进行深度审查 python code_review.py --reasoner # 保存报告到文件 python code_review.py -o review_report.md

预期效果

• 输出结构化的 Markdown 审查报告，包含安全、质量、性能、最佳实践四个维度
• R1 推理模式可发现深层次逻辑问题和潜在 bug
• 支持 JSON 格式输出，方便集成 CI/CD 流水线
• 每个问题附带严重程度评级和具体修复建议

技术栈

完整代码

# rag_qa.py import os import streamlit as st import fitz # PyMuPDF import chromadb from chromadb.utils import embedding_functions from openai import OpenAI from langchain.text_splitter import RecursiveCharacterTextSplitter client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") # 初始化 ChromaDB chroma_client = chromadb.PersistentClient(path="./chroma_db") sentence_transformer_ef = embedding_functions.SentenceTransformerEmbeddingFunction( model_name="BAAI/bge-small-zh-v1.5" ) def get_or_create_collection(name="documents"): try: return chroma_client.get_collection(name, embedding_function=sentence_transformer_ef) except: return chroma_client.create_collection(name, embedding_function=sentence_transformer_ef) def extract_text_from_pdf(pdf_file): doc = fitz.open(stream=pdf_file.read(), filetype="pdf") text = "" for page in doc: text += page.get_text() return text def split_text(text, chunk_size=500, overlap=50): splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=overlap, separators=["\n\n", "\n", "。", "！", "？", "；", ".", " ", ""] ) return splitter.split_text(text) def index_document(text_chunks, doc_name): collection = get_or_create_collection() existing = collection.get()["ids"] if existing: collection.delete(ids=existing) ids = [f"{doc_name}_{i}" for i in range(len(text_chunks))] collection.add(documents=text_chunks, ids=ids) return len(text_chunks) def search_documents(query, top_k=3): collection = get_or_create_collection() results = collection.query(query_texts=[query], n_results=top_k) return results["documents"][0] if results["documents"] else [] def generate_answer(query, context_chunks): context = "\n\n---\n\n".join(context_chunks) prompt = f"""你是一个专业的文档问答助手。请基于以下文档内容回答用户的问题。 规则： 1. 如果文档中有相关信息，请准确引用并回答 2. 如果文档中没有相关信息，请明确告知用户 3. 在回答末尾标注引用来源（如"根据文档第X段"） 文档内容： {context} 用户问题：{query} 请回答：""" response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=2048 ) return response.choices[0].message.content # Streamlit UI st.set_page_config(page_title="DeepSeek 文档问答", layout="wide") st.title("DeepSeek 智能文档问答") tab1, tab2 = st.tabs(["上传文档", "提问"]) with tab1: uploaded_file = st.file_uploader("上传 PDF 文档", type=["pdf"]) if uploaded_file: with st.spinner("正在解析文档..."): text = extract_text_from_pdf(uploaded_file) st.info(f"已提取 {len(text)} 个字符") chunks = split_text(text) count = index_document(chunks, uploaded_file.name) st.success(f"已索引 {count} 个文本块，现在可以提问了！") with tab2: query = st.text_input("输入你的问题") if query: with st.spinner("正在检索和生成回答..."): chunks = search_documents(query) if chunks: answer = generate_answer(query, chunks) st.markdown(answer) with st.expander("查看检索到的文档片段"): for i, chunk in enumerate(chunks): st.markdown(f"**片段 {i+1}:**") st.text(chunk[:300] + "...") else: st.warning("未找到相关文档内容，请先上传文档")

运行方式

# 安装依赖 pip install streamlit PyMuPDF chromadb openai langchain sentence-transformers # 运行 streamlit run rag_qa.py # 访问 http://localhost:8501

预期效果

• 上传 PDF 后自动索引，支持中文文档的语义检索
• 提问时检索最相关的文档片段，基于上下文生成准确回答
• 回答中标注引用来源，可展开查看原始文档片段
• 支持多文档上传，切换文档自动重新索引

技术栈

完整代码

#!/usr/bin/env python3 # translator.py import os import json import argparse from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") LANGUAGE_MAP = { "zh": "中文", "en": "英语", "ja": "日语", "ko": "韩语", "fr": "法语", "de": "德语", "es": "西班牙语", "ru": "俄语", "pt": "葡萄牙语", "ar": "阿拉伯语", "th": "泰语", "vi": "越南语" } TRANSLATE_PROMPT = """你是一个专业翻译引擎。请将以下文本从 {source} 翻译成 {target}。 翻译要求： 1. 保持原文语气和风格 2. 技术术语使用行业标准译法 3. 如有术语表，严格遵守术语表定义 4. 代码、数字、专有名词不翻译 5. 返回纯翻译文本，不要添加解释 {glossary_section} 原文： {text}""" def load_glossary(filepath): if not filepath or not os.path.exists(filepath): return {} with open(filepath, "r", encoding="utf-8") as f: glossary = json.load(f) if filepath.endswith(".json") else {} if isinstance(glossary, list): glossary = {item["source"]: item["target"] for item in glossary} return glossary def build_glossary_section(glossary): if not glossary: return "" items = "\n".join([f"- {k} -> {v}" for k, v in glossary.items()]) return f"术语表（必须严格遵守）：\n{items}" def translate_text(text, source="zh", target="en", glossary=None): source_name = LANGUAGE_MAP.get(source, source) target_name = LANGUAGE_MAP.get(target, target) glossary_section = build_glossary_section(glossary) prompt = TRANSLATE_PROMPT.format( source=source_name, target=target_name, glossary_section=glossary_section, text=text ) response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=4096 ) return response.choices[0].message.content def batch_translate(texts, source="zh", target="en", glossary=None): results = [] for i, text in enumerate(texts): print(f"翻译中... ({i+1}/{len(texts)})") translated = translate_text(text, source, target, glossary) results.append({"source": text, "target": translated}) return results def main(): parser = argparse.ArgumentParser(description="DeepSeek 多语言翻译工具") parser.add_argument("text", nargs="?", help="要翻译的文本") parser.add_argument("--file", "-f", help="从文件读取文本") parser.add_argument("--source", "-s", default="zh", help="源语言代码 (默认: zh)") parser.add_argument("--target", "-t", default="en", help="目标语言代码 (默认: en)") parser.add_argument("--glossary", "-g", help="术语表 JSON 文件路径") parser.add_argument("--output", "-o", help="输出 JSON 文件路径") parser.add_argument("--batch", "-b", help="批量翻译: JSON 文件路径 (每行一条)") args = parser.parse_args() glossary = load_glossary(args.glossary) if args.batch: with open(args.batch, "r", encoding="utf-8") as f: texts = json.load(f) results = batch_translate(texts, args.source, args.target, glossary) output = json.dumps(results, ensure_ascii=False, indent=2) elif args.file: with open(args.file, "r", encoding="utf-8") as f: text = f.read() result = translate_text(text, args.source, args.target, glossary) output = json.dumps({"source": text, "target": result}, ensure_ascii=False, indent=2) elif args.text: result = translate_text(args.text, args.source, args.target, glossary) output = json.dumps({"source": args.text, "target": result}, ensure_ascii=False, indent=2) else: parser.print_help() return if args.output: with open(args.output, "w", encoding="utf-8") as f: f.write(output) print(f"翻译结果已保存至: {args.output}") else: print(output) if __name__ == "__main__": main()

运行方式

# 安装依赖 pip install openai # 单句翻译 python translator.py "人工智能正在改变世界" --target en # 文件翻译 python translator.py -f document.txt -s zh -t ja # 批量翻译 python translator.py -b texts.json -s zh -t en -o results.json # 带术语表翻译 python translator.py -f tech_doc.txt -g glossary.json -o translated.json

预期效果

• 支持 20+ 语言互译，翻译质量接近专业人工水平
• 术语表确保专业术语翻译一致性
• 批量翻译支持 JSON 输入输出，方便集成自动化流程
• 代码、数字、专有名词自动保留不翻译

技术栈

完整代码

# data_analyst.py import os import io import pandas as pd import matplotlib.pyplot as plt import streamlit as st from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") CODE_GEN_PROMPT = """你是一个 Pandas 数据分析专家。根据用户的数据分析需求，生成可执行的 Python 代码。 数据信息： - 列名: {columns} - 数据类型: {dtypes} - 数据行数: {rows} - 前5行预览: {preview} 用户需求：{query} 请生成一段 Python 代码来实现用户的分析需求。代码要求： 1. 使用变量 df 作为输入 DataFrame 2. 分析结果存储在 result_df 变量中（DataFrame 格式） 3. 如果需要图表，将图表对象存储在 fig 变量中 4. 只输出代码，不要有解释文字 5. 代码必须安全，不要执行任何文件操作或网络请求 ```python""" REPORT_PROMPT = """你是一个数据分析师。根据以下分析结果，生成一份简洁的数据分析报告。 分析需求：{query} 分析结果摘要：{summary} 请生成一份结构化的分析报告，包含： 1. 分析概述 2. 关键发现（3-5条） 3. 数据洞察 4. 建议""" plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False st.set_page_config(page_title="DeepSeek 数据分析助手", layout="wide") st.title("DeepSeek 数据分析助手") uploaded_file = st.file_uploader("上传数据文件", type=["csv", "xlsx", "xls"]) if uploaded_file: if uploaded_file.name.endswith(".csv"): df = pd.read_csv(uploaded_file) else: df = pd.read_excel(uploaded_file) st.subheader("数据预览") st.dataframe(df.head(10), use_container_width=True) col1, col2, col3 = st.columns(3) col1.metric("行数", len(df)) col2.metric("列数", len(df.columns)) col3.metric("缺失值", df.isnull().sum().sum()) query = st.text_area("用中文描述你的分析需求", placeholder="例如：按月份统计销售额，画出趋势图；找出销售额最高的前10个产品；分析用户年龄分布等") if st.button("开始分析", type="primary") and query: with st.spinner("DeepSeek 正在生成分析代码..."): preview = df.head(5).to_string() columns = ", ".join(df.columns.tolist()) dtypes = ", ".join([f"{c}: {t}" for c, t in df.dtypes.items()]) prompt = CODE_GEN_PROMPT.format( columns=columns, dtypes=dtypes, rows=len(df), preview=preview, query=query ) response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=2048 ) code = response.choices[0].message.content.strip() code = code.replace("```python", "").replace("```", "").strip() st.code(code, language="python") with st.spinner("正在执行分析..."): try: local_vars = {"df": df.copy(), "pd": pd, "plt": plt, "result_df": None, "fig": None} exec(code, {"__builtins__": __builtins__}, local_vars) result_df = local_vars.get("result_df") fig = local_vars.get("fig") if result_df is not None: st.subheader("分析结果") st.dataframe(result_df, use_container_width=True) csv = result_df.to_csv(index=False).encode("utf-8-sig") st.download_button("下载 CSV", csv, "analysis_result.csv", "text/csv") buffer = io.BytesIO() with pd.ExcelWriter(buffer, engine="openpyxl") as writer: result_df.to_excel(writer, index=False, sheet_name="分析结果") st.download_button("下载 Excel", buffer.getvalue(), "analysis_result.xlsx") if fig is not None: st.subheader("可视化图表") st.pyplot(fig) if result_df is not None: with st.spinner("正在生成分析报告..."): summary = result_df.describe().to_string() report_prompt = REPORT_PROMPT.format(query=query, summary=summary[:3000]) report = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": report_prompt}], temperature=0.3, max_tokens=2048 ) st.subheader("分析报告") st.markdown(report.choices[0].message.content) except Exception as e: st.error(f"代码执行出错: {str(e)}") st.info("请尝试重新描述你的分析需求")

运行方式

# 安装依赖 pip install streamlit pandas openpyxl matplotlib openai # 运行 streamlit run data_analyst.py # 访问 http://localhost:8501

预期效果

• 上传 CSV/Excel 文件后，用中文描述分析需求即可自动生成分析代码
• 自动执行 Pandas 分析，展示结果表格和可视化图表
• 一键导出分析结果为 CSV 或 Excel 格式
• 自动生成结构化的数据分析报告

技术栈

完整代码

#!/usr/bin/env python3 # blog_generator.py import os import json import argparse from datetime import datetime from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") OUTLINE_PROMPT = """你是一位资深技术博客作者。请为以下主题生成一个详细的文章大纲。 主题：{topic} 写作风格：{style} 目标读者：{audience} 请生成一份包含 5-8 个章节的详细大纲，每个章节包含 2-3 个子要点。""" ARTICLE_PROMPT = """你是一位资深技术博客作者。请根据大纲撰写一篇完整的 Markdown 格式博客文章。 主题：{topic} 大纲： {outline} 写作风格：{style} 目标读者：{audience} 要求： 1. 使用 Markdown 格式，包含适当的标题层级 2. 包含代码示例（如适用） 3. 内容详实，每个章节至少 150 字 4. 开头要有吸引人的引言 5. 结尾要有总结和展望 6. 不要使用 HTML 标签，纯 Markdown 格式""" SERIES_PROMPT = """你是一位技术博客策划。请为以下主题生成一个系列文章规划。 主题：{topic} 系列文章数量：{count} 请生成一个 JSON 数组，每个元素包含： - title: 文章标题 - slug: URL 友好的标题 - description: 简短描述（20字以内） - order: 序号 只返回 JSON 数组，不要其他内容。""" def generate_outline(topic, style="技术深度", audience="开发者"): response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": OUTLINE_PROMPT.format( topic=topic, style=style, audience=audience )}], temperature=0.7, max_tokens=2048 ) return response.choices[0].message.content def generate_article(topic, outline, style="技术深度", audience="开发者"): response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": ARTICLE_PROMPT.format( topic=topic, outline=outline, style=style, audience=audience )}], temperature=0.7, max_tokens=8192 ) return response.choices[0].message.content def generate_series(topic, count=5): response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": SERIES_PROMPT.format( topic=topic, count=count )}], temperature=0.7, max_tokens=2048 ) text = response.choices[0].message.content text = text.replace("```json", "").replace("```", "").strip() return json.loads(text) def create_front_matter(title, description, tags, platform="hugo"): date = datetime.now().strftime("%Y-%m-%d") if platform == "hugo": return f"""--- title: "{title}" date: {date} draft: false tags: [{", ".join([f'"{t}"' for t in tags])}] description: "{description}" categories: ["技术"] ---""" elif platform == "hexo": return f"""--- title: {title} date: {date} tags: [{", ".join(tags)}] description: {description} ---""" else: return f"""--- title: {title} date: {date} tags: [{", ".join(tags)}] description: {description} ---""" def save_article(content, filename, front_matter=""): full_content = f"{front_matter}\n\n{content}" if front_matter else content os.makedirs("output", exist_ok=True) filepath = os.path.join("output", filename) with open(filepath, "w", encoding="utf-8") as f: f.write(full_content) return filepath def main(): parser = argparse.ArgumentParser(description="DeepSeek Markdown 博客生成器") parser.add_argument("topic", help="文章主题") parser.add_argument("--style", "-s", default="技术深度", help="写作风格") parser.add_argument("--audience", "-a", default="开发者", help="目标读者") parser.add_argument("--tags", "-t", nargs="+", default=[], help="文章标签") parser.add_argument("--platform", "-p", default="hugo", choices=["hugo", "hexo", "jekyll", "vitepress"]) parser.add_argument("--series", type=int, help="生成系列文章规划（指定数量）") parser.add_argument("--desc", "-d", default="", help="文章描述") args = parser.parse_args() if args.series: print(f"正在生成「{args.topic}」系列文章规划 ({args.series} 篇)...") series = generate_series(args.topic, args.series) filepath = save_article(json.dumps(series, ensure_ascii=False, indent=2), "series_plan.json") print(f"系列规划已保存至: {filepath}") for item in series: print(f" [{item['order']}] {item['title']} - {item['description']}") return print(f"正在生成「{args.topic}」大纲...") outline = generate_outline(args.topic, args.style, args.audience) print("大纲:\n" + outline) print(f"\n正在生成完整文章...") article = generate_article(args.topic, outline, args.style, args.audience) slug = args.topic.lower().replace(" ", "-").replace("/", "-") filename = f"{slug}.md" description = args.desc or f"关于{args.topic}的深度技术文章" tags = args.tags or ["技术", "教程"] front_matter = create_front_matter(args.topic, description, tags, args.platform) filepath = save_article(article, filename, front_matter) word_count = len(article) print(f"\n文章已保存至: {filepath}") print(f"字数: {word_count}") if __name__ == "__main__": main()

运行方式

# 安装依赖 pip install openai # 生成单篇文章 python blog_generator.py "Python 异步编程最佳实践" --tags Python 异步 编程 # 指定写作风格和目标读者 python blog_generator.py "Kubernetes 入门指南" --style "通俗易懂" --audience "运维新手" # 生成系列文章规划 python blog_generator.py "微服务架构设计" --series 5 # 指定博客平台 python blog_generator.py "React Hooks 深度解析" --platform hexo

预期效果

• 输入主题自动生成结构化大纲和完整 Markdown 文章
• 自动生成 SEO 友好的 front matter（Hugo/Hexo/Jekyll/VitePress）
• 支持系列文章规划，一次性生成多篇文章的选题和描述
• 文章包含代码示例、引言、总结等完整结构

技术栈

完整代码

#!/usr/bin/env python3 # ai_terminal.py import os import json from datetime import datetime from openai import OpenAI from rich.console import Console from rich.markdown import Markdown from rich.panel import Panel from rich.prompt import Prompt from rich.table import Table console = Console() client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") HISTORY_FILE = os.path.expanduser("~/.deepseek_terminal_history.json") MODELS = { "1": {"name": "deepseek-chat", "label": "DeepSeek V3 (对话)", "desc": "通用对话模型，适合日常使用"}, "2": {"name": "deepseek-reasoner", "label": "DeepSeek R1 (推理)", "desc": "推理模型，适合复杂问题分析"}, } def load_history(): if os.path.exists(HISTORY_FILE): with open(HISTORY_FILE, "r", encoding="utf-8") as f: return json.load(f) return {"conversations": [], "settings": {"model": "deepseek-chat"}} def save_history(history): with open(HISTORY_FILE, "w", encoding="utf-8") as f: json.dump(history, f, ensure_ascii=False, indent=2) def show_banner(): banner = """ [bold cyan] ____ ____ __ / __ \\___ ____ / __/_________ / /__ / / / / _ \\/ __ \\/ /_/ ___/ __ \\/ //_/ / /_/ / __/ /_/ / __(__ ) /_/ / ,< /_____/\\___/ .___/_/ /____/\\____/_/|_| /_/ [/bold cyan] [dim]AI 对话终端 v1.0 | 输入 /help 查看命令 | /quit 退出[/dim]""" console.print(banner) def show_help(): help_table = Table(title="可用命令") help_table.add_column("命令", style="cyan") help_table.add_column("说明", style="green") help_table.add_row("/help", "显示帮助信息") help_table.add_row("/model", "切换模型 (V3/R1)") help_table.add_row("/history", "查看对话历史") help_table.add_row("/clear", "清除当前对话") help_table.add_row("/save", "保存当前对话") help_table.add_row("/load [id]", "加载历史对话") help_table.add_row("/quit", "退出程序") console.print(help_table) def show_model_selector(): table = Table(title="选择模型") table.add_column("编号", style="cyan") table.add_column("模型", style="green") table.add_column("说明") for key, model in MODELS.items(): table.add_row(key, model["label"], model["desc"]) console.print(table) def chat(messages, model="deepseek-chat"): try: response = client.chat.completions.create( model=model, messages=messages, temperature=0.7, max_tokens=4096 ) return response.choices[0].message.content except Exception as e: return f"[错误] API 调用失败: {str(e)}" def main(): history = load_history() show_banner() messages = [{"role": "system", "content": "你是一个有用的 AI 助手，由 DeepSeek 驱动。请用中文回答。"}] current_model = history["settings"]["model"] conv_id = datetime.now().strftime("%Y%m%d_%H%M%S") while True: try: user_input = Prompt.ask("\n[bold green]你[/bold green]") if user_input.startswith("/"): cmd = user_input.strip().lower() if cmd == "/quit": console.print("[yellow]再见！[/yellow]") break elif cmd == "/help": show_help() elif cmd == "/model": show_model_selector() choice = Prompt.ask("选择模型编号", default="1") if choice in MODELS: current_model = MODELS[choice]["name"] history["settings"]["model"] = current_model save_history(history) console.print(f"[green]已切换至: {MODELS[choice]['label']}[/green]") elif cmd == "/clear": messages = [messages[0]] conv_id = datetime.now().strftime("%Y%m%d_%H%M%S") console.print("[green]对话已清除[/green]") elif cmd == "/save": history["conversations"].append({ "id": conv_id, "date": datetime.now().isoformat(), "model": current_model, "messages": messages[1:] }) save_history(history) console.print(f"[green]对话已保存 (ID: {conv_id})[/green]") elif cmd == "/history": if not history["conversations"]: console.print("[dim]暂无历史对话[/dim]") else: for conv in history["conversations"][-10:]: console.print(f" [cyan]{conv['id']}[/cyan] {conv['date'][:10]} ({conv['model']}) - {len(conv['messages'])} 轮") elif cmd.startswith("/load"): parts = user_input.split() if len(parts) > 1: target_id = parts[1] for conv in history["conversations"]: if conv["id"] == target_id: messages = [messages[0]] + conv["messages"] conv_id = target_id current_model = conv["model"] console.print(f"[green]已加载对话: {target_id}[/green]") break else: console.print("[red]未找到该对话[/red]") continue messages.append({"role": "user", "content": user_input}) with console.status("[cyan]思考中...[/cyan]"): reply = chat(messages, current_model) messages.append({"role": "assistant", "content": reply}) console.print(f"\n[bold blue]DeepSeek[/bold blue]:") console.print(Markdown(reply)) except KeyboardInterrupt: console.print("\n[yellow]再见！[/yellow]") break except EOFError: break if __name__ == "__main__": main()

运行方式

# 安装依赖 pip install openai rich # 运行 python ai_terminal.py

预期效果

• 终端内直接与 DeepSeek 对话，支持 Markdown 渲染和代码高亮
• 支持 V3 和 R1 模型实时切换
• 对话历史自动保存到本地 JSON 文件，支持加载和搜索
• 通过 / 命令快速操作，无需离开终端

技术栈

完整代码

#!/usr/bin/env python3 # email_reply.py import os import imaplib import smtplib import email from email.mime.text import MIMEText from email.header import decode_header from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") # 邮件配置 IMAP_SERVER = os.getenv("IMAP_SERVER", "imap.gmail.com") IMAP_PORT = int(os.getenv("IMAP_PORT", "993")) SMTP_SERVER = os.getenv("SMTP_SERVER", "smtp.gmail.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) EMAIL_USER = os.getenv("EMAIL_USER") EMAIL_PASS = os.getenv("EMAIL_PASS") REPLY_TEMPLATES = { "产品咨询": """感谢您的咨询！关于 {product} 的问题，以下是详细信息： {details} 如有其他问题，请随时联系我们。""", "售后支持": """感谢您联系售后支持。关于您反馈的问题，我们建议： {solution} 如需进一步协助，请提供更多信息。""", "商务合作": """感谢您的合作意向！我们非常乐意与您进一步沟通。 {response} 期待与您的合作！""", "其他": """感谢您的来信。 {response} 祝好！""" } CLASSIFY_PROMPT = """你是一个邮件分类助手。请将以下邮件分类为以下类别之一： - 产品咨询：询问产品功能、价格、使用方法 - 售后支持：投诉、故障报告、退换货 - 商务合作：合作意向、商务洽谈 - 垃圾邮件：广告、诈骗、无关内容 - 其他：无法归类的邮件 只返回类别名称，不要其他内容。 邮件主题：{subject} 邮件内容：{body}""" REPLY_PROMPT = """你是一个专业的邮件回复助手。请为以下邮件生成回复内容。 邮件类别：{category} 邮件主题：{subject} 邮件内容：{body} 回复模板： {template} 请生成具体的回复内容，填入模板的变量中。要求： 1. 回复专业、礼貌、简洁 2. 针对邮件中的具体问题给出实质性回答 3. 如果邮件信息不足，请主动询问需要补充的信息 请以 JSON 格式返回，包含填充模板所需的变量： {{"product": "...", "details": "...", "solution": "...", "response": "..."}}""" def decode_email_header(header): if not header: return "" decoded_parts = decode_header(header) subject = "" for part, charset in decoded_parts: if isinstance(part, bytes): subject += part.decode(charset or "utf-8", errors="ignore") else: subject += str(part) return subject def get_email_body(msg): body = "" if msg.is_multipart(): for part in msg.walk(): content_type = part.get_content_type() if content_type == "text/plain": try: body = part.get_payload(decode=True).decode("utf-8", errors="ignore") except: body = str(part.get_payload()) break else: try: body = msg.get_payload(decode=True).decode("utf-8", errors="ignore") except: body = str(msg.get_payload()) return body[:3000] def fetch_unread_emails(limit=10): mail = imaplib.IMAP4_SSL(IMAP_SERVER, IMAP_PORT) mail.login(EMAIL_USER, EMAIL_PASS) mail.select("inbox") _, data = mail.search(None, "UNSEEN") email_ids = data[0].split()[-limit:] emails = [] for eid in email_ids: _, msg_data = mail.fetch(eid, "(RFC822)") msg = email.message_from_bytes(msg_data[0][1]) subject = decode_email_header(msg["Subject"]) sender = msg["From"] body = get_email_body(msg) emails.append({"id": eid.decode(), "subject": subject, "sender": sender, "body": body}) mail.logout() return emails def classify_email(subject, body): response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": CLASSIFY_PROMPT.format( subject=subject[:200], body=body[:2000] )}], temperature=0.1, max_tokens=50 ) return response.choices[0].message.content.strip() def generate_reply(category, subject, body): template = REPLY_TEMPLATES.get(category, REPLY_TEMPLATES["其他"]) response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": REPLY_PROMPT.format( category=category, subject=subject[:200], body=body[:2000], template=template )}], temperature=0.3, max_tokens=2048 ) import json try: text = response.choices[0].message.content text = text.replace("```json", "").replace("```", "").strip() vars_dict = json.loads(text) except: vars_dict = {"response": "我们已收到您的邮件，将尽快回复您。"} return template.format(**{k: v for k, v in vars_dict.items() if k in template}) def send_reply(to_address, subject, body): msg = MIMEText(body, "plain", "utf-8") msg["Subject"] = f"Re: {subject}" msg["From"] = EMAIL_USER msg["To"] = to_address server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT) server.starttls() server.login(EMAIL_USER, EMAIL_PASS) server.send_message(msg) server.quit() print(f"回复已发送至: {to_address}") def main(): print("正在检查未读邮件...") emails = fetch_unread_emails() if not emails: print("没有未读邮件") return print(f"共发现 {len(emails)} 封未读邮件\n") for i, mail in enumerate(emails): print(f"\n{'='*50}") print(f"[{i+1}/{len(emails)}] 发件人: {mail['sender']}") print(f"主题: {mail['subject']}") category = classify_email(mail["subject"], mail["body"]) if category == "垃圾邮件": print(f"分类: {category} (已跳过)") continue print(f"分类: {category}") reply = generate_reply(category, mail["subject"], mail["body"]) print(f"\n生成的回复:\n{reply}") choice = input("\n是否发送此回复? (y=发送 / n=跳过 / e=编辑): ").strip().lower() if choice == "y": send_reply(mail["sender"], mail["subject"], reply) elif choice == "e": custom_reply = input("请输入你的回复: ") if custom_reply.strip(): send_reply(mail["sender"], mail["subject"], custom_reply) print("自定义回复已发送") else: print("已跳过") if __name__ == "__main__": if not EMAIL_USER or not EMAIL_PASS: print("请设置环境变量 EMAIL_USER 和 EMAIL_PASS") print("对于 Gmail，需要使用应用专用密码") exit(1) main()

运行方式

# 安装依赖 pip install openai # 设置环境变量 (Gmail 示例) export EMAIL_USER="your-email@gmail.com" export EMAIL_PASS="your-app-password" export IMAP_SERVER="imap.gmail.com" export SMTP_SERVER="smtp.gmail.com" # 运行 python email_reply.py

预期效果

• 自动检测未读邮件并分类（产品咨询/售后/商务/垃圾邮件）
• 根据分类使用对应模板生成专业回复
• 人工审核环节：发送前确认，支持编辑自定义回复
• 自动过滤垃圾邮件，提高处理效率

技术栈

完整代码

# resume_optimizer.py import os import streamlit as st import fitz from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") ANALYZE_PROMPT = """你是一位资深 HR 和简历优化专家。请分析以下简历与职位描述的匹配度。 简历内容： {resume} 职位描述： {jd} 请从以下维度进行分析，并给出 JSON 格式的结果： 1. 技能匹配度（0-100%） 2. 经验匹配度（0-100%） 3. 综合匹配度（0-100%） 4. 匹配的技能列表 5. 缺失的技能列表 6. 简历优点（3-5条） 7. 简历改进建议（3-5条） 8. ATS 优化建议（3-5条） 返回 JSON 格式： {{ "skill_match": 数字, "experience_match": 数字, "overall_match": 数字, "matched_skills": ["技能1", "技能2"], "missing_skills": ["技能1", "技能2"], "strengths": ["优点1", "优点2"], "improvements": ["建议1", "建议2"], "ats_suggestions": ["建议1", "建议2"] }}""" OPTIMIZE_PROMPT = """你是一位资深简历优化专家。请根据以下分析和建议，优化简历内容。 原始简历： {resume} 职位描述： {jd} 匹配分析： {analysis} 优化要求： 1. 突出与 JD 匹配的技能和经验 2. 使用行业关键词（ATS 友好） 3. 量化成果（使用数字和百分比） 4. 保持原简历的真实信息，不编造经历 5. 优化措辞和排版结构 请输出优化后的完整简历。""" def extract_pdf_text(pdf_file): doc = fitz.open(stream=pdf_file.read(), filetype="pdf") text = "" for page in doc: text += page.get_text() return text def analyze_resume(resume_text, jd_text): response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": ANALYZE_PROMPT.format( resume=resume_text[:5000], jd=jd_text[:3000] )}], temperature=0.1, max_tokens=2048 ) import json content = response.choices[0].message.content content = content.replace("```json", "").replace("```", "").strip() return json.loads(content) def optimize_resume(resume_text, jd_text, analysis): response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": OPTIMIZE_PROMPT.format( resume=resume_text[:5000], jd=jd_text[:3000], analysis=json.dumps(analysis, ensure_ascii=False, indent=2) )}], temperature=0.3, max_tokens=4096 ) return response.choices[0].message.content st.set_page_config(page_title="DeepSeek 简历优化", layout="wide") st.title("DeepSeek 简历优化工具") col1, col2 = st.columns(2) with col1: st.subheader("上传简历") resume_file = st.file_uploader("上传 PDF 简历", type=["pdf"], key="resume") if resume_file: resume_text = extract_pdf_text(resume_file) st.text_area("简历内容预览", resume_text, height=300) with col2: st.subheader("职位描述") jd_text = st.text_area("粘贴职位描述 (JD)", height=300, placeholder="请粘贴目标职位的完整描述...") if resume_file and jd_text and st.button("开始分析", type="primary"): with st.spinner("正在分析简历与职位匹配度..."): analysis = analyze_resume(resume_text, jd_text) st.subheader("匹配度分析") col_a, col_b, col_c = st.columns(3) col_a.metric("技能匹配", f"{analysis.get('skill_match', 0)}%") col_b.metric("经验匹配", f"{analysis.get('experience_match', 0)}%") col_c.metric("综合匹配", f"{analysis.get('overall_match', 0)}%") col_m, col_n = st.columns(2) with col_m: st.markdown("**匹配的技能**") for skill in analysis.get("matched_skills", []): st.markdown(f"- {skill}") with col_n: st.markdown("**缺失的技能**") for skill in analysis.get("missing_skills", []): st.markdown(f"- {skill}") st.markdown("**简历优点**") for s in analysis.get("strengths", []): st.markdown(f"- {s}") st.markdown("**改进建议**") for s in analysis.get("improvements", []): st.markdown(f"- {s}") st.markdown("**ATS 优化建议**") for s in analysis.get("ats_suggestions", []): st.markdown(f"- {s}") if st.button("生成优化版简历"): with st.spinner("正在优化简历..."): optimized = optimize_resume(resume_text, jd_text, analysis) st.subheader("优化后的简历") st.markdown(optimized) st.download_button("下载优化版简历", optimized, "optimized_resume.md")

运行方式

# 安装依赖 pip install streamlit PyMuPDF openai # 运行 streamlit run resume_optimizer.py # 访问 http://localhost:8501

预期效果

• 上传 PDF 简历和粘贴 JD 后，自动分析匹配度
• 三维度评分（技能/经验/综合），直观展示匹配情况
• 列出匹配和缺失的技能，给出具体改进建议
• 一键生成 ATS 友好的优化版简历，可下载为 Markdown

技术栈

完整代码

#!/usr/bin/env python3 # daily_briefing.py import os import json import smtplib import feedparser from datetime import datetime from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from openai import OpenAI client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com") # 邮件配置 SMTP_SERVER = os.getenv("SMTP_SERVER", "smtp.gmail.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) EMAIL_USER = os.getenv("EMAIL_USER") EMAIL_PASS = os.getenv("EMAIL_PASS") RECIPIENTS = os.getenv("RECIPIENTS", "").split(",") # AI 新闻 RSS 源 RSS_FEEDS = [ "https://hnrss.org/frontpage?count=20", "https://arxiv.org/rss/cs.AI", "https://www.reddit.com/r/MachineLearning/.rss", ] SUMMARIZE_PROMPT = """你是一位 AI 行业分析师。请为以下新闻条目生成每日早报摘要。 要求： 1. 将新闻分类为：大模型动态、开源工具、行业应用、学术研究、其他 2. 每个类别选择最重要的 3-5 条新闻 3. 每条新闻包含：标题、一句话摘要（30字以内）、重要性评分（1-5星） 4. 生成一个"今日亮点"区块，突出最重要的 1-2 条新闻 5. 在末尾添加"今日思考"（100字以内的行业洞察） 新闻条目（标题 + 摘要）： {articles} 请以 JSON 格式返回： {{ "date": "日期", "highlight": [{{"title": "...", "summary": "...", "reason": "..."}}], "categories": {{ "大模型动态": [{{"title": "...", "summary": "...", "rating": 5}}], "开源工具": [...], "行业应用": [...], "学术研究": [...], "其他": [...] }}, "thought": "今日思考内容" }}""" def fetch_news(): articles = [] for feed_url in RSS_FEEDS: try: feed = feedparser.parse(feed_url) for entry in feed.entries[:10]: title = entry.get("title", "无标题") summary = entry.get("summary", entry.get("description", "")) summary = summary[:200] articles.append(f"- {title}: {summary}") except Exception as e: print(f"获取 RSS 失败 ({feed_url}): {e}") return articles def generate_briefing(articles): articles_text = "\n".join(articles[:50]) response = client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": SUMMARIZE_PROMPT.format( articles=articles_text[:8000] )}], temperature=0.3, max_tokens=4096 ) content = response.choices[0].message.content content = content.replace("```json", "").replace("```", "").strip() return json.loads(content) def build_html_email(briefing): date_str = briefing.get("date", datetime.now().strftime("%Y年%m月%d日")) html = f"""

' return html def send_email(html_content): msg = MIMEMultipart("alternative") msg["Subject"] = f"AI 每日早报 - {datetime.now().strftime('%Y-%m-%d')}" msg["From"] = EMAIL_USER msg["To"] = ", ".join(RECIPIENTS) msg.attach(MIMEText(html_content, "html", "utf-8")) server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT) server.starttls() server.login(EMAIL_USER, EMAIL_PASS) server.send_message(msg) server.quit() print(f"早报已发送至: {', '.join(RECIPIENTS)}") def main(): print(f"[{datetime.now()}] 开始生成每日 AI 早报...") print("正在获取新闻...") articles = fetch_news() print(f"获取到 {len(articles)} 条新闻") print("正在生成早报...") briefing = generate_briefing(articles) html = build_html_email(briefing) if EMAIL_USER and RECIPIENTS and RECIPIENTS[0]: print("正在发送邮件...") send_email(html) else: output_path = f"briefing_{datetime.now().strftime('%Y%m%d')}.html" with open(output_path, "w", encoding="utf-8") as f: f.write(html) print(f"早报已保存至: {output_path}") if __name__ == "__main__": if not EMAIL_USER or not EMAIL_PASS: print("未配置邮件发送，将保存为本地 HTML 文件") print("如需发送邮件，请设置 EMAIL_USER、EMAIL_PASS、RECIPIENTS 环境变量") main()

运行方式

# 安装依赖 pip install openai feedparser # 设置环境变量 export EMAIL_USER="your-email@gmail.com" export EMAIL_PASS="your-app-password" export RECIPIENTS="user1@example.com,user2@example.com" # 手动运行 python daily_briefing.py # 配置 cron 定时任务 (每天早上 8:00) # crontab -e # 0 8 * * * cd /path/to/project && python daily_briefing.py

预期效果

• 自动从多个 RSS 源聚合 AI 领域新闻
• DeepSeek 智能分类和摘要，生成结构化的早报
• 精美的 HTML 邮件模板，包含今日亮点和行业洞察
• 支持 cron 定时调度，每天自动发送