feat: 四层架构数据分析 Agent

- Layer 1 Planner: 意图规划,将问题转为结构化分析计划
- Layer 2 Explorer: 自适应探索循环,多轮迭代动态生成 SQL
- Layer 3 InsightEngine: 异常检测 + 主动洞察
- Layer 4 ContextManager: 多轮对话上下文记忆

安全设计:AI 只看 Schema + 聚合结果,不接触原始数据。
支持任意 OpenAI 兼容 API(OpenAI / Ollama / DeepSeek / vLLM)
This commit is contained in:
OpenClaw Agent
2026-03-19 12:21:04 +08:00
commit 96927a789d
14 changed files with 1683 additions and 0 deletions

110
reporter.py Normal file
View File

@@ -0,0 +1,110 @@
"""
报告生成器
将分析计划 + 探索结果 + 洞察,综合为一份可读报告。
"""
import json
from typing import Any
import openai
from config import LLM_CONFIG
from explorer import ExplorationStep
from insights import Insight
REPORT_PROMPT = """你是一个数据分析报告撰写专家。基于以下信息撰写报告。
## 用户问题
{question}
## 分析计划
{plan}
## 探索过程
{exploration}
## 主动洞察
{insights_text}
## 撰写要求
1. **开头**:一句话总结核心结论(先给答案)
2. **核心发现**:按重要性排列,每个发现带具体数字
3. **深入洞察**:异常、趋势、关联(从洞察中提取)
4. **建议**:基于数据的行动建议
5. **审计**:末尾附上所有执行的 SQL
使用 Markdown中文撰写。
语气:专业但不枯燥,像一个聪明的分析师在做简报。"""
class ReportGenerator:
"""报告生成器"""
def __init__(self):
self.client = openai.OpenAI(
api_key=LLM_CONFIG["api_key"],
base_url=LLM_CONFIG["base_url"],
)
self.model = LLM_CONFIG["model"]
def generate(
self,
question: str,
plan: dict,
steps: list[ExplorationStep],
insights: list[Insight],
) -> str:
"""生成分析报告"""
# 构建探索过程文本
exploration = self._build_exploration(steps)
# 构建洞察文本
insights_text = self._build_insights(insights)
prompt = REPORT_PROMPT.format(
question=question,
plan=json.dumps(plan, ensure_ascii=False, indent=2),
exploration=exploration,
insights_text=insights_text,
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是专业的数据分析师,撰写清晰、有洞察力的分析报告。"},
{"role": "user", "content": prompt},
],
temperature=0.3,
max_tokens=4096,
)
return response.choices[0].message.content
def _build_exploration(self, steps: list[ExplorationStep]) -> str:
parts = []
for step in steps:
if step.action == "done":
parts.append(f"### 探索结束\n{step.reasoning}")
continue
if step.success:
parts.append(
f"### 第 {step.round_num} 轮:{step.purpose}\n"
f"思考: {step.reasoning}\n"
f"SQL: `{step.sql}`\n"
f"结果 ({step.row_count} 行):\n"
f"列: {step.columns}\n"
f"数据: {json.dumps(step.rows, ensure_ascii=False)}"
)
else:
parts.append(
f"### 第 {step.round_num} 轮:{step.purpose}\n"
f"SQL: `{step.sql}`\n"
f"执行失败: {step.error}"
)
return "\n\n".join(parts) if parts else "无探索步骤"
def _build_insights(self, insights: list[Insight]) -> str:
if not insights:
return "未检测到异常。"
return "\n".join(str(i) for i in insights)