""" PPTist AI 后端服务 接收 PPTist 的 AIPPT 请求，调用 deepseek-v3 生成大纲和PPT内容 端口: 18086 """ from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse import httpx import json app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) LLM_API = "http://10.102.24.75:3000/v1/chat/completions" LLM_KEY = "sk-BlQIGRrotbVDWE5mXCPBFjVWIvJ83hldzz67xInNwzVo7pPb" LLM_MODEL = "deepseek-v3" OUTLINE_SYSTEM_PROMPT = """你是一个专业的PPT大纲生成助手。用户会给你一个主题，你需要生成一份内容丰富、结构清晰的PPT大纲。 要求： 1. 使用Markdown格式，用 # 表示演示文稿标题，## 表示章节标题，### 表示该章节下的内容页标题 2. 必须包含6个主要章节（##） 3. 每个章节下必须有2-3个内容页（###） 4. 每个内容页标题要具体、有信息量，不要太笼统 5. 内容要专业、逻辑清晰、层层递进 6. 直接输出Markdown内容，不要有多余的解释 示例格式： # 人工智能技术发展 ## 人工智能概述 ### 定义与发展历史 ### 核心技术与分支 ## 机器学习基础 ### 监督学习与无监督学习 ### 深度学习与神经网络 ### 常用算法对比 ...""" AIPPT_SYSTEM_PROMPT = """你是一个专业的PPT内容生成助手。用户会给你一份PPT大纲，你需要根据大纲生成完整、丰富、详实的PPT内容。 你必须严格按照以下JSON数组格式输出，不要输出任何其他内容。 结构规则： 1. 第一页：cover类型，包含演示标题和副标题描述 2. 第二页：contents类型，列出所有章节名称 3. 每个章节：先一个transition过渡页，然后每个###子内容对应一个content页 4. 每个content页必须包含4个items，每个item有title和text 5. text内容必须详实具体，每条20-40字，包含实质性的数据、观点或分析，不要空泛 6. 最后一页：end类型 输出格式（JSON数组，每个元素是一页slide）： [ {"type": "cover", "data": {"title": "演示标题", "text": "一句话描述演示主题和价值"}}, {"type": "contents", "data": {"items": ["章节1名称", "章节2名称", "章节3名称", "章节4名称", "章节5名称", "章节6名称"]}}, {"type": "transition", "data": {"title": "章节标题", "text": "本章将介绍什么内容，一句话概述"}}, {"type": "content", "data": {"title": "内容页标题", "items": [ {"title": "要点1标题", "text": "要点1的详细说明，包含具体信息、数据或分析"}, {"title": "要点2标题", "text": "要点2的详细说明"}, {"title": "要点3标题", "text": "要点3的详细说明"}, {"title": "要点4标题", "text": "要点4的详细说明"} ]}}, {"type": "end"} ] 重要： - 生成的总页数应在20-25页之间 - 只输出JSON数组，不要markdown代码块，不要任何解释 - 确保JSON格式完全正确，可被直接解析""" async def stream_llm(messages, temperature=0.7, max_tokens=4096): """流式调用LLM并逐块返回文本""" async with httpx.AsyncClient(timeout=120) as client: async with client.stream( "POST", LLM_API, json={ "model": LLM_MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "stream": True, }, headers={"Authorization": f"Bearer {LLM_KEY}"}, ) as resp: async for line in resp.aiter_lines(): if line.startswith("data: "): data = line[6:] if data.strip() == "[DONE]": break try: chunk = json.loads(data) delta = chunk["choices"][0].get("delta", {}) content = delta.get("content", "") if content: yield content except (json.JSONDecodeError, KeyError, IndexError): continue @app.post("/tools/aippt_outline") async def aippt_outline(request: Request): data = await request.json() content = data.get("content", "") language = data.get("language", "中文") print(f"[AIPPT] Outline request: {content}, language: {language}") user_prompt = f"请用{language}为以下主题生成PPT大纲：\n\n{content}" async def generate(): async for text in stream_llm( [ {"role": "system", "content": OUTLINE_SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ], temperature=0.7, max_tokens=2048, ): yield text return StreamingResponse(generate(), media_type="text/event-stream") @app.post("/tools/aippt") async def aippt(request: Request): data = await request.json() content = data.get("content", "") language = data.get("language", "中文") style = data.get("style", "通用") print(f"[AIPPT] Generate request, style: {style}, language: {language}") user_prompt = f"请根据以下大纲生成PPT内容，风格为「{style}」，语言为{language}：\n\n{content}" async def generate(): """流式收集LLM输出，按JSON对象级别缓冲，每检测到一个完整对象就立即发送""" buf = "" brace_depth = 0 in_string = False escape_next = False found_array = False # 是否已经跳过了数组开头的 [ try: async for chunk in stream_llm( [ {"role": "system", "content": AIPPT_SYSTEM_PROMPT}, {"role": "user", "content": user_prompt}, ], temperature=0.7, max_tokens=8192, ): for ch in chunk: # 跳过数组开头的 [ 和结尾的 ] 以及逗号分隔符 if not found_array: if ch == '[': found_array = True continue if brace_depth == 0 and ch in (' ', '\n', '\r', '\t', ',', ']'): continue buf += ch if escape_next: escape_next = False continue if ch == '\\' and in_string: escape_next = True continue if ch == '"' and not escape_next: in_string = not in_string continue if in_string: continue if ch == '{': brace_depth += 1 elif ch == '}': brace_depth -= 1 if brace_depth == 0: # 得到一个完整的JSON对象 try: obj = json.loads(buf) yield json.dumps(obj, ensure_ascii=False) + "\n" print(f"[AIPPT] Sent slide: {obj.get('type', '?')}") except json.JSONDecodeError as e: print(f"[AIPPT] JSON parse error: {e}, buf: {buf[:100]}") buf = "" except Exception as e: print(f"[AIPPT] Error: {e}") yield json.dumps({"type": "cover", "data": {"title": "生成失败", "text": str(e)[:100]}}) + "\n" yield json.dumps({"type": "end"}) + "\n" return StreamingResponse(generate(), media_type="text/event-stream") WRITING_COMMANDS = { "rewrite": "请改写以下文本，保持原意但使用不同的表达方式：", "expand": "请扩写以下文本，补充更多细节和内容：", "abbreviate": "请缩写以下文本，保留核心要点，使其更简洁：", "polish": "请润色以下文本，使其更流畅、专业：", "translate": "请将以下文本翻译为英文：", } @app.post("/tools/ai_writing") async def ai_writing(request: Request): data = await request.json() content = data.get("content", "") command = data.get("command", "rewrite") print(f"[AIPPT] Writing request, command: {command}") instruction = WRITING_COMMANDS.get(command, WRITING_COMMANDS["rewrite"]) user_prompt = f"{instruction}\n\n{content}" async def generate(): async for text in stream_llm( [ {"role": "system", "content": "你是一个专业的文字助手，帮助用户改写、扩写、缩写、润色或翻译文本。直接输出结果，不要有额外解释。"}, {"role": "user", "content": user_prompt}, ], temperature=0.7, max_tokens=2048, ): yield text return StreamingResponse(generate(), media_type="text/event-stream") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=18086)