企业官网建设流程全解析

系列第4篇：Python+Go构建企业级AI Agent实战指南（4/13）

标签：DeerFlow | Skill开发 | 数据分析 | Python | 源码解析

一、开篇：Skill是Agent的"超能力"

如果把AI Agent比作一个职场新人，那么**Skill（技能）**就是他的专业能力：

• 会Excel → 数据处理Skill
• 会Python → 编程Skill
• 会写报告 → 文案Skill

DeerFlow的Skill机制，让Agent的能力扩展变得像安装APP一样简单。本文将深入Skill.md开发，带你构建一个生产级的数据分析Agent。

二、Skill.md文件详解

2.1 Skill.md结构

# Skill.md 标准结构 ## 基本信息 name: Skill名称 description: Skill功能描述 version: 版本号 author: 作者 ## 能力列表 (capabilities) 定义Agent能做什么 ## 工作流程 (workflow) 定义Agent如何执行任务 ## 约束条件 (constraints) 定义边界和限制 ## 示例 (examples) 提供使用示例

2.2 完整Skill.md示例

# sales_analyzer/Skill.md ## 基本信息 name: SalesAnalyzer description: 专业的销售数据分析智能体，提供销售趋势分析、预测和洞察 version: 2.0.0 author: AI Team ## 能力列表 capabilities: - name: load_sales_data description: 加载销售数据，支持CSV、Excel格式 parameters: - name: file_path type: string required: true description: 数据文件路径 - name: date_column type: string required: false default: "date" description: 日期列名称 returns: type: dataframe description: 加载的数据框 - name: analyze_trends description: 分析销售趋势，包括同比、环比、移动平均线 parameters: - name: df type: dataframe required: true - name: metric type: string required: true description: 分析指标，如sales、revenue - name: period type: string enum: [daily, weekly, monthly] default: "monthly" returns: type: dict description: 趋势分析结果 - name: forecast_sales description: 使用 Prophet 模型预测未来销售 parameters: - name: df type: dataframe required: true - name: periods type: integer required: false default: 30 description: 预测天数 returns: type: dataframe description: 预测结果 - name: generate_insights description: 基于分析结果生成业务洞察 parameters: - name: analysis_results type: dict required: true returns: type: list description: 洞察列表 ## 工作流程 workflow: 1. 接收用户的数据文件路径和分析需求 2. 验证文件格式和完整性 3. 加载数据并执行基础清洗 4. 根据需求选择分析维度（趋势/预测/对比） 5. 执行分析并生成可视化 6. 提取关键洞察和业务建议 7. 整合输出完整的分析报告 ## 约束条件 constraints: - 数据文件大小不超过500MB - 支持格式：CSV、Excel (.xlsx, .xls) - 必须包含日期列 - 数值列不能包含非数字字符 - 预测功能需要至少3个月历史数据 ## 示例 examples: - input: "分析sales_2024.csv的销售趋势" output: | 📊 销售趋势分析报告 1. 整体趋势：2024年销售额同比增长23% 2. 季节性：Q4为销售旺季，占全年35% 3. 异常点：3月份销售额下降15%，建议关注 [附：趋势图、预测图] - input: "预测下个月的销售额" output: | 🔮 销售预测 基于历史数据，预测下月销售额： - 预测值：$125,000 - 置信区间：$118,000 - $132,000 - 环比预期：+8%

三、Skill的Python实现

3.1 项目结构

sales_analyzer/ ├── Skill.md # 技能定义文件 ├── __init__.py ├── agent.py # Agent主类 ├── tools/ │ ├── __init__.py │ ├── data_loader.py # 数据加载工具 │ ├── analyzer.py # 分析工具 │ ├── forecaster.py # 预测工具 │ └── visualizer.py # 可视化工具 ├── prompts/ │ └── analysis_prompt.txt └── tests/ └── test_analyzer.py

3.2 完整代码实现

# sales_analyzer/agent.py import pandas as pd import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Optional import matplotlib.pyplot as plt import seaborn as sns from prophet import Prophet import warnings warnings.filterwarnings('ignore') from deerflow import Agent, Tool from deerflow.types import ToolResult plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS'] plt.rcParams['axes.unicode_minus'] = False class SalesAnalyzerAgent(Agent): """ 销售数据分析智能体 提供销售趋势分析、预测和洞察生成功能 """ def __init__(self, config: Optional[Dict] = None): super().__init__() self.config = config or {} self.data = None self.setup_tools() def setup_tools(self): """注册所有工具""" @Tool( name="load_sales_data", description="加载销售数据文件" ) def load_data(file_path: str, date_column: str = "date") -> ToolResult: """加载并验证销售数据""" try: # 根据扩展名选择读取方式 if file_path.endswith('.csv'): df = pd.read_csv(file_path) elif file_path.endswith(('.xlsx', '.xls')): df = pd.read_excel(file_path) else: return ToolResult.error("不支持的文件格式，请使用CSV或Excel") # 验证日期列 if date_column not in df.columns: return ToolResult.error(f"未找到日期列: {date_column}") # 转换日期格式 df[date_column] = pd.to_datetime(df[date_column]) df = df.sort_values(date_column) # 存储数据 self.data = df # 返回数据摘要 summary = { "rows": len(df), "columns": list(df.columns), "date_range": f"{df[date_column].min()} 至 {df[date_column].max()}", "numeric_columns": df.select_dtypes(include=[np.number]).columns.tolist() } return ToolResult.success(summary) except Exception as e: return ToolResult.error(f"数据加载失败: {str(e)}") @Tool( name="analyze_trends", description="分析销售趋势" ) def analyze_trends( metric: str = "sales", period: str = "monthly" ) -> ToolResult: """分析销售趋势指标""" if self.data is None: return ToolResult.error("请先加载数据") if metric not in self.data.columns: return ToolResult.error(f"未找到指标列: {metric}") try: df = self.data.copy() # 按周期聚合 if period == "daily": df_grouped = df.groupby(df['date'].dt.date)[metric].sum() elif period == "weekly": df_grouped = df.groupby(df['date'].dt.to_period('W'))[metric].sum() else: # monthly df_grouped = df.groupby(df['date'].dt.to_period('M'))[metric].sum() # 计算关键指标 total = df_grouped.sum() mean = df_grouped.mean() growth_rate = (df_grouped.iloc[-1] - df_grouped.iloc[0]) / df_grouped.iloc[0] * 100 # 计算移动平均线 ma_7 = df_grouped.rolling(window=7, min_periods=1).mean() ma_30 = df_grouped.rolling(window=30, min_periods=1).mean() # 找出峰值和谷值 peak_idx = df_grouped.idxmax() valley_idx = df_grouped.idxmin() result = { "period": period, "metric": metric, "total": round(total, 2), "average": round(mean, 2), "growth_rate": round(growth_rate, 2), "peak": {"period": str(peak_idx), "value": round(df_grouped.max(), 2)}, "valley": {"period": str(valley_idx), "value": round(df_grouped.min(), 2)}, "trend_data": df_grouped.to_dict() } return ToolResult.success(result) except Exception as e: return ToolResult.error(f"趋势分析失败: {str(e)}") @Tool( name="forecast_sales", description="预测未来销售趋势" ) def forecast_sales( metric: str = "sales", periods: int = 30 ) -> ToolResult: """使用Prophet模型进行销售预测""" if self.data is None: return ToolResult.error("请先加载数据") try: # 准备Prophet格式数据 df_prophet = self.data.rename(columns={ 'date': 'ds', metric: 'y' })[['ds', 'y']] # 创建并训练模型 model = Prophet( yearly_seasonality=True, weekly_seasonality=True, daily_seasonality=False ) model.fit(df_prophet) # 生成预测 future = model.make_future_dataframe(periods=periods) forecast = model.predict(future) # 提取预测结果 forecast_future = forecast[forecast['ds'] > df_prophet['ds'].max()] result = { "forecast_periods": periods, "predictions": [ { "date": row['ds'].strftime('%Y-%m-%d'), "value": round(row['yhat'], 2), "lower": round(row['yhat_lower'], 2), "upper": round(row['yhat_upper'], 2) } for _, row in forecast_future.head(10).iterrows() ], "trend_direction": "up" if forecast['trend'].iloc[-1] > forecast['trend'].iloc[-30] else "down" } return ToolResult.success(result) except Exception as e: return ToolResult.error(f"预测失败: {str(e)}") @Tool( name="generate_report", description="生成完整的分析报告" ) def generate_report( trend_analysis: Dict, forecast: Optional[Dict] = None ) -> ToolResult: """生成Markdown格式的分析报告""" report = f"""# 📊 销售数据分析报告 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M')} ## 1. 数据概览 - **分析指标**: {trend_analysis['metric']} - **时间周期**: {trend_analysis['period']} - **总销售额**: ¥{trend_analysis['total']:,.2f} - **平均销售额**: ¥{trend_analysis['average']:,.2f} ## 2. 趋势分析 ### 2.1 整体趋势 - **增长率**: {trend_analysis['growth_rate']:+.2f}% - **趋势方向**: {'📈 上升' if trend_analysis['growth_rate'] > 0 else '📉 下降'} ### 2.2 关键节点 - **销售峰值**: ¥{trend_analysis['peak']['value']:,.2f} ({trend_analysis['peak']['period']}) - **销售谷值**: ¥{trend_analysis['valley']['value']:,.2f} ({trend_analysis['valley']['period']}) """ if forecast: report += f""" ## 3. 销售预测 ### 3.1 未来{trend_analysis['period']}预测 - **趋势方向**: {'📈 预期上升' if forecast['trend_direction'] == 'up' else '📉 预期下降'} - **预测周期**: {forecast['forecast_periods']} 天 ### 3.2 近期预测值 | 日期 | 预测值 | 置信区间下限 | 置信区间上限 | |------|--------|-------------|-------------| """ for pred in forecast['predictions'][:5]: report += f"| {pred['date']} | ¥{pred['value']:,.2f} | ¥{pred['lower']:,.2f} | ¥{pred['upper']:,.2f} |\n" report += """ ## 4. 业务建议 基于以上分析，建议关注以下方面： 1. **趋势把握**: 根据增长趋势调整库存和人员配置 2. **季节性**: 关注销售峰值周期，提前做好营销准备 3. **异常监控**: 建立预警机制，及时发现销售异常 --- *报告由 SalesAnalyzer Agent 自动生成* """ return ToolResult.success({"report": report}) # 注册所有工具 self.register_tool(load_data) self.register_tool(analyze_trends) self.register_tool(forecast_sales) self.register_tool(generate_report) async def run_analysis(self, file_path: str, metric: str = "sales") -> str: """执行完整的分析流程""" print("🚀 开始销售数据分析...") # 1. 加载数据 print("📖 加载数据...") load_result = await self.call_tool("load_sales_data", file_path=file_path) if not load_result.success: return f"❌ 数据加载失败: {load_result.data}" print(f"✅ 数据加载成功: {load_result.data}") # 2. 趋势分析 print("📊 分析趋势...") trend_result = await self.call_tool("analyze_trends", metric=metric) if not trend_result.success: return f"❌ 趋势分析失败: {trend_result.data}" print("✅ 趋势分析完成") # 3. 销售预测 print("🔮 预测未来趋势...") forecast_result = await self.call_tool("forecast_sales", metric=metric, periods=30) if not forecast_result.success: print(f"⚠️ 预测失败: {forecast_result.data}") forecast_result = None else: print("✅ 预测完成") # 4. 生成报告 print("📝 生成报告...") report_result = await self.call_tool( "generate_report", trend_analysis=trend_result.data, forecast=forecast_result.data if forecast_result else None ) if report_result.success: # 保存报告 output_path = f"sales_report_{datetime.now().strftime('%Y%m%d_%H%M')}.md" with open(output_path, 'w', encoding='utf-8') as f: f.write(report_result.data['report']) print(f"✅ 报告已保存: {output_path}") return report_result.data['report'] else: return f"❌ 报告生成失败: {report_result.data}" # 使用示例 async def main(): # 创建示例数据 np.random.seed(42) dates = pd.date_range('2024-01-01', periods=365, freq='D') base_sales = 1000 trend = np.linspace(0, 500, 365) seasonal = 200 * np.sin(2 * np.pi * np.arange(365) / 365.25) noise = np.random.normal(0, 100, 365) sales = base_sales + trend + seasonal + noise sample_df = pd.DataFrame({ 'date': dates, 'sales': sales.astype(int), 'orders': (sales / 50).astype(int) }) sample_df.to_csv('sample_sales.csv', index=False) print("✅ 示例数据已创建: sample_sales.csv") # 创建Agent并执行分析 agent = SalesAnalyzerAgent() report = await agent.run_analysis('sample_sales.csv', metric='sales') print("\n" + "="*60) print("📄 报告预览:") print("="*60) print(report[:1000]) print("...") if __name__ == "__main__": import asyncio asyncio.run(main())

四、测试与调试

4.1 单元测试

# tests/test_analyzer.py import pytest import pandas as pd from sales_analyzer.agent import SalesAnalyzerAgent @pytest.fixture def sample_data(): """创建测试数据""" dates = pd.date_range('2024-01-01', periods=100, freq='D') return pd.DataFrame({ 'date': dates, 'sales': range(100), 'revenue': [x * 10 for x in range(100)] }) @pytest.fixture def agent(): return SalesAnalyzerAgent() @pytest.mark.asyncio async def test_load_data(agent, tmp_path): """测试数据加载""" # 创建临时CSV csv_path = tmp_path / "test.csv" sample_data().to_csv(csv_path, index=False) result = await agent.call_tool("load_sales_data", file_path=str(csv_path)) assert result.success assert result.data['rows'] == 100 @pytest.mark.asyncio async def test_analyze_trends(agent, tmp_path): """测试趋势分析""" # 先加载数据 csv_path = tmp_path / "test.csv" sample_data().to_csv(csv_path, index=False) await agent.call_tool("load_sales_data", file_path=str(csv_path)) # 测试分析 result = await agent.call_tool("analyze_trends", metric="sales") assert result.success assert 'growth_rate' in result.data

五、结语

通过本文，你已经掌握了：

1. ✅ Skill.md文件的完整结构和编写规范
2. ✅ DeerFlow Agent的Python实现
3. ✅ 工具注册和调用机制
4. ✅ 完整的数据分析Agent开发流程

下一篇，我们将探讨Python与国产大模型的深度集成。

系列文章导航：← 3. Python构建AI多智能体系统实战 4.Python技能开发实战（本文） 5. Python与国产大模型的深度集成 →

本文首发于CSDN，转载请注明出处。

标签：DeerFlow | Skill开发 | 数据分析 | Python | 源码解析