Spaces:
Sleeping
Sleeping
| import argparse | |
| import ast | |
| import json | |
| import os | |
| import re | |
| import sys | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| from typing import Any, Dict, List, Optional, Tuple | |
| class ToolCall: | |
| name: str | |
| args: Dict[str, Any] | |
| class ToolResult: | |
| ok: bool | |
| output: str | |
| artifacts: List[str] | |
| def _tokenize(s: str) -> List[str]: | |
| s = s.lower() | |
| s = re.sub(r"[^0-9a-z\u4e00-\u9fff]+", " ", s) | |
| parts = [p.strip() for p in s.split() if p.strip()] | |
| return parts | |
| def tool_search_kb(query: str, kb_dir: str, top_k: int = 4) -> ToolResult: | |
| base = Path(kb_dir) | |
| if not base.exists(): | |
| return ToolResult(ok=True, output="[]", artifacts=[]) | |
| q_tokens = _tokenize(query) | |
| if not q_tokens: | |
| return ToolResult(ok=True, output="[]", artifacts=[]) | |
| hits: List[Tuple[int, str, str]] = [] | |
| for p in sorted(base.glob("**/*.md")): | |
| try: | |
| text = p.read_text(encoding="utf-8") | |
| except Exception: | |
| continue | |
| chunks = [c.strip() for c in re.split(r"\n\s*\n+", text) if c.strip()] | |
| for c in chunks: | |
| c_tokens = _tokenize(c) | |
| score = sum(1 for t in q_tokens if t in c_tokens) | |
| if score <= 0: | |
| continue | |
| hits.append((score, str(p), c)) | |
| hits.sort(key=lambda x: (-x[0], x[1])) | |
| picked = hits[: max(0, int(top_k))] | |
| out = [{"score": s, "source": src, "text": txt} for (s, src, txt) in picked] | |
| return ToolResult(ok=True, output=json.dumps(out, ensure_ascii=False, indent=2), artifacts=[]) | |
| class _SafeEval(ast.NodeVisitor): | |
| def visit(self, node): | |
| return super().visit(node) | |
| def generic_visit(self, node): | |
| raise ValueError(f"unsupported: {node.__class__.__name__}") | |
| def visit_Expression(self, node: ast.Expression): | |
| return self.visit(node.body) | |
| def visit_BinOp(self, node: ast.BinOp): | |
| left = self.visit(node.left) | |
| right = self.visit(node.right) | |
| if isinstance(node.op, ast.Add): | |
| return left + right | |
| if isinstance(node.op, ast.Sub): | |
| return left - right | |
| if isinstance(node.op, ast.Mult): | |
| return left * right | |
| if isinstance(node.op, ast.Div): | |
| return left / right | |
| if isinstance(node.op, ast.FloorDiv): | |
| return left // right | |
| if isinstance(node.op, ast.Mod): | |
| return left % right | |
| if isinstance(node.op, ast.Pow): | |
| return left ** right | |
| raise ValueError("unsupported operator") | |
| def visit_UnaryOp(self, node: ast.UnaryOp): | |
| v = self.visit(node.operand) | |
| if isinstance(node.op, ast.UAdd): | |
| return +v | |
| if isinstance(node.op, ast.USub): | |
| return -v | |
| raise ValueError("unsupported unary operator") | |
| def visit_Constant(self, node: ast.Constant): | |
| if isinstance(node.value, (int, float)): | |
| return node.value | |
| raise ValueError("only numbers allowed") | |
| def tool_calc(expression: str) -> ToolResult: | |
| expr = expression.strip() | |
| if not re.fullmatch(r"[0-9\.\s\+\-\*\/\(\)\%\^]+", expr): | |
| return ToolResult(ok=False, output="表达式包含不允许的字符", artifacts=[]) | |
| expr = expr.replace("^", "**") | |
| try: | |
| tree = ast.parse(expr, mode="eval") | |
| value = _SafeEval().visit(tree) | |
| except Exception as e: | |
| return ToolResult(ok=False, output=f"计算失败: {e}", artifacts=[]) | |
| return ToolResult(ok=True, output=str(value), artifacts=[]) | |
| def tool_write_text(path: str, text: str) -> ToolResult: | |
| p = Path(path) | |
| p.parent.mkdir(parents=True, exist_ok=True) | |
| p.write_text(text, encoding="utf-8") | |
| return ToolResult(ok=True, output=f"wrote: {p}", artifacts=[str(p)]) | |
| def tool_read_text(path: str) -> ToolResult: | |
| p = Path(path) | |
| if not p.exists(): | |
| return ToolResult(ok=False, output="not found", artifacts=[]) | |
| return ToolResult(ok=True, output=p.read_text(encoding="utf-8"), artifacts=[]) | |
| def _extract_math_expr(goal: str) -> Optional[str]: | |
| m = re.search(r"(\d[\d\s\+\-\*\/\(\)\%\^\.]*\d)", goal) | |
| if not m: | |
| return None | |
| expr = m.group(1) | |
| expr = re.sub(r"\s+", "", expr) | |
| if len(expr) > 120: | |
| return None | |
| if not re.search(r"[\+\-\*\/\%\^]", expr): | |
| return None | |
| return expr | |
| def _extract_faq_count(goal: str) -> Optional[int]: | |
| m = re.search(r"(\d+)\s*条\s*FAQ", goal, flags=re.IGNORECASE) | |
| if m: | |
| return max(1, min(20, int(m.group(1)))) | |
| m = re.search(r"(\d+)\s*条", goal) | |
| if m and ("faq" in goal.lower() or "常见问题" in goal): | |
| return max(1, min(20, int(m.group(1)))) | |
| if "faq" in goal.lower() or "常见问题" in goal: | |
| return 5 | |
| return None | |
| def _norm_space(s: str) -> str: | |
| return re.sub(r"\s+", " ", (s or "").strip()) | |
| def _lead_to_goal(lead: Dict[str, Any]) -> str: | |
| company = _norm_space(str(lead.get("company") or "")) | |
| contact = _norm_space(str(lead.get("contact") or "")) | |
| role = _norm_space(str(lead.get("role") or "")) | |
| channel = _norm_space(str(lead.get("channel") or "")) | |
| product = _norm_space(str(lead.get("product") or "")) | |
| stage = _norm_space(str(lead.get("stage") or "")) | |
| pain = _norm_space(str(lead.get("pain_points") or "")) | |
| budget = _norm_space(str(lead.get("budget") or "")) | |
| timeline = _norm_space(str(lead.get("timeline") or "")) | |
| notes = _norm_space(str(lead.get("notes") or "")) | |
| parts: List[str] = [] | |
| parts.append("生成一份销售线索跟进方案（含补问清单、下一步任务清单、跟进话术：微信/电话/邮件、风险提示），并写进报告。") | |
| if company: | |
| parts.append(f"公司/组织：{company}") | |
| if contact or role: | |
| who = contact if contact else "（未提供姓名）" | |
| if role: | |
| who = f"{who}（{role}）" | |
| parts.append(f"联系人：{who}") | |
| if channel: | |
| parts.append(f"来源渠道：{channel}") | |
| if product: | |
| parts.append(f"关注产品：{product}") | |
| if stage: | |
| parts.append(f"当前阶段：{stage}") | |
| if pain: | |
| parts.append(f"主要诉求/痛点：{pain}") | |
| if budget: | |
| parts.append(f"预算：{budget}") | |
| if timeline: | |
| parts.append(f"期望时间：{timeline}") | |
| if notes: | |
| parts.append(f"备注：{notes}") | |
| return "\n".join(parts) | |
| def make_plan(goal: str, kb_dir: str, out_report: str) -> List[ToolCall]: | |
| plan: List[ToolCall] = [] | |
| plan.append(ToolCall(name="search_kb", args={"query": goal, "kb_dir": kb_dir, "top_k": 6})) | |
| expr = _extract_math_expr(goal) | |
| if expr: | |
| plan.append(ToolCall(name="calc", args={"expression": expr})) | |
| plan.append(ToolCall(name="write_report", args={"goal": goal, "out_report": out_report})) | |
| return plan | |
| def _format_faq_items(snippets: List[Dict[str, Any]], n: int) -> List[str]: | |
| pool: List[str] = [] | |
| for s in snippets: | |
| t = str(s.get("text", "")).strip() | |
| t = re.sub(r"\s+", " ", t) | |
| if len(t) >= 12: | |
| pool.append(t) | |
| if not pool: | |
| pool = [ | |
| "我们提供从获客到转化的自动化流程编排能力，减少重复操作。", | |
| "支持接入常见广告与社媒渠道，并统一沉淀线索与素材。", | |
| "关键数据可追踪，可用规则触发通知与任务分配。", | |
| "提供基础的权限与审计，方便多人协作与风控。", | |
| "支持将常见问题与业务规范整理进知识库，供智能体检索。", | |
| ] | |
| items: List[str] = [] | |
| i = 0 | |
| while len(items) < n: | |
| items.append(pool[i % len(pool)]) | |
| i += 1 | |
| return items | |
| def _render_report(goal: str, snippets: List[Dict[str, Any]], faq_count: Optional[int], math_expr: Optional[str], math_result: Optional[str]) -> str: | |
| lines: List[str] = [] | |
| lines.append("# 智能体生成报告") | |
| lines.append("") | |
| lines.append("## 目标") | |
| lines.append(goal.strip()) | |
| lines.append("") | |
| if snippets: | |
| lines.append("## 知识库检索摘要") | |
| for i, s in enumerate(snippets[:4], start=1): | |
| src = os.path.basename(str(s.get("source", ""))) | |
| txt = str(s.get("text", "")).strip().replace("\n", " ") | |
| txt = re.sub(r"\s+", " ", txt) | |
| lines.append(f"- 片段{i}（{src}）：{txt}") | |
| lines.append("") | |
| if faq_count: | |
| lines.append(f"## FAQ（{faq_count} 条）") | |
| items = _format_faq_items(snippets, faq_count) | |
| for it in items: | |
| lines.append(f"- {it}") | |
| lines.append("") | |
| if math_expr: | |
| lines.append("## 计算结果") | |
| if math_result is None: | |
| lines.append(f"- 表达式：{math_expr}") | |
| lines.append("- 结果：计算失败") | |
| else: | |
| lines.append(f"- 表达式：{math_expr}") | |
| lines.append(f"- 结果：{math_result}") | |
| lines.append("") | |
| lines.append("## 结论") | |
| lines.append("已基于本地知识库完成检索，并生成可复核的报告工件。") | |
| lines.append("") | |
| return "\n".join(lines) | |
| def _lead_sections(goal: str, snippets: List[Dict[str, Any]]) -> Dict[str, Any]: | |
| lines = goal.splitlines() | |
| facts: Dict[str, str] = {} | |
| for ln in lines: | |
| if "：" in ln: | |
| k, v = ln.split("：", 1) | |
| k = _norm_space(k) | |
| v = _norm_space(v) | |
| if k and v: | |
| facts[k] = v | |
| company = facts.get("公司/组织", "") | |
| product = facts.get("关注产品", "") | |
| stage = facts.get("当前阶段", "") | |
| pain = facts.get("主要诉求/痛点", "") | |
| budget = facts.get("预算", "") | |
| timeline = facts.get("期望时间", "") | |
| summary = "；".join([p for p in [company, product, stage, pain] if p]) or "已记录线索信息。" | |
| questions: List[str] = [ | |
| "当前目标是什么（增长/获客/转化/复购/效率/风控）？优先级如何排序？", | |
| "现有流程里最耗时或最容易丢信息的环节是哪一步？", | |
| "线索来源结构与规模（渠道、日均量、转化率、客单价）大概是多少？", | |
| "团队协作方式（谁负责、如何交接、是否需要权限与审计）？", | |
| "期望在多长周期内看到可量化改进？评估指标是什么？", | |
| ] | |
| if budget: | |
| questions.append("预算口径（一次性/按月/按量）与采购流程（招采/合同/付款节点）？") | |
| if timeline: | |
| questions.append("上线时间点是否有外部约束（活动/发布/投放节奏）？") | |
| tasks: List[Dict[str, str]] = [ | |
| {"task": "补齐关键字段：目标、现状流程、线索规模、关键指标", "owner": "你", "due": "今天"}, | |
| {"task": "确认决策链路：需求方/使用方/审批方/预算方", "owner": "你", "due": "今天"}, | |
| {"task": "整理 3 个典型线索样本与跟进记录（脱敏）", "owner": "客户", "due": "本周"}, | |
| {"task": "对齐一版 SOP：触达→跟进→转化→复盘的关键动作与责任人", "owner": "你", "due": "本周"}, | |
| {"task": "输出落地方案：自动化触发点、权限与审计、指标口径", "owner": "你", "due": "本周"}, | |
| {"task": "安排演示/试用：用样本数据跑一遍闭环", "owner": "你", "due": "下周"}, | |
| ] | |
| wechat = "我先基于你们当前的线索流程整理一版“可落地的跟进方案+自动化触发点”，你看方便补充下：目标/线索规模/当前痛点三项吗？补齐后我今天给你一版可执行清单。" | |
| phone = "你好，我是这边负责增长流程自动化方案的。想快速了解下：你们目前线索从哪个渠道进来、谁负责跟进、最容易卡在哪一步？我这边把方案做成你们团队可直接执行的清单。" | |
| email = "\n".join( | |
| [ | |
| "主题：线索跟进闭环方案梳理（补问清单 + 可执行任务）", | |
| "", | |
| "你好，", | |
| "", | |
| "我已根据当前线索信息先拟了一版跟进方案（含补问清单、下一步任务、话术与风险点）。", | |
| "为保证方案能直接落地，麻烦补充 3 点：目标/线索规模/当前流程卡点（可简单一句话）。", | |
| "", | |
| "我收到后会在当天给出：", | |
| "1) 可执行的任务清单（负责人/截止时间）", | |
| "2) 自动化触发点建议", | |
| "3) 指标口径与复盘方式", | |
| "", | |
| "谢谢", | |
| ] | |
| ) | |
| risks = [ | |
| "对外触达需避免夸大承诺，话术以“可量化目标/可验证步骤”为准。", | |
| "涉及个人信息与线索数据时，需明确数据来源与留存周期，按权限最小化原则访问。", | |
| "对接渠道 API/表单时，先做字段映射与去重规则，避免重复触达与口径混乱。", | |
| ] | |
| cited: List[str] = [] | |
| for s in snippets[:3]: | |
| src = os.path.basename(str(s.get("source", ""))) | |
| txt = _norm_space(str(s.get("text", "")).replace("\n", " ")) | |
| if txt: | |
| cited.append(f"{src}：{txt}") | |
| return { | |
| "summary": summary, | |
| "questions": questions[:7], | |
| "tasks": tasks, | |
| "scripts": {"wechat": wechat, "phone": phone, "email": email}, | |
| "risks": risks, | |
| "cited": cited, | |
| } | |
| def _render_lead_report(goal: str, snippets: List[Dict[str, Any]]) -> Tuple[str, Dict[str, Any]]: | |
| data = _lead_sections(goal=goal, snippets=snippets) | |
| lines: List[str] = [] | |
| lines.append("# 线索跟进方案") | |
| lines.append("") | |
| lines.append("## 线索摘要") | |
| lines.append(data["summary"]) | |
| lines.append("") | |
| lines.append("## 需要补问（建议按顺序）") | |
| for q in data["questions"]: | |
| lines.append(f"- {q}") | |
| lines.append("") | |
| lines.append("## 下一步任务清单") | |
| for t in data["tasks"]: | |
| lines.append(f"- {t['task']}（负责人：{t['owner']}；截止：{t['due']}）") | |
| lines.append("") | |
| lines.append("## 跟进话术") | |
| lines.append("### 微信/IM") | |
| lines.append(data["scripts"]["wechat"]) | |
| lines.append("") | |
| lines.append("### 电话开场") | |
| lines.append(data["scripts"]["phone"]) | |
| lines.append("") | |
| lines.append("### 邮件") | |
| lines.append(data["scripts"]["email"]) | |
| lines.append("") | |
| lines.append("## 风险提示") | |
| for r in data["risks"]: | |
| lines.append(f"- {r}") | |
| lines.append("") | |
| if data["cited"]: | |
| lines.append("## 参考知识片段") | |
| for c in data["cited"]: | |
| lines.append(f"- {c}") | |
| lines.append("") | |
| lines.append("## 原始目标") | |
| lines.append(goal.strip()) | |
| lines.append("") | |
| return "\n".join(lines), data | |
| def _needs_report(goal: str) -> bool: | |
| g = goal.lower() | |
| if "报告" in goal or "写进报告" in goal or "写入报告" in goal: | |
| return True | |
| if "report" in g: | |
| return True | |
| return True | |
| def check_closure(goal: str, out_report: str, mode: str) -> Tuple[bool, str]: | |
| p = Path(out_report) | |
| if _needs_report(goal): | |
| if not p.exists(): | |
| return False, "报告文件不存在" | |
| txt = p.read_text(encoding="utf-8") | |
| if mode == "lead_followup": | |
| if "线索跟进方案" not in txt: | |
| return False, "报告缺少标题" | |
| if "需要补问" not in txt or "下一步任务清单" not in txt or "跟进话术" not in txt: | |
| return False, "报告结构不完整" | |
| if len(re.findall(r"^\-\s+", txt, flags=re.MULTILINE)) < 8: | |
| return False, "清单项不足" | |
| else: | |
| if "智能体生成报告" not in txt: | |
| return False, "报告缺少标题" | |
| faq_count = _extract_faq_count(goal) | |
| if faq_count: | |
| if len(re.findall(r"^\-\s+", txt, flags=re.MULTILINE)) < faq_count: | |
| return False, "FAQ 条数不足" | |
| expr = _extract_math_expr(goal) | |
| if expr: | |
| if expr not in txt: | |
| return False, "报告未包含表达式" | |
| calc_res = tool_calc(expr) | |
| if calc_res.ok and str(calc_res.output) not in txt: | |
| return False, "报告未包含计算结果" | |
| return True, "ok" | |
| def run_agent(goal: str, kb_dir: str, out_report: str, max_rounds: int = 3, mode: str = "general") -> Dict[str, Any]: | |
| round_logs: List[Dict[str, Any]] = [] | |
| state: Dict[str, Any] = {"snippets": [], "math_expr": None, "math_result": None, "artifacts": [], "structured": None} | |
| for r in range(1, max_rounds + 1): | |
| plan = make_plan(goal=goal, kb_dir=kb_dir, out_report=out_report) | |
| step_logs: List[Dict[str, Any]] = [] | |
| for step in plan: | |
| if step.name == "search_kb": | |
| res = tool_search_kb(**step.args) | |
| if res.ok: | |
| try: | |
| state["snippets"] = json.loads(res.output) | |
| except Exception: | |
| state["snippets"] = [] | |
| step_logs.append({"tool": step.name, "ok": res.ok, "output": res.output[:800]}) | |
| continue | |
| if step.name == "calc": | |
| state["math_expr"] = step.args.get("expression") | |
| res = tool_calc(**step.args) | |
| state["math_result"] = res.output if res.ok else None | |
| step_logs.append({"tool": step.name, "ok": res.ok, "output": res.output}) | |
| continue | |
| if step.name == "write_report": | |
| faq_count = _extract_faq_count(goal) | |
| if mode == "lead_followup": | |
| report, structured = _render_lead_report(goal=goal, snippets=state.get("snippets") or []) | |
| state["structured"] = structured | |
| else: | |
| report = _render_report( | |
| goal=goal, | |
| snippets=state.get("snippets") or [], | |
| faq_count=faq_count, | |
| math_expr=state.get("math_expr"), | |
| math_result=state.get("math_result"), | |
| ) | |
| res = tool_write_text(path=out_report, text=report) | |
| state["artifacts"].extend(res.artifacts) | |
| step_logs.append({"tool": step.name, "ok": res.ok, "output": res.output}) | |
| continue | |
| step_logs.append({"tool": step.name, "ok": False, "output": "unknown tool"}) | |
| ok, reason = check_closure(goal=goal, out_report=out_report, mode=mode) | |
| round_logs.append({"round": r, "plan": [c.__dict__ for c in plan], "steps": step_logs, "check": {"ok": ok, "reason": reason}}) | |
| if ok: | |
| break | |
| final = [] | |
| final.append("已完成闭环执行：规划→工具执行→自检→产出。") | |
| final.append(f"报告工件：{out_report}") | |
| expr = state.get("math_expr") | |
| if expr: | |
| final.append(f"计算：{expr} = {state.get('math_result')}") | |
| faq_count = _extract_faq_count(goal) | |
| if faq_count: | |
| final.append(f"FAQ：{faq_count} 条（见报告）") | |
| return { | |
| "goal": goal, | |
| "out_report": out_report, | |
| "round_logs": round_logs, | |
| "final_answer": "\n".join(final), | |
| "artifacts": state.get("artifacts"), | |
| "structured": state.get("structured"), | |
| "mode": mode, | |
| } | |
| def run_agent_mode(mode: str, payload: Dict[str, Any], kb_dir: str, out_report: str, max_rounds: int = 3) -> Dict[str, Any]: | |
| mode = (mode or "general").strip() | |
| if mode == "lead_followup": | |
| lead = payload.get("lead") or {} | |
| goal = _lead_to_goal(lead if isinstance(lead, dict) else {}) | |
| result = run_agent(goal=goal, kb_dir=kb_dir, out_report=out_report, max_rounds=max_rounds, mode=mode) | |
| return result | |
| goal = str(payload.get("goal") or "").strip() | |
| return run_agent(goal=goal, kb_dir=kb_dir, out_report=out_report, max_rounds=max_rounds, mode="general") | |
| def _print_round_logs(round_logs: List[Dict[str, Any]]) -> None: | |
| for r in round_logs: | |
| print(f"\n=== Round {r['round']} ===") | |
| print("Plan:") | |
| for s in r["plan"]: | |
| print(f"- {s['name']} {json.dumps(s['args'], ensure_ascii=False)}") | |
| print("Steps:") | |
| for s in r["steps"]: | |
| out = s["output"] | |
| out = out.replace("\n", " ") | |
| if len(out) > 180: | |
| out = out[:180] + "..." | |
| print(f"- {s['tool']}: ok={s['ok']} output={out}") | |
| print(f"Check: ok={r['check']['ok']} reason={r['check']['reason']}") | |
| def cmd_demo() -> int: | |
| here = Path(__file__).resolve().parent | |
| goal = "基于知识库，生成一份 5 条 FAQ，并计算 17*23 的结果写进报告" | |
| out_report = str(here / "out" / "demo_report.md") | |
| kb_dir = str(here / "kb") | |
| result = run_agent(goal=goal, kb_dir=kb_dir, out_report=out_report, max_rounds=3) | |
| _print_round_logs(result["round_logs"]) | |
| print("\n=== Final ===") | |
| print(result["final_answer"]) | |
| try: | |
| txt = (here / "out" / "demo_report.md").read_text(encoding="utf-8") | |
| head = "\n".join(txt.splitlines()[:22]) | |
| print("\n=== Report Preview (Top) ===") | |
| print(head) | |
| except Exception: | |
| pass | |
| return 0 | |
| def cmd_run(goal: str) -> int: | |
| here = Path(__file__).resolve().parent | |
| out_report = str(here / "out" / "demo_report.md") | |
| kb_dir = str(here / "kb") | |
| result = run_agent(goal=goal, kb_dir=kb_dir, out_report=out_report, max_rounds=3) | |
| _print_round_logs(result["round_logs"]) | |
| print("\n=== Final ===") | |
| print(result["final_answer"]) | |
| return 0 | |
| def main(argv: List[str]) -> int: | |
| p = argparse.ArgumentParser(prog="agent.py") | |
| sub = p.add_subparsers(dest="cmd", required=True) | |
| sub.add_parser("demo") | |
| p_run = sub.add_parser("run") | |
| p_run.add_argument("--goal", required=True) | |
| args = p.parse_args(argv) | |
| if args.cmd == "demo": | |
| return cmd_demo() | |
| if args.cmd == "run": | |
| return cmd_run(goal=args.goal) | |
| return 2 | |
| if __name__ == "__main__": | |
| raise SystemExit(main(sys.argv[1:])) | |