| 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 |
|
|
|
|
| @dataclass |
| class ToolCall: |
| name: str |
| args: Dict[str, Any] |
|
|
|
|
| @dataclass |
| 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:])) |
|
|
