{inline(m.group(2))}
") continue if m: if in_list: out.append(""); in_list = False out.append(f"{inline(s)}
") if in_list: out.append("") return "\n".join(out) def export_pdf(report_md, session): os.makedirs(TMP_DIR, exist_ok=True) ts = time.strftime("%Y%m%d_%H%M%S") date = time.strftime("%Y-%m-%d") html_path = os.path.join(TMP_DIR, f"report_{ts}.html") pdf_path = os.path.join(TMP_DIR, f"yy_wenlaoyang_{ts}.pdf") html = REPORT_TMPL.replace("__BODY__", md2html(report_md)).replace("__DATE__", date) with open(html_path, "w", encoding="utf-8") as f: f.write(html) subprocess.run( [CHROME, "--headless=new", "--disable-gpu", "--no-sandbox", "--no-pdf-header-footer", f"--print-to-pdf={pdf_path}", f"file://{html_path}"], timeout=60, capture_output=True) if os.path.exists(pdf_path) and os.path.getsize(pdf_path) > 1000: return pdf_path return None def build_stats(): """聚合 JSONL：会话数/问题数/每日趋势/关键词频次/最近问题""" if not os.path.exists(LOG_FILE): return {"sessions": 0, "questions": 0, "daily": {}, "keywords": {}, "recent": []} KW = {"获客": ["获客", "客源", "流量", "线索", "留资"], "短视频/直播": ["抖音", "短视频", "直播", "视频号", "快手"], "小红书/私域": ["小红书", "私域", "微信", "社群", "朋友圈"], "定价/利润": ["定价", "多少钱", "利润", "成本", "收费", "降价"], "员工/团队": ["员工", "团队", "招聘", "管理", "合伙人"], "复购/转化": ["复购", "转化", "成交", "回头客", "签单"], "门店/本地": ["店", "到店", "餐厅", "餐饮", "本地", "门店"], "AI应用": ["AI", "ai", "人工智能", "数字人", "GPT", "智谱"]} sessions, questions, daily, kwcount, recent = set(), 0, {}, {}, [] feedbacks = [] for line in open(LOG_FILE, encoding="utf-8"): try: r = json.loads(line) except Exception: continue sessions.add(r.get("session", "")) if r.get("role") == "feedback": feedbacks.append({"ts": r.get("ts"), "text": r.get("content", "")[:200]}) if r.get("role") == "user": questions += 1 day = r.get("ts", "")[:10] daily[day] = daily.get(day, 0) + 1 c = r.get("content", "") if len(recent) < 30: recent.append({"ts": r.get("ts"), "q": c[:60]}) for cat, kws in KW.items(): if any(k in c for k in kws): kwcount[cat] = kwcount.get(cat, 0) + 1 return {"sessions": len(sessions), "questions": questions, "daily": dict(sorted(daily.items())[-14:]), "keywords": dict(sorted(kwcount.items(), key=lambda x: -x[1])), "recent": recent[-10:], "feedbacks": feedbacks[-10:]} DIR = os.path.dirname(os.path.abspath(__file__)) PORT = 8899 API_URL = "https://open.bigmodel.cn/api/anthropic/v1/messages" # Anthropic兼容端点（glm-5.3-flash 资源包在这里） MODEL = "glm-5.3-flash" # 老杨指定主力模型（Claude Code 同款） FALLBACK_URL = "https://open.bigmodel.cn/api/paas/v4/chat/completions" FALLBACK_MODEL = "glm-4.5-flash" # 智谱内部降级（快、余额独立） DS_URL = "https://api.deepseek.com/chat/completions" DS_MODEL = "deepseek-chat" # 终极兜底：智谱全挂（额度耗尽/故障）时接 DeepSeek，客户无感知 def get_key(): return os.environ.get("ZHIPU_API_KEY") or os.environ.get("ZHIPUAI_API_KEY") or "" def get_ds_key(): return os.environ.get("DEEPSEEK_API_KEY") or "" def call_deepseek(messages, temperature=0.7, max_tokens=900, timeout=60): """DeepSeek-chat（OpenAI 兼容协议），智谱全挂时的终极兜底""" if not get_ds_key(): raise RuntimeError("DEEPSEEK_API_KEY 未配置") body = {"model": DS_MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens} req = urllib.request.Request( DS_URL, data=json.dumps(body).encode(), headers={"Content-Type": "application/json", "Authorization": f"Bearer {get_ds_key()}"}) with urllib.request.urlopen(req, timeout=timeout) as resp: out = json.load(resp) return out["choices"][0]["message"]["content"] def build_prompt(q): """路由 prompt：用户问题 + 资产清单 → 严格 JSON 诊断""" data = json.load(open(os.path.join(DIR, 'yy_workbench_data.json'))) catalog = "\n".join( f'{c["id"]}|{c["name"]}|{c["d1"]}|{c["d2"]}|{c["prob"][:40]}' for c in data["cards"] if c["st"] != "Merged") knowledge = "\n".join(f'{k["id"]}|{k["name"]}|{k["summary"][:30]}' for k in data["knowledge"]) return f"""你是杨运才的「AI能力资产路由器+商业诊断师」，融合「问问老杨」方法论：先纠偏问题，再挖根因，最后给7天可执行行动清单。杨运才有一个能力资产库（107个Skill+14条方法论知识）。 用户输入一个商业/工作问题，你要输出完整诊断（不要只纠偏就停）： ① 判断问题域 ② 一句话点破问题本质（可纠偏）③ 根因诊断 ④ 给出3-5条具体的7天行动清单（每条一句话、可执行、含具体渠道/动作/数字标准，禁止"加强""优化"类空话）⑤ 推荐最该调用的能力（最多4个）和知识（最多2个）。 【能力清单】 {catalog} 【知识清单】 {knowledge} 【用户问题】{q} 严格只返回 JSON（不要 markdown 代码块）： {{"domain":"01 商业","reasoning":"50字内点破问题本质","answer":"80字内根因诊断","actions":["第1天:...","第2-3天:...","第4-7天:..."],"skillIds":["SK001"],"knowledgeIds":["K002"]}}""" # ============ 问问老杨 · 多轮追问诊断 ============ LAOYANG_SYSTEM = """你是「问问老杨」（杨运才本人）：苏格拉底式AI商业诊断师。用户在工作台已收到一次路由诊断，现在与你做多轮追问式诊断。 【六阶段流程】①问题定义（他问的问题对不对）→②根因（为什么发生）→③目标与约束（预算/人手/时间）→④方案比较（为什么选这条不选那条）→⑤验证选择（先花小钱试哪个）→⑥行动反馈（谁做/多久/看什么指标/什么算成）。问诊段走①—③，收网段一次交付④—⑥。 【字数纪律（最重要）】每轮回复严格 ≤150字：1-2句判断 + 1个问题，说完就停，禁止长篇大论、禁止排比铺陈、禁止一次问多个问题。只有最终收网的完整诊断报告可以放宽到 ≤500字。 【铁律】 - 先给判断再提问，一轮只问一个问题；问诊信息够了就收网，不车轱辘 - 当前阶段没问透不进下一阶段；进了不回头重问 - 看家武器按需调用：留资率>播放量（流量问题先纠这个）；赚钱三层验证（凭什么有人买你×算得过账×活得到）；先分阶段再查四层（初创0—1只谈生存：首批客户/成交/案例，别聊品牌） - 【自动收网】当问诊信息足够（①②③问透，通常3—5轮），**不再提问、不等用户开口**，直接输出完整诊断报告，且输出第一行必须是 [REPORT]（系统标记，不会显示给用户）。用户主动要报告时同样以 [REPORT] 开头。报告结构固定（markdown加粗小标题）： **阶段判断**（0-1/1-10/10-100哪一阶段，一句话）→**根因链**（2-3条每条一行）→**最小行动方案**（1-3个动作）→**最小可验证实验**：先找出这个决定背后最需要验证的3个假设→选出最可能改变最终结论的那1个→围绕它设计低成本、可逆、7天内能完成的最小实验，写清六件事：①具体要做什么②投入多少时间和资源③观察什么指标④什么结果支持继续⑤什么结果提醒我停止⑥实验结束能获得什么新信息→**明天就能开始的第一个动作**（一句话）。报告≤1200字，其余对话轮仍守≤150字纪律 - 人格真实：2017年1.2万阅读公众号文章当日加微上千、卖过大米、负债300万翻身、女儿北大；不编造新战绩 - 语气：直接、像老乡聊天但不客气，敢说「你这个问法就错了」""" def call_anthropic(messages, temperature=0.7, max_tokens=2048, timeout=90): """glm-5.3-flash @ Anthropic兼容端点（主力）。system role 提到顶层 system 字段""" sys_txt = "" msgs = [] for m in messages: if m["role"] == "system": sys_txt += m["content"] + "\n" else: msgs.append(m) body = {"model": MODEL, "max_tokens": max_tokens, "temperature": temperature, "thinking": {"type": "disabled"}, # 智谱扩展：关思考（默认开会思考1分半） "messages": msgs} if sys_txt: body["system"] = sys_txt.strip() req = urllib.request.Request( API_URL, data=json.dumps(body).encode(), headers={"Content-Type": "application/json", "x-api-key": get_key(), "anthropic-version": "2023-06-01"}) with urllib.request.urlopen(req, timeout=timeout) as resp: out = json.load(resp) return "".join(b.get("text", "") for b in out.get("content", []) if b.get("type") == "text") def call_flash(messages, temperature=0.3, timeout=60, max_tokens=900): """glm-4.5-flash @ chat端点（降级备用：快、余额独立）""" body = {"model": FALLBACK_MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "thinking": {"type": "disabled"}} req = urllib.request.Request( FALLBACK_URL, data=json.dumps(body).encode(), headers={"Content-Type": "application/json", "Authorization": f"Bearer {get_key()}"}) with urllib.request.urlopen(req, timeout=timeout) as resp: out = json.load(resp) return out["choices"][0]["message"]["content"] def do_route(q): prompt = [{"role": "user", "content": build_prompt(q)}] try: text = call_deepseek(prompt, temperature=0.3, max_tokens=2048) m = re.search(r'\{.*\}', text, re.S) if m: return json.loads(m.group(0)) except Exception: pass # DeepSeek 异常 → 智谱 try: text = call_anthropic(prompt, temperature=0.3, max_tokens=2048) m = re.search(r'\{.*\}', text, re.S) if m: return json.loads(m.group(0)) except Exception: pass text = call_flash(prompt) m = re.search(r'\{.*\}', text, re.S) return json.loads(m.group(0)) if m else {} def do_chat(history, q, mode="chat"): msgs = [{"role": "system", "content": LAOYANG_SYSTEM}] for h in history[-16:]: if h.get("role") in ("user", "assistant") and h.get("content"): msgs.append({"role": h["role"], "content": str(h["content"])[:3000]}) if q: msgs.append({"role": "user", "content": str(q)[:2000]}) mt = 2600 if mode == "report" else 900 # 报告轮放宽配额 # 主力=DeepSeek（1.5s/轮），降级=智谱 5.3 → 4.5（老杨 2026-10-17 拍板切换） try: reply = call_deepseek(msgs, temperature=0.7, max_tokens=mt) if reply.strip(): return reply except Exception: pass # DeepSeek 异常 → 智谱接管 try: reply = call_anthropic(msgs, temperature=0.7, max_tokens=mt) if reply.strip(): return reply except Exception: pass return call_flash(msgs, temperature=0.7, max_tokens=mt) class Handler(SimpleHTTPRequestHandler): def __init__(self, *a, **kw): super().__init__(*a, directory=DIR, **kw) def log_message(self, fmt, *args): pass # 静默访问日志 def _json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if self.path == "/api/ping": return self._json({"ok": True, "ai": bool(get_key()), "backup_ds": bool(get_ds_key())}) if self.path == "/api/stats": return self._json({"ok": True, **build_stats()}) return super().do_GET() def do_POST(self): try: length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length)) if length else {} if self.path == "/api/route": q = str(body.get("q", "")).strip()[:500] if not q: return self._json({"ok": False, "error": "empty q"}) if not get_key(): return self._json({"ok": False, "error": "ZHIPU_API_KEY 未设置"}) return self._json({"ok": True, **do_route(q)}) ip = self.client_address[0] if self.path in ("/api/chat", "/api/export_pdf"): if not rate_ok(ip): return self._json({"ok": False, "error": "调用太频繁了，每小时最多30次，休息一下再来"}, 429) if self.path == "/api/chat": if not get_key(): return self._json({"ok": False, "error": "ZHIPU_API_KEY 未设置"}) sess = str(body.get("sessionId", "anon"))[:40] q = str(body.get("q", "")) mode = body.get("mode", "chat") reply = do_chat(body.get("history", []), q, mode=mode) is_report = reply.strip().startswith("[REPORT]") if is_report: reply = reply.strip()[len("[REPORT]"):].lstrip() if q: log_jsonl(sess, "user", q) if reply: log_jsonl(sess, "report" if (is_report or mode == "report") else "lao", reply) return self._json({"ok": True, "reply": reply, "is_report": is_report}) if self.path == "/api/feedback": fb = str(body.get("text", "")).strip()[:1000] if not fb: return self._json({"ok": False, "error": "empty"}) log_jsonl(str(body.get("sessionId", "anon"))[:40], "feedback", fb) return self._json({"ok": True}) if self.path == "/api/export_pdf": rep = str(body.get("report", "")).strip() if not rep: return self._json({"ok": False, "error": "empty report"}) pdf = export_pdf(rep, str(body.get("sessionId", "anon"))[:40]) if not pdf: return self._json({"ok": False, "error": "PDF 生成失败（Chrome headless）"}) with open(pdf, "rb") as f: data = f.read() self.send_response(200) self.send_header("Content-Type", "application/pdf") self.send_header("Content-Disposition", "attachment") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return return self._json({"ok": False, "error": "not found"}, 404) except Exception as e: return self._json({"ok": False, "error": str(e)[:200]}) if __name__ == "__main__": srv = ThreadingHTTPServer(("127.0.0.1", PORT), Handler) # 多线程：单个慢请求不卡整站 print(f"OK server running(threaded): http://localhost:{PORT}") print(f" route+chat: {'ON (' + MODEL + ')' if get_key() else 'OFF - no ZHIPU_API_KEY'}") srv.serve_forever()