Manus AI 通用 Agent 平台完全教程
适用人群：AI 开发者、产品经理、自动化工程师 关键词：Manus AI, 通用 Agent, 任务规划, 工具链, 浏览器自动化, 数据分析 预计学习时间：8-12 小时
目录
- Manus AI 概述与核心概念
- Manus Agent 架构与设计哲学
- 任务分解与规划系统
- 工具链集成基础
- 浏览器操控实战
- 文件处理与管理
- 代码执行环境
- 数据分析与可视化
- 多模态理解能力
- 自定义 Agent 开发
- 企业级工作流设计
- 实战项目一：自动化市场研究报告生成
- 实战项目二：竞品分析系统
- 常见问题与解决方案
- 进阶资源与社区
1. Manus AI 概述与核心概念
1.1 什么是 Manus AI
Manus AI 是一个通用型 AI Agent 平台，其核心理念是让 AI 不再局限于"对话"，而是能够像人类一样使用工具、操作软件、完成复杂任务。与传统的聊天机器人不同，Manus Agent 可以：
- 自主浏览网页并提取信息
- 操作文件系统（创建、编辑、管理文件）
- 执行代码并分析结果
- 调用外部 API 完成特定任务
- 处理图片、文档等多种格式
- 按照用户意图拆解并逐步完成复杂项目
1.2 Agent 与传统 AI 的区别
| 维度 | 传统 AI 对话 | Manus Agent |
|---|---|---|
| 交互模式 | 一问一答 | 自主规划执行 |
| 工具使用 | 无 | 浏览器/代码/文件/API |
| 任务复杂度 | 单轮简单 | 多步骤复杂项目 |
| 输出形式 | 纯文本 | 文件/报告/代码/部署 |
| 持续性 | 无状态 | 有记忆和上下文 |
1.3 快速体验
使用 Manus 的第一步是理解其工作流程。以下是一个典型的交互示例：
用户指令：帮我调研一下2024年中国新能源汽车市场的前三名品牌，
整理成一份包含销量数据、市场占比、核心优势的报告。
Manus Agent 自动执行：
1. [规划] → 分解为搜索、数据提取、分析、报告生成四个子任务
2. [搜索] → 打开浏览器，搜索多个数据源
3. [提取] → 从网页中提取结构化数据
4. [分析] → 用 Python 进行数据清洗和统计
5. [生成] → 输出格式化的 Markdown/PDF 报告
2. Manus Agent 架构与设计哲学
2.1 核心架构
Manus Agent 采用感知-规划-执行-反馈的循环架构：
┌─────────────────────────────────────────┐
│ 用户指令输入 │
└─────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────┐
│ 感知层 (Perception) │
│ · 解析用户意图 │
│ · 识别任务类型 │
│ · 提取关键实体和约束 │
└─────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────┐
│ 规划层 (Planning) │
│ · 任务分解 (Task Decomposition) │
│ · 工具选择 (Tool Selection) │
│ · 执行顺序编排 (Orchestration) │
└─────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────┐
│ 执行层 (Execution) │
│ · 调用浏览器/代码/文件等工具 │
│ · 处理中间结果 │
│ · 异常捕获与重试 │
└─────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────┐
│ 反馈层 (Feedback) │
│ · 结果验证 │
│ · 质量评估 │
│ · 必要时重新规划 │
└─────────────┬───────────────────────────┘
▼
┌─────────────────────────────────────────┐
│ 输出交付 │
└─────────────────────────────────────────┘
2.2 设计哲学
哲学一：工具即能力
Agent 的能力上限取决于它能使用的工具。Manus 将浏览器、代码解释器、文件系统等封装为标准化工具，Agent 通过学习如何组合这些工具来完成任意复杂任务。
哲学二：规划优于蛮力
面对复杂任务，盲目执行往往导致低效和错误。Manus 强调先规划后执行，通过任务分解将大问题拆解为可管理的小步骤。
哲学三：容错与自愈
真实世界的任务执行充满不确定性——网页加载失败、API 超时、数据格式异常。Manus 内置重试机制和备选方案，确保任务在遇到障碍时能自适应调整。
2.3 Agent 状态管理
# Agent 状态的数据模型示例
class AgentState:
def __init__(self, task_id: str):
self.task_id = task_id
self.status = "pending" # pending, planning, executing, completed, failed
self.plan = [] # 任务计划列表
self.current_step = 0 # 当前执行步骤
self.context = {} # 上下文信息（中间结果）
self.tools_used = [] # 已使用的工具记录
self.errors = [] # 错误日志
self.artifacts = [] # 产出物列表
def advance(self):
"""推进到下一步"""
self.current_step += 1
if self.current_step >= len(self.plan):
self.status = "completed"
def add_artifact(self, artifact_type: str, path: str, description: str):
"""记录产出物"""
self.artifacts.append({
"type": artifact_type,
"path": path,
"description": description,
"timestamp": datetime.now().isoformat()
})
def get_current_task(self) -> dict:
"""获取当前任务描述"""
if self.current_step < len(self.plan):
return self.plan[self.current_step]
return None
3. 任务分解与规划系统
3.1 任务分解的核心原则
任务分解是 Agent 能力的核心。一个好的分解策略应该满足：
- 可执行性：每个子任务对应一个具体的工具调用
- 独立性：子任务之间的依赖关系清晰且最小化
- 可验证性：每个子任务的结果可以被验证
- 粒度适中：不过于粗略（难以执行），也不过于细碎（低效）
3.2 任务分解模式
模式一：顺序分解
适用于有明确先后依赖的任务链。
# 顺序分解示例
task_plan = [
{
"step": 1,
"action": "search",
"description": "搜索新能源汽车2024年销量数据",
"tool": "browser",
"params": {"query": "2024年中国新能源汽车品牌销量排名"},
"depends_on": None
},
{
"step": 2,
"action": "extract",
"description": "从搜索结果中提取TOP3品牌数据",
"tool": "browser",
"params": {"selectors": [".brand-name", ".sales-data"]},
"depends_on": 1
},
{
"step": 3,
"action": "analyze",
"description": "计算市场份额和增长率",
"tool": "python",
"params": {"script": "analyze_market.py"},
"depends_on": 2
},
{
"step": 4,
"action": "generate",
"description": "生成最终报告",
"tool": "file",
"params": {"template": "market_report.md"},
"depends_on": 3
}
]
模式二：并行分解
适用于相互独立的子任务，可以同时执行以提高效率。
# 并行分解示例
parallel_tasks = [
{
"group": "data_collection",
"parallel": True,
"tasks": [
{"action": "search", "query": "比亚迪2024年销量"},
{"action": "search", "query": "特斯拉中国2024年销量"},
{"action": "search", "query": "蔚来2024年销量"}
]
},
{
"group": "analysis",
"parallel": False, # 依赖数据收集完成
"depends_on": "data_collection",
"tasks": [
{"action": "analyze", "type": "comparison"}
]
}
]
模式三：递归分解
适用于不确定深度的复杂任务，需要逐层拆解。
def decompose_task(task: dict, max_depth: int = 3) -> list:
"""递归分解任务"""
if max_depth <= 0 or is_atomic(task):
return [task]
subtasks = generate_subtasks(task)
result = []
for subtask in subtasks:
result.extend(decompose_task(subtask, max_depth - 1))
return result
def is_atomic(task: dict) -> bool:
"""判断任务是否已是最小可执行单元"""
atomic_actions = ["search", "click", "type", "read_file", "write_file",
"execute_code", "api_call"]
return task.get("action") in atomic_actions
def generate_subtasks(task: dict) -> list:
"""根据任务描述生成子任务"""
prompt = f"""
将以下任务分解为更小的可执行子任务：
任务：{task['description']}
要求：每个子任务必须对应一个具体的操作（搜索、点击、读写文件、执行代码等）
"""
# 调用 LLM 生成子任务分解
return llm_plan(prompt)
3.3 动态规划与重规划
在实际执行中，经常需要根据中间结果调整计划：
class DynamicPlanner:
def __init__(self, original_plan: list):
self.plan = original_plan
self.completed = []
self.context = {}
def replan(self, current_result: dict, error: str = None) -> list:
"""根据当前结果动态调整计划"""
if error:
# 错误恢复：生成替代方案
alternative = self._generate_alternative(
failed_step=self.plan[len(self.completed)],
error=error,
context=self.context
)
self.plan = self.completed + alternative + self.plan[len(self.completed)+1:]
else:
# 根据结果优化后续步骤
self.context.update(current_result)
optimized = self._optimize_remaining(
remaining=self.plan[len(self.completed):],
context=self.context
)
self.plan = self.completed + optimized
return self.plan
def _generate_alternative(self, failed_step, error, context):
"""生成失败步骤的替代方案"""
prompt = f"""
以下步骤执行失败，请生成替代方案：
原步骤：{failed_step}
错误信息：{error}
已有上下文：{context}
"""
return llm_plan(prompt)
4. 工具链集成基础
4.1 工具注册机制
Manus 通过统一的工具注册接口管理所有可用工具：
from abc import ABC, abstractmethod
from typing import Any, Dict
class Tool(ABC):
"""工具基类"""
@property
@abstractmethod
def name(self) -> str:
"""工具名称"""
pass
@property
@abstractmethod
def description(self) -> str:
"""工具描述，供 Agent 理解工具用途"""
pass
@property
@abstractmethod
def parameters(self) -> Dict[str, Any]:
"""工具参数 schema"""
pass
@abstractmethod
def execute(self, **kwargs) -> Dict[str, Any]:
"""执行工具操作"""
pass
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self._tools: Dict[str, Tool] = {}
def register(self, tool: Tool):
"""注册工具"""
self._tools[tool.name] = tool
print(f"已注册工具: {tool.name}")
def get(self, name: str) -> Tool:
"""获取工具"""
if name not in self._tools:
raise ValueError(f"工具 '{name}' 未注册")
return self._tools[name]
def list_tools(self) -> list:
"""列出所有可用工具"""
return [
{
"name": t.name,
"description": t.description,
"parameters": t.parameters
}
for t in self._tools.values()
]
def execute(self, tool_name: str, **kwargs) -> Dict[str, Any]:
"""执行指定工具"""
tool = self.get(tool_name)
return tool.execute(**kwargs)
4.2 内置工具集
Manus 提供以下核心内置工具：
# 浏览器工具
class BrowserTool(Tool):
name = "browser"
description = "网页浏览、信息提取、表单填写"
parameters = {
"action": {"type": "string", "enum": ["navigate", "click", "type", "screenshot", "extract"]},
"target": {"type": "string"},
"value": {"type": "string"}
}
# 代码执行工具
class CodeExecutionTool(Tool):
name = "code_execution"
description = "执行 Python/JavaScript 代码"
parameters = {
"language": {"type": "string", "enum": ["python", "javascript", "bash"]},
"code": {"type": "string"}
}
# 文件管理工具
class FileTool(Tool):
name = "file_manager"
description = "文件读写和管理"
parameters = {
"action": {"type": "string", "enum": ["read", "write", "list", "delete", "move"]},
"path": {"type": "string"},
"content": {"type": "string"}
}
# 搜索工具
class SearchTool(Tool):
name = "web_search"
description = "网络搜索"
parameters = {
"query": {"type": "string"},
"num_results": {"type": "integer", "default": 5}
}
# API 调用工具
class APITool(Tool):
name = "api_call"
description = "调用外部 API"
parameters = {
"method": {"type": "string", "enum": ["GET", "POST", "PUT", "DELETE"]},
"url": {"type": "string"},
"headers": {"type": "object"},
"body": {"type": "object"}
}
4.3 自定义工具开发
开发者可以创建自定义工具来扩展 Agent 能力：
class DatabaseQueryTool(Tool):
"""数据库查询自定义工具"""
name = "db_query"
description = "执行 SQL 查询，支持 MySQL/PostgreSQL/SQLite"
parameters = {
"connection_string": {"type": "string", "description": "数据库连接字符串"},
"query": {"type": "string", "description": "SQL 查询语句"},
"params": {"type": "object", "description": "查询参数"}
}
def execute(self, connection_string: str, query: str, params: dict = None) -> dict:
import sqlite3
try:
conn = sqlite3.connect(connection_string)
cursor = conn.cursor()
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
if query.strip().upper().startswith("SELECT"):
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
return {
"success": True,
"columns": columns,
"data": [dict(zip(columns, row)) for row in rows],
"row_count": len(rows)
}
else:
conn.commit()
return {
"success": True,
"affected_rows": cursor.rowcount
}
except Exception as e:
return {"success": False, "error": str(e)}
finally:
conn.close()
# 注册自定义工具
registry = ToolRegistry()
registry.register(DatabaseQueryTool())
# Agent 使用示例
result = registry.execute(
"db_query",
connection_string="data/analytics.db",
query="SELECT brand, sales FROM market_data WHERE year = ?",
params={"year": 2024}
)
5. 浏览器操控实战
5.1 浏览器自动化基础
Manus 的浏览器操控能力基于 Playwright/Puppeteer 等底层框架，提供了更高层的抽象接口：
class BrowserAgent:
"""浏览器操控 Agent"""
def __init__(self):
self.page = None
self.history = []
async def navigate(self, url: str) -> dict:
"""导航到指定 URL"""
await self.page.goto(url, wait_until="networkidle")
self.history.append({"action": "navigate", "url": url})
title = await self.page.title()
return {"status": "success", "title": title, "url": url}
async def search_and_extract(self, query: str, selectors: dict) -> dict:
"""搜索并提取信息"""
# 导航到搜索引擎
await self.navigate(f"https://www.bing.com/search?q={query}")
# 等待搜索结果加载
await self.page.wait_for_selector(".b_algo")
# 提取搜索结果
results = await self.page.query_selector_all(".b_algo")
extracted = []
for result in results[:5]: # 取前5条
title = await result.query_selector("h2")
link = await result.query_selector("a")
snippet = await result.query_selector(".b_caption p")
extracted.append({
"title": await title.inner_text() if title else "",
"url": await link.get_attribute("href") if link else "",
"snippet": await snippet.inner_text() if snippet else ""
})
return {"query": query, "results": extracted}
async def fill_form(self, form_config: list) -> dict:
"""自动填写表单"""
for field in form_config:
selector = field["selector"]
value = field["value"]
field_type = field.get("type", "text")
if field_type == "text":
await self.page.fill(selector, value)
elif field_type == "select":
await self.page.select_option(selector, value)
elif field_type == "checkbox":
if value:
await self.page.check(selector)
elif field_type == "click":
await self.page.click(selector)
return {"status": "form_filled", "fields": len(form_config)}
async def screenshot(self, path: str = None) -> str:
"""截取当前页面截图"""
if path is None:
path = f"screenshots/{int(time.time())}.png"
await self.page.screenshot(path=path, full_page=True)
return path
async def extract_table_data(self, table_selector: str) -> list:
"""提取网页表格数据为结构化格式"""
table = await self.page.query_selector(table_selector)
if not table:
return []
rows = await table.query_selector_all("tr")
data = []
headers = []
for i, row in enumerate(rows):
cells = await row.query_selector_all("th, td")
cell_texts = [await cell.inner_text() for cell in cells]
if i == 0:
# 第一行作为表头
if await rows[0].query_selector("th"):
headers = cell_texts
continue
if headers:
data.append(dict(zip(headers, cell_texts)))
else:
data.append(cell_texts)
return data
5.2 实战：自动采集商品信息
async def scrape_product_info(urls: list) -> list:
"""批量采集商品信息"""
browser_agent = BrowserAgent()
all_products = []
for url in urls:
try:
await browser_agent.navigate(url)
product = {
"url": url,
"name": await browser_agent.page.inner_text(".product-title"),
"price": await browser_agent.page.inner_text(".product-price"),
"rating": await browser_agent.page.inner_text(".rating-score"),
"reviews": await browser_agent.page.inner_text(".review-count"),
"description": await browser_agent.page.inner_text(".product-desc"),
"images": []
}
# 提取商品图片
images = await browser_agent.page.query_selector_all(".product-image img")
for img in images:
src = await img.get_attribute("src")
if src:
product["images"].append(src)
all_products.append(product)
print(f"✓ 已采集: {product['name']}")
except Exception as e:
print(f"✗ 采集失败 {url}: {str(e)}")
all_products.append({"url": url, "error": str(e)})
return all_products
6. 文件处理与管理
6.1 文件系统操作
import os
import json
import csv
from pathlib import Path
from datetime import datetime
class FileAgent:
"""文件管理 Agent"""
def __init__(self, workspace: str = "./workspace"):
self.workspace = Path(workspace)
self.workspace.mkdir(parents=True, exist_ok=True)
def read_file(self, path: str, encoding: str = "utf-8") -> str:
"""读取文件内容"""
full_path = self.workspace / path
if not full_path.exists():
raise FileNotFoundError(f"文件不存在: {full_path}")
return full_path.read_text(encoding=encoding)
def write_file(self, path: str, content: str, encoding: str = "utf-8") -> str:
"""写入文件"""
full_path = self.workspace / path
full_path.parent.mkdir(parents=True, exist_ok=True)
full_path.write_text(content, encoding=encoding)
return str(full_path)
def list_files(self, pattern: str = "**/*", recursive: bool = True) -> list:
"""列出文件"""
if recursive:
files = list(self.workspace.glob(pattern))
else:
files = list(self.workspace.glob(pattern))
return [
{
"path": str(f.relative_to(self.workspace)),
"size": f.stat().st_size,
"modified": datetime.fromtimestamp(f.stat().st_mtime).isoformat(),
"is_dir": f.is_dir()
}
for f in files
]
def create_report(self, title: str, sections: list, output_format: str = "md") -> str:
"""创建格式化报告"""
if output_format == "md":
content = f"# {title}\n\n"
content += f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"
content += "---\n\n"
for section in sections:
content += f"## {section['heading']}\n\n"
content += f"{section['content']}\n\n"
if "data" in section:
content += self._format_table(section["data"])
content += "\n\n"
return self.write_file(f"reports/{title}.{output_format}", content)
def _format_table(self, data: list) -> str:
"""将数据格式化为 Markdown 表格"""
if not data:
return ""
headers = list(data[0].keys())
table = "| " + " | ".join(headers) + " |\n"
table += "| " + " | ".join(["---"] * len(headers)) + " |\n"
for row in data:
table += "| " + " | ".join([str(row.get(h, "")) for h in headers]) + " |\n"
return table
6.2 多格式文件处理
class MultiFormatProcessor:
"""多格式文件处理器"""
@staticmethod
def process_csv(file_path: str) -> dict:
"""处理 CSV 文件"""
import pandas as pd
df = pd.read_csv(file_path)
return {
"rows": len(df),
"columns": list(df.columns),
"preview": df.head(5).to_dict(orient="records"),
"statistics": df.describe().to_dict()
}
@staticmethod
def process_json(file_path: str) -> dict:
"""处理 JSON 文件"""
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
def count_depth(obj, depth=0):
if isinstance(obj, dict):
return max(count_depth(v, depth+1) for v in obj.values()) if obj else depth
elif isinstance(obj, list):
return max(count_depth(v, depth+1) for v in obj) if obj else depth
return depth
return {
"type": type(data).__name__,
"size": len(data) if isinstance(data, (list, dict)) else 1,
"depth": count_depth(data),
"preview": str(data)[:500]
}
@staticmethod
def process_markdown(file_path: str) -> dict:
"""处理 Markdown 文件"""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
import re
headings = re.findall(r'^(#{1,6})\s+(.+)$', content, re.MULTILINE)
code_blocks = re.findall(r'```(\w*)\n(.*?)```', content, re.DOTALL)
links = re.findall(r'\[([^\]]+)\]\(([^)]+)\)', content)
return {
"headings": [(len(h[0]), h[1]) for h in headings],
"code_blocks": len(code_blocks),
"code_languages": list(set(b[0] for b in code_blocks if b[0])),
"links": len(links),
"word_count": len(content.split()),
"char_count": len(content)
}
7. 代码执行环境
7.1 安全的代码沙箱
import subprocess
import tempfile
import resource
from typing import Optional
class CodeSandbox:
"""安全代码执行沙箱"""
def __init__(self, timeout: int = 30, memory_limit_mb: int = 256):
self.timeout = timeout
self.memory_limit_mb = memory_limit_mb
def execute_python(self, code: str, input_data: str = None) -> dict:
"""执行 Python 代码"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code)
script_path = f.name
try:
result = subprocess.run(
['python3', script_path],
capture_output=True,
text=True,
timeout=self.timeout,
input=input_data
)
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr,
"return_code": result.returncode
}
except subprocess.TimeoutExpired:
return {
"success": False,
"error": f"代码执行超时（限制：{self.timeout}秒）"
}
finally:
os.unlink(script_path)
def execute_python_with_analysis(self, code: str) -> dict:
"""执行代码并捕获分析结果"""
# 包装代码以捕获更多上下文
wrapped_code = f"""
import sys
import io
import traceback
# 捕获输出
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
{self._indent_code(code, 4)}
output = sys.stdout.getvalue()
sys.stdout = old_stdout
print("=== OUTPUT ===")
print(output)
print("=== STATUS ===")
print("SUCCESS")
except Exception as e:
sys.stdout = old_stdout
print("=== ERROR ===")
traceback.print_exc()
print("=== STATUS ===")
print("FAILED")
"""
return self.execute_python(wrapped_code)
@staticmethod
def _indent_code(code: str, spaces: int) -> str:
"""缩进代码"""
indent = " " * spaces
return "\n".join(indent + line for line in code.split("\n"))
7.2 代码生成与执行流水线
class CodePipeline:
"""代码生成与执行流水线"""
def __init__(self, sandbox: CodeSandbox):
self.sandbox = sandbox
self.history = []
def generate_and_execute(self, task: str, context: dict = None) -> dict:
"""根据任务描述生成代码并执行"""
# 第一步：生成代码
code = self._generate_code(task, context)
# 第二步：执行代码
result = self.sandbox.execute_python_with_analysis(code)
# 第三步：如果失败，尝试修复
if not result.get("success", False):
fixed_code = self._fix_code(code, result.get("stderr", ""))
result = self.sandbox.execute_python_with_analysis(fixed_code)
code = fixed_code
# 记录历史
self.history.append({
"task": task,
"code": code,
"result": result,
"timestamp": datetime.now().isoformat()
})
return result
def _generate_code(self, task: str, context: dict = None) -> str:
"""根据任务描述生成 Python 代码"""
# 实际实现中会调用 LLM
prompt = f"""
根据以下任务描述生成 Python 代码：
任务：{task}
上下文：{json.dumps(context, ensure_ascii=False) if context else '无'}
要求：
1. 代码必须完整可执行
2. 包含必要的 import 语句
3. 输出结果使用 print 语句
4. 处理可能的异常
"""
return llm_generate_code(prompt)
def _fix_code(self, code: str, error: str) -> str:
"""根据错误信息修复代码"""
prompt = f"""
以下代码执行出错，请修复：
代码：
```python
{code}
```
错误信息：
{error}
请输出修复后的完整代码。
"""
return llm_generate_code(prompt)
8. 数据分析与可视化
8.1 数据分析工具链
import pandas as pd
import numpy as np
from typing import Union
class DataAnalysisAgent:
"""数据分析 Agent"""
def load_data(self, source: str, format: str = "auto") -> pd.DataFrame:
"""加载数据，支持多种格式"""
if format == "auto":
if source.endswith(".csv"):
format = "csv"
elif source.endswith(".json"):
format = "json"
elif source.endswith(".xlsx"):
format = "excel"
elif source.endswith(".parquet"):
format = "parquet"
loaders = {
"csv": lambda s: pd.read_csv(s),
"json": lambda s: pd.read_json(s),
"excel": lambda s: pd.read_excel(s),
"parquet": lambda s: pd.read_parquet(s),
}
if format not in loaders:
raise ValueError(f"不支持的格式: {format}")
df = loaders[format](source)
print(f"已加载数据: {df.shape[0]} 行 × {df.shape[1]} 列")
return df
def explore(self, df: pd.DataFrame) -> dict:
"""数据探索性分析"""
result = {
"shape": df.shape,
"columns": list(df.columns),
"dtypes": df.dtypes.to_dict(),
"missing_values": df.isnull().sum().to_dict(),
"missing_percentage": (df.isnull().sum() / len(df) * 100).to_dict(),
"duplicates": df.duplicated().sum(),
}
# 数值列统计
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
if numeric_cols:
result["numeric_stats"] = df[numeric_cols].describe().to_dict()
# 分类列统计
cat_cols = df.select_dtypes(include=['object']).columns.tolist()
if cat_cols:
result["categorical_stats"] = {}
for col in cat_cols:
result["categorical_stats"][col] = {
"unique_count": df[col].nunique(),
"top_values": df[col].value_counts().head(5).to_dict()
}
return result
def clean_data(self, df: pd.DataFrame, config: dict = None) -> pd.DataFrame:
"""数据清洗"""
if config is None:
config = {}
original_shape = df.shape
# 去除重复行
if config.get("remove_duplicates", True):
df = df.drop_duplicates()
# 处理缺失值
fill_strategy = config.get("fill_strategy", {})
for col, strategy in fill_strategy.items():
if col in df.columns:
if strategy == "mean":
df[col] = df[col].fillna(df[col].mean())
elif strategy == "median":
df[col] = df[col].fillna(df[col].median())
elif strategy == "mode":
df[col] = df[col].fillna(df[col].mode()[0])
elif strategy == "drop":
df = df.dropna(subset=[col])
else:
df[col] = df[col].fillna(strategy)
# 数据类型转换
type_conversions = config.get("type_conversions", {})
for col, dtype in type_conversions.items():
if col in df.columns:
try:
df[col] = df[col].astype(dtype)
except (ValueError, TypeError) as e:
print(f"列 {col} 类型转换失败: {e}")
print(f"数据清洗: {original_shape} → {df.shape}")
return df
def analyze(self, df: pd.DataFrame, analysis_type: str, **kwargs) -> dict:
"""执行分析"""
analyzers = {
"correlation": self._correlation_analysis,
"trend": self._trend_analysis,
"distribution": self._distribution_analysis,
"comparison": self._comparison_analysis,
"outlier": self._outlier_detection,
}
if analysis_type not in analyzers:
raise ValueError(f"不支持的分析类型: {analysis_type}")
return analyzers[analysis_type](df, **kwargs)
def _correlation_analysis(self, df: pd.DataFrame, **kwargs) -> dict:
"""相关性分析"""
numeric_df = df.select_dtypes(include=[np.number])
corr_matrix = numeric_df.corr()
# 找出强相关对
strong_correlations = []
for i in range(len(corr_matrix.columns)):
for j in range(i+1, len(corr_matrix.columns)):
corr_value = corr_matrix.iloc[i, j]
if abs(corr_value) > 0.7:
strong_correlations.append({
"var1": corr_matrix.columns[i],
"var2": corr_matrix.columns[j],
"correlation": round(corr_value, 4)
})
return {
"correlation_matrix": corr_matrix.to_dict(),
"strong_correlations": strong_correlations
}
def _distribution_analysis(self, df: pd.DataFrame, column: str = None, **kwargs) -> dict:
"""分布分析"""
if column:
cols = [column]
else:
cols = df.select_dtypes(include=[np.number]).columns.tolist()
results = {}
for col in cols:
series = df[col].dropna()
results[col] = {
"mean": round(series.mean(), 4),
"median": round(series.median(), 4),
"std": round(series.std(), 4),
"skewness": round(series.skew(), 4),
"kurtosis": round(series.kurtosis(), 4),
"min": series.min(),
"max": series.max(),
"q25": round(series.quantile(0.25), 4),
"q75": round(series.quantile(0.75), 4),
}
return results
def _outlier_detection(self, df: pd.DataFrame, **kwargs) -> dict:
"""异常值检测"""
method = kwargs.get("method", "iqr")
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
outliers = {}
for col in numeric_cols:
series = df[col].dropna()
if method == "iqr":
q1 = series.quantile(0.25)
q3 = series.quantile(0.75)
iqr = q3 - q1
lower = q1 - 1.5 * iqr
upper = q3 + 1.5 * iqr
mask = (series < lower) | (series > upper)
elif method == "zscore":
z_scores = np.abs((series - series.mean()) / series.std())
mask = z_scores > 3
outlier_count = mask.sum()
if outlier_count > 0:
outliers[col] = {
"count": int(outlier_count),
"percentage": round(outlier_count / len(series) * 100, 2),
"values": series[mask].tolist()[:10] # 最多展示10个
}
return outliers
8.2 自动生成可视化
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg')
class VisualizationAgent:
"""可视化 Agent"""
def __init__(self, output_dir: str = "./charts"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
def auto_visualize(self, df: pd.DataFrame, analysis_result: dict) -> list:
"""根据数据和分析结果自动选择并生成图表"""
charts = []
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
cat_cols = df.select_dtypes(include=['object']).columns.tolist()
# 数值分布直方图
for col in numeric_cols[:4]:
path = self._histogram(df, col)
charts.append({"type": "histogram", "column": col, "path": path})
# 分类计数图
for col in cat_cols[:4]:
path = self._bar_chart(df, col)
charts.append({"type": "bar", "column": col, "path": path})
# 相关性热力图
if len(numeric_cols) > 1:
path = self._heatmap(df[numeric_cols].corr())
charts.append({"type": "heatmap", "path": path})
# 时间序列图（如果存在日期列）
date_cols = df.select_dtypes(include=['datetime64']).columns.tolist()
if date_cols and numeric_cols:
path = self._time_series(df, date_cols[0], numeric_cols[0])
charts.append({"type": "time_series", "path": path})
return charts
def _histogram(self, df: pd.DataFrame, column: str) -> str:
fig, ax = plt.subplots(figsize=(10, 6))
df[column].hist(bins=30, ax=ax, color='steelblue', edgecolor='white')
ax.set_title(f'{column} 分布图', fontsize=14)
ax.set_xlabel(column)
ax.set_ylabel('频次')
path = str(self.output_dir / f"hist_{column}.png")
fig.savefig(path, dpi=150, bbox_inches='tight')
plt.close(fig)
return path
def _bar_chart(self, df: pd.DataFrame, column: str) -> str:
fig, ax = plt.subplots(figsize=(10, 6))
counts = df[column].value_counts().head(10)
counts.plot(kind='barh', ax=ax, color='steelblue')
ax.set_title(f'{column} 分类统计', fontsize=14)
ax.set_xlabel('数量')
path = str(self.output_dir / f"bar_{column}.png")
fig.savefig(path, dpi=150, bbox_inches='tight')
plt.close(fig)
return path
def _heatmap(self, corr_matrix) -> str:
fig, ax = plt.subplots(figsize=(10, 8))
im = ax.imshow(corr_matrix, cmap='RdYlBu_r', aspect='auto', vmin=-1, vmax=1)
ax.set_xticks(range(len(corr_matrix.columns)))
ax.set_yticks(range(len(corr_matrix.columns)))
ax.set_xticklabels(corr_matrix.columns, rotation=45, ha='right')
ax.set_yticklabels(corr_matrix.columns)
plt.colorbar(im)
ax.set_title('相关性热力图', fontsize=14)
path = str(self.output_dir / "heatmap_correlation.png")
fig.savefig(path, dpi=150, bbox_inches='tight')
plt.close(fig)
return path
9. 多模态理解能力
9.1 图像理解
class MultimodalAgent:
"""多模态理解 Agent"""
def analyze_image(self, image_path: str, task: str = "describe") -> dict:
"""分析图像内容"""
import base64
with open(image_path, "rb") as f:
image_data = base64.b64encode(f.read()).decode()
prompts = {
"describe": "请详细描述这张图片的内容",
"ocr": "请提取图片中的所有文字内容",
"data": "如果图片中包含图表或数据，请提取其中的数据",
"analyze": "请分析这张图片，识别其中的关键信息和模式"
}
prompt = prompts.get(task, task)
# 调用多模态模型
result = llm_analyze_image(image_data, prompt)
return {
"task": task,
"image": image_path,
"analysis": result
}
def extract_chart_data(self, image_path: str) -> dict:
"""从图表图片中提取数据"""
prompt = """
请分析这张图表，提取以下信息：
1. 图表类型（柱状图/折线图/饼图等）
2. X轴和Y轴的含义
3. 所有数据点的数值
4. 图例信息
请以JSON格式输出提取的数据。
"""
result = self.analyze_image(image_path, prompt)
return json.loads(result["analysis"])
def compare_images(self, image1: str, image2: str) -> dict:
"""对比两张图片"""
prompt = """
请对比这两张图片，分析：
1. 相同点
2. 不同点
3. 各自的特点
4. 总结性对比结论
"""
# 实际实现需要支持多图输入的模型
return {"comparison": "对比结果"}
9.2 文档理解
class DocumentAgent:
"""文档理解 Agent"""
def process_pdf(self, pdf_path: str) -> dict:
"""处理 PDF 文档"""
import fitz # PyMuPDF
doc = fitz.open(pdf_path)
pages = []
for page_num in range(len(doc)):
page = doc[page_num]
text = page.get_text()
images = page.get_images()
pages.append({
"page": page_num + 1,
"text": text,
"image_count": len(images),
"char_count": len(text)
})
return {
"total_pages": len(doc),
"pages": pages,
"full_text": "\n".join(p["text"] for p in pages)
}
def summarize_document(self, text: str, max_length: int = 500) -> str:
"""文档摘要"""
prompt = f"""
请对以下文档内容进行摘要，要求：
1. 保留关键信息
2. 摘要长度不超过{max_length}字
3. 使用清晰的结构
文档内容：
{text[:5000]} # 限制输入长度
"""
return llm_summarize(prompt)
10. 自定义 Agent 开发
10.1 Agent 框架
from typing import Callable, List, Dict, Any
from dataclasses import dataclass, field
from enum import Enum
class AgentRole(Enum):
RESEARCHER = "researcher" # 调研型 Agent
ANALYST = "analyst" # 分析型 Agent
CODER = "coder" # 编码型 Agent
WRITER = "writer" # 写作型 Agent
COORDINATOR = "coordinator" # 协调型 Agent
@dataclass
class AgentConfig:
name: str
role: AgentRole
description: str
tools: List[str]
system_prompt: str
max_iterations: int = 10
temperature: float = 0.7
class CustomAgent:
"""自定义 Agent 基类"""
def __init__(self, config: AgentConfig, tool_registry: ToolRegistry):
self.config = config
self.tools = tool_registry
self.memory = []
self.state = AgentState(task_id=config.name)
async def run(self, task: str) -> dict:
"""执行任务"""
self.state.status = "planning"
# 生成执行计划
plan = await self._plan(task)
self.state.plan = plan
self.state.status = "executing"
# 逐步执行
results = []
for step in plan:
try:
result = await self._execute_step(step)
results.append(result)
self.state.advance()
self._update_memory(step, result)
except Exception as e:
self.state.errors.append({
"step": step,
"error": str(e)
})
# 尝试错误恢复
recovery = await self._handle_error(step, e)
if recovery:
results.append(recovery)
else:
self.state.status = "failed"
break
self.state.status = "completed"
return {
"task": task,
"status": self.state.status,
"results": results,
"artifacts": self.state.artifacts
}
async def _plan(self, task: str) -> list:
"""生成任务计划"""
prompt = f"""
{self.config.system_prompt}
可用工具：{self.tools.list_tools()}
任务：{task}
请生成详细的执行计划，每个步骤指定：
1. action: 具体操作
2. tool: 使用的工具
3. params: 工具参数
4. expected_output: 预期输出
"""
return await llm_plan(prompt)
async def _execute_step(self, step: dict) -> dict:
"""执行单个步骤"""
tool_name = step.get("tool")
params = step.get("params", {})
# 处理参数中的变量替换
resolved_params = self._resolve_params(params)
result = self.tools.execute(tool_name, **resolved_params)
return result
def _resolve_params(self, params: dict) -> dict:
"""解析参数中的变量引用"""
resolved = {}
for key, value in params.items():
if isinstance(value, str) and value.startswith("$"):
# 从上下文中获取变量值
var_name = value[1:]
resolved[key] = self.state.context.get(var_name, value)
elif isinstance(value, dict):
resolved[key] = self._resolve_params(value)
else:
resolved[key] = value
return resolved
def _update_memory(self, step: dict, result: dict):
"""更新 Agent 记忆"""
self.memory.append({
"step": step,
"result": result,
"timestamp": datetime.now().isoformat()
})
# 更新上下文
if "output_key" in step:
self.state.context[step["output_key"]] = result
10.2 多 Agent 协作
class AgentOrchestrator:
"""多 Agent 协作编排器"""
def __init__(self):
self.agents: Dict[str, CustomAgent] = {}
self.message_queue = []
def register_agent(self, agent: CustomAgent):
"""注册 Agent"""
self.agents[agent.config.name] = agent
async def execute_workflow(self, workflow: dict) -> dict:
"""执行多 Agent 工作流"""
results = {}
for stage in workflow["stages"]:
stage_type = stage.get("type", "sequential")
if stage_type == "parallel":
# 并行执行
tasks = []
for agent_task in stage["tasks"]:
agent = self.agents[agent_task["agent"]]
tasks.append(agent.run(agent_task["task"]))
stage_results = await asyncio.gather(*tasks)
for agent_task, result in zip(stage["tasks"], stage_results):
results[agent_task["agent"]] = result
elif stage_type == "sequential":
# 顺序执行
for agent_task in stage["tasks"]:
agent = self.agents[agent_task["agent"]]
# 将前序结果注入任务上下文
task = self._inject_context(agent_task["task"], results)
result = await agent.run(task)
results[agent_task["agent"]] = result
elif stage_type == "review":
# 审查模式：执行者 + 审查者
executor = self.agents[stage["executor"]]
reviewer = self.agents[stage["reviewer"]]
exec_result = await executor.run(stage["task"])
review_result = await reviewer.run(
f"请审查以下执行结果的质量和正确性：\n{json.dumps(exec_result, ensure_ascii=False)}"
)
results[stage["executor"]] = exec_result
results[stage["reviewer"]] = review_result
return results
def _inject_context(self, task: str, previous_results: dict) -> str:
"""将前序结果注入到任务描述中"""
context_str = json.dumps(previous_results, ensure_ascii=False, indent=2)
return f"{task}\n\n前序结果参考：\n{context_str}"
11. 企业级工作流设计
11.1 工作流引擎
from dataclasses import dataclass
from typing import Optional
import asyncio
@dataclass
class WorkflowStep:
id: str
name: str
agent: str
task_template: str
dependencies: list = field(default_factory=list)
retry_count: int = 2
timeout: int = 300 # 秒
condition: Optional[str] = None # 条件执行
class WorkflowEngine:
"""企业级工作流引擎"""
def __init__(self, orchestrator: AgentOrchestrator):
self.orchestrator = orchestrator
self.workflows = {}
self.executions = {}
def define_workflow(self, workflow_id: str, steps: List[WorkflowStep]):
"""定义工作流"""
self.workflows[workflow_id] = steps
async def execute(self, workflow_id: str, inputs: dict) -> dict:
"""执行工作流"""
if workflow_id not in self.workflows:
raise ValueError(f"工作流 '{workflow_id}' 未定义")
steps = self.workflows[workflow_id]
execution_id = f"{workflow_id}_{int(time.time())}"
self.executions[execution_id] = {
"workflow_id": workflow_id,
"status": "running",
"started_at": datetime.now().isoformat(),
"inputs": inputs,
"step_results": {},
"logs": []
}
# 构建依赖图
completed = set()
context = inputs.copy()
try:
while len(completed) < len(steps):
# 找出可执行的步骤（依赖已完成）
ready_steps = [
s for s in steps
if s.id not in completed
and all(dep in completed for dep in s.dependencies)
]
if not ready_steps:
raise RuntimeError("检测到循环依赖或无法满足的依赖")
# 检查条件执行
executable = []
for step in ready_steps:
if step.condition:
if self._evaluate_condition(step.condition, context):
executable.append(step)
else:
completed.add(step.id)
self._log(execution_id, f"跳过步骤 {step.name}（条件不满足）")
else:
executable.append(step)
# 并行执行无依赖关系的步骤
if executable:
tasks = []
for step in executable:
task = self._execute_step_with_retry(execution_id, step, context)
tasks.append((step, task))
for step, task in tasks:
result = await task
completed.add(step.id)
context[step.id] = result
self.executions[execution_id]["step_results"][step.id] = result
self.executions[execution_id]["status"] = "completed"
self.executions[execution_id]["completed_at"] = datetime.now().isoformat()
except Exception as e:
self.executions[execution_id]["status"] = "failed"
self.executions[execution_id]["error"] = str(e)
return self.executions[execution_id]
async def _execute_step_with_retry(self, execution_id: str, step: WorkflowStep, context: dict) -> dict:
"""带重试的步骤执行"""
for attempt in range(step.retry_count + 1):
try:
self._log(execution_id, f"执行步骤: {step.name} (尝试 {attempt + 1})")
# 解析任务模板
task = step.task_template.format(**context)
# 执行
agent = self.orchestrator.agents[step.agent]
result = await asyncio.wait_for(
agent.run(task),
timeout=step.timeout
)
self._log(execution_id, f"步骤 {step.name} 完成")
return result
except asyncio.TimeoutError:
self._log(execution_id, f"步骤 {step.name} 超时")
if attempt == step.retry_count:
raise
except Exception as e:
self._log(execution_id, f"步骤 {step.name} 失败: {str(e)}")
if attempt == step.retry_count:
raise
await asyncio.sleep(2 ** attempt) # 指数退避
def _evaluate_condition(self, condition: str, context: dict) -> bool:
"""评估条件表达式"""
try:
return eval(condition, {"__builtins__": {}}, context)
except:
return True
def _log(self, execution_id: str, message: str):
"""记录日志"""
self.executions[execution_id]["logs"].append({
"timestamp": datetime.now().isoformat(),
"message": message
})
11.2 完整企业工作流示例
async def setup_enterprise_workflow():
"""搭建企业级数据分析工作流"""
# 1. 创建 Agent 编排器
orchestrator = AgentOrchestrator()
# 2. 注册各类 Agent
orchestrator.register_agent(CustomAgent(
config=AgentConfig(
name="data_collector",
role=AgentRole.RESEARCHER,
description="数据采集 Agent",
tools=["browser", "web_search", "api_call"],
system_prompt="你是一个专业的数据采集专家，擅长从各种来源收集结构化数据。"
),
tool_registry=registry
))
orchestrator.register_agent(CustomAgent(
config=AgentConfig(
name="data_analyst",
role=AgentRole.ANALYST,
description="数据分析 Agent",
tools=["code_execution", "file_manager"],
system_prompt="你是一个资深数据分析师，擅长数据清洗、统计分析和趋势预测。"
),
tool_registry=registry
))
orchestrator.register_agent(CustomAgent(
config=AgentConfig(
name="report_writer",
role=AgentRole.WRITER,
description="报告撰写 Agent",
tools=["file_manager", "code_execution"],
system_prompt="你是一个专业的商业报告撰写专家，擅长将数据转化为有洞察力的报告。"
),
tool_registry=registry
))
# 3. 定义工作流
engine = WorkflowEngine(orchestrator)
workflow_steps = [
WorkflowStep(
id="collect",
name="数据采集",
agent="data_collector",
task_template="采集{industry}行业2024年的市场数据，包括主要品牌的销量、市场份额、增长率",
dependencies=[]
),
WorkflowStep(
id="analyze",
name="数据分析",
agent="data_analyst",
task_template="对采集到的数据进行分析：{collect}，计算关键指标并识别趋势",
dependencies=["collect"]
),
WorkflowStep(
id="report",
name="报告生成",
agent="report_writer",
task_template="基于分析结果生成专业的市场分析报告：{analyze}",
dependencies=["analyze"]
)
]
engine.define_workflow("market_analysis", workflow_steps)
# 4. 执行工作流
result = await engine.execute(
"market_analysis",
inputs={"industry": "新能源汽车"}
)
return result
12. 实战项目一：自动化市场研究报告生成
12.1 项目概述
本项目构建一个全自动的市场研究报告生成系统，输入一个行业名称，自动完成数据采集、分析、可视化和报告生成。
12.2 项目架构
market_research_system/
├── agents/
│ ├── __init__.py
│ ├── collector.py # 数据采集 Agent
│ ├── analyzer.py # 数据分析 Agent
│ ├── visualizer.py # 可视化 Agent
│ └── writer.py # 报告撰写 Agent
├── tools/
│ ├── __init__.py
│ ├── search_tool.py # 搜索工具
│ ├── browser_tool.py # 浏览器工具
│ └── chart_tool.py # 图表工具
├── templates/
│ └── report_template.md # 报告模板
├── output/
│ ├── data/ # 采集的原始数据
│ ├── charts/ # 生成的图表
│ └── reports/ # 最终报告
├── config.py # 配置文件
└── main.py # 主程序入口
12.3 核心代码实现
# main.py - 主程序入口
import asyncio
from agents.collector import MarketDataCollector
from agents.analyzer import MarketAnalyzer
from agents.visualizer import ChartGenerator
from agents.writer import ReportWriter
class MarketResearchSystem:
"""市场研究报告自动化系统"""
def __init__(self):
self.collector = MarketDataCollector()
self.analyzer = MarketAnalyzer()
self.visualizer = ChartGenerator()
self.writer = ReportWriter()
async def generate_report(self, industry: str, output_dir: str = "./output") -> str:
"""
生成市场研究报告
Args:
industry: 行业名称（如"新能源汽车"、"云计算"、"人工智能"）
output_dir: 输出目录
Returns:
报告文件路径
"""
print(f"🚀 开始生成 {industry} 行业市场研究报告...")
# 第一阶段：数据采集
print("\n📊 第一阶段：数据采集")
raw_data = await self.collector.collect(industry)
print(f" ✓ 已采集 {len(raw_data['sources'])} 个数据源")
print(f" ✓ 获得 {len(raw_data['brands'])} 个品牌数据")
# 第二阶段：数据分析
print("\n📈 第二阶段：数据分析")
analysis = self.analyzer.analyze(raw_data)
print(f" ✓ 市场规模: {analysis['market_size']}")
print(f" ✓ TOP3品牌: {', '.join(analysis['top_brands'])}")
print(f" ✓ 识别 {len(analysis['trends'])} 个关键趋势")
# 第三阶段：可视化
print("\n🎨 第三阶段：生成图表")
charts = self.visualizer.generate_all(analysis, output_dir=f"{output_dir}/charts")
print(f" ✓ 已生成 {len(charts)} 张图表")
# 第四阶段：报告撰写
print("\n📝 第四阶段：撰写报告")
report_path = self.writer.write_report(
industry=industry,
data=raw_data,
analysis=analysis,
charts=charts,
output_dir=f"{output_dir}/reports"
)
print(f"\n✅ 报告已生成: {report_path}")
return report_path
# 数据采集 Agent
class MarketDataCollector:
"""市场数据采集 Agent"""
async def collect(self, industry: str) -> dict:
"""采集指定行业的市场数据"""
data = {
"industry": industry,
"collected_at": datetime.now().isoformat(),
"sources": [],
"brands": [],
"market_data": {}
}
# 搜索行业报告
search_queries = [
f"{industry} 2024年市场规模 数据",
f"{industry} 品牌排名 销量",
f"{industry} 市场份额 TOP10",
f"{industry} 行业趋势 发展前景"
]
for query in search_queries:
results = await self._search(query)
data["sources"].extend(results)
# 提取品牌数据
data["brands"] = await self._extract_brand_data(data["sources"])
# 提取市场数据
data["market_data"] = await self._extract_market_data(data["sources"])
return data
async def _search(self, query: str) -> list:
"""执行搜索"""
# 使用搜索工具
return [{"query": query, "results": []}]
async def _extract_brand_data(self, sources: list) -> list:
"""从搜索结果中提取品牌数据"""
# 使用 LLM 从非结构化文本中提取结构化数据
return []
async def _extract_market_data(self, sources: list) -> dict:
"""提取市场整体数据"""
return {}
# 数据分析 Agent
class MarketAnalyzer:
"""市场数据分析 Agent"""
def analyze(self, raw_data: dict) -> dict:
"""分析采集的原始数据"""
analysis = {
"market_size": self._estimate_market_size(raw_data),
"top_brands": self._rank_brands(raw_data),
"market_share": self._calculate_share(raw_data),
"growth_rate": self._calculate_growth(raw_data),
"trends": self._identify_trends(raw_data),
"insights": self._generate_insights(raw_data)
}
return analysis
def _estimate_market_size(self, data: dict) -> str:
"""估算市场规模"""
# 实际实现中会基于多源数据交叉验证
return "待计算"
def _rank_brands(self, data: dict) -> list:
"""品牌排名"""
brands = data.get("brands", [])
return sorted(brands, key=lambda x: x.get("sales", 0), reverse=True)[:10]
def _calculate_share(self, data: dict) -> dict:
"""计算市场份额"""
return {}
def _calculate_growth(self, data: dict) -> dict:
"""计算增长率"""
return {}
def _identify_trends(self, data: dict) -> list:
"""识别行业趋势"""
return []
def _generate_insights(self, data: dict) -> list:
"""生成洞察"""
return []
# 报告撰写 Agent
class ReportWriter:
"""报告撰写 Agent"""
def write_report(self, industry: str, data: dict, analysis: dict,
charts: list, output_dir: str) -> str:
"""撰写完整报告"""
report = self._generate_report_content(industry, data, analysis, charts)
# 保存报告
os.makedirs(output_dir, exist_ok=True)
filename = f"{industry}_市场研究报告_{datetime.now().strftime('%Y%m%d')}.md"
filepath = os.path.join(output_dir, filename)
with open(filepath, 'w', encoding='utf-8') as f:
f.write(report)
return filepath
def _generate_report_content(self, industry, data, analysis, charts) -> str:
"""生成报告内容"""
report = f"""# {industry}行业市场研究报告
> 报告日期：{datetime.now().strftime('%Y年%m月%d日')}
> 数据来源：{len(data['sources'])} 个公开数据源
> 分析方法：多源数据交叉验证 + 趋势分析
---
## 目录
1. 行业概述
2. 市场规模与增长
3. 竞争格局分析
4. 主要品牌深度分析
5. 行业趋势与展望
6. 风险与机遇
7. 结论与建议
---
## 1. 行业概述
{industry}行业近年来呈现快速增长态势。本报告基于多源公开数据，
对该行业的市场规模、竞争格局、发展趋势进行系统分析。
## 2. 市场规模与增长
### 2.1 市场规模
根据综合分析，{industry}行业当前市场规模约为 **{analysis.get('market_size', 'N/A')}**。
### 2.2 增长趋势
{self._format_growth_section(analysis.get('growth_rate', {}))}
## 3. 竞争格局分析
### 3.1 市场份额分布
{self._format_share_section(analysis.get('market_share', {}))}
### 3.2 竞争态势
{self._format_competition_section(analysis)}
## 4. 主要品牌深度分析
{self._format_brand_analysis(analysis.get('top_brands', []))}
## 5. 行业趋势与展望
{self._format_trends(analysis.get('trends', []))}
## 6. 风险与机遇
### 风险因素
{self._format_risks(analysis)}
### 发展机遇
{self._format_opportunities(analysis)}
## 7. 结论与建议
{self._format_conclusions(analysis)}
---
*本报告由 AI 自动生成，数据来源于公开信息，仅供参考。*
"""
return report
def _format_growth_section(self, growth_data):
if not growth_data:
return "数据收集中..."
return "增长率分析内容"
def _format_share_section(self, share_data):
if not share_data:
return "数据收集中..."
return "市场份额分析内容"
def _format_competition_section(self, analysis):
return "竞争态势分析内容"
def _format_brand_analysis(self, brands):
if not brands:
return "品牌数据收集中..."
sections = []
for i, brand in enumerate(brands[:5], 1):
sections.append(f"### 4.{i} {brand.get('name', '未知品牌')}\n\n待补充详细分析")
return "\n\n".join(sections)
def _format_trends(self, trends):
if not trends:
return "趋势数据收集中..."
return "\n".join(f"- {t}" for t in trends)
def _format_risks(self, analysis):
return "- 市场竞争加剧\n- 政策变化风险\n- 技术迭代风险"
def _format_opportunities(self, analysis):
return "- 新兴市场需求增长\n- 技术创新带来新机遇\n- 产业链整合机会"
def _format_conclusions(self, analysis):
return "综合以上分析，该行业整体呈现良好发展态势，建议关注头部品牌动态和技术创新方向。"
# 运行示例
async def main():
system = MarketResearchSystem()
report_path = await system.generate_report("新能源汽车")
print(f"\n报告路径: {report_path}")
if __name__ == "__main__":
asyncio.run(main())
13. 实战项目二：竞品分析系统
13.1 项目概述
构建一个自动化竞品分析系统，能够自动采集竞争对手的产品信息、定价策略、用户评价等数据，并生成结构化的竞品分析报告。
13.2 系统设计
# competitor_analysis.py
import asyncio
from dataclasses import dataclass
from typing import List, Dict, Optional
@dataclass
class CompetitorProfile:
"""竞争对手画像"""
name: str
website: str
products: List[Dict]
pricing: Dict
strengths: List[str]
weaknesses: List[str]
user_reviews: List[Dict]
market_position: str
recent_news: List[Dict]
class CompetitorAnalysisSystem:
"""竞品分析系统"""
def __init__(self):
self.collector = CompetitorDataCollector()
self.analyzer = CompetitorAnalyzer()
self.reporter = CompetitorReporter()
async def analyze(self, our_product: str, competitors: List[str]) -> dict:
"""
执行竞品分析
Args:
our_product: 我方产品名称
competitors: 竞争对手列表
Returns:
完整的竞品分析报告
"""
print(f"🔍 开始竞品分析: {our_product} vs {competitors}")
# 1. 采集竞品数据
profiles = []
for comp in competitors:
print(f"\n 采集 {comp} 的数据...")
profile = await self.collector.collect_competitor(comp)
profiles.append(profile)
print(f" ✓ {comp}: {len(profile.products)} 个产品, {len(profile.user_reviews)} 条评价")
# 2. 采集我方产品数据
our_profile = await self.collector.collect_competitor(our_product)
# 3. 对比分析
print("\n📊 执行对比分析...")
comparison = self.analyzer.compare(our_profile, profiles)
# 4. 生成报告
print("\n📝 生成分析报告...")
report = self.reporter.generate(our_product, profiles, comparison)
return report
class CompetitorDataCollector:
"""竞品数据采集器"""
async def collect_competitor(self, company: str) -> CompetitorProfile:
"""采集单个竞品的完整数据"""
profile = CompetitorProfile(
name=company,
website="",
products=[],
pricing={},
strengths=[],
weaknesses=[],
user_reviews=[],
market_position="",
recent_news=[]
)
# 采集官网信息
profile.website = await self._find_website(company)
profile.products = await self._scrape_products(profile.website)
profile.pricing = await self._extract_pricing(profile.website)
# 采集用户评价
profile.user_reviews = await self._collect_reviews(company)
# 采集新闻动态
profile.recent_news = await self._collect_news(company)
# 分析优劣势
profile.strengths = await self._analyze_strengths(profile)
profile.weaknesses = await self._analyze_weaknesses(profile)
return profile
async def _find_website(self, company: str) -> str:
"""查找公司官网"""
# 实现搜索逻辑
return f"https://www.{company.lower().replace(' ', '')}.com"
async def _scrape_products(self, website: str) -> list:
"""采集产品信息"""
# 实现网页爬取逻辑
return []
async def _extract_pricing(self, website: str) -> dict:
"""提取定价信息"""
return {}
async def _collect_reviews(self, company: str) -> list:
"""采集用户评价"""
return []
async def _collect_news(self, company: str) -> list:
"""采集新闻动态"""
return []
async def _analyze_strengths(self, profile: CompetitorProfile) -> list:
"""分析优势"""
return []
async def _analyze_weaknesses(self, profile: CompetitorProfile) -> list:
"""分析劣势"""
return []
class CompetitorAnalyzer:
"""竞品对比分析器"""
def compare(self, our: CompetitorProfile, competitors: List[CompetitorProfile]) -> dict:
"""执行多维度对比分析"""
comparison = {
"product_comparison": self._compare_products(our, competitors),
"pricing_comparison": self._compare_pricing(our, competitors),
"feature_matrix": self._build_feature_matrix(our, competitors),
"sentiment_analysis": self._analyze_sentiment(our, competitors),
"competitive_position": self._assess_position(our, competitors),
"recommendations": self._generate_recommendations(our, competitors)
}
return comparison
def _compare_products(self, our, competitors) -> dict:
"""产品对比"""
return {
"our_products": len(our.products),
"competitor_products": {c.name: len(c.products) for c in competitors}
}
def _compare_pricing(self, our, competitors) -> dict:
"""定价对比"""
return {}
def _build_feature_matrix(self, our, competitors) -> list:
"""构建功能矩阵"""
all_features = set()
# 收集所有功能点
for product in our.products:
all_features.update(product.get("features", []))
for comp in competitors:
for product in comp.products:
all_features.update(product.get("features", []))
# 构建矩阵
matrix = []
for feature in sorted(all_features):
row = {"feature": feature}
row["our_product"] = any(feature in p.get("features", []) for p in our.products)
for comp in competitors:
row[comp.name] = any(feature in p.get("features", []) for p in comp.products)
matrix.append(row)
return matrix
def _analyze_sentiment(self, our, competitors) -> dict:
"""用户情感分析"""
return {}
def _assess_position(self, our, competitors) -> dict:
"""评估竞争位置"""
return {
"our_position": "待评估",
"competitive_landscape": "待分析"
}
def _generate_recommendations(self, our, competitors) -> list:
"""生成竞争策略建议"""
return [
"建议加强差异化功能开发",
"建议优化定价策略",
"建议提升用户体验"
]
class CompetitorReporter:
"""竞品分析报告生成器"""
def generate(self, our_product: str, competitors: List[CompetitorProfile],
comparison: dict) -> dict:
"""生成完整报告"""
report_content = self._build_report(our_product, competitors, comparison)
# 保存报告
output_path = f"./output/reports/竞品分析_{our_product}_{datetime.now().strftime('%Y%m%d')}.md"
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(report_content)
return {
"path": output_path,
"our_product": our_product,
"competitors": [c.name for c in competitors],
"sections": len(comparison)
}
def _build_report(self, our_product, competitors, comparison) -> str:
"""构建报告内容"""
report = f"""# {our_product} 竞品分析报告
> 分析日期：{datetime.now().strftime('%Y年%m月%d日')}
> 分析范围：{our_product} vs {', '.join(c.name for c in competitors)}
---
## 1. 执行摘要
本报告对{our_product}与主要竞争对手进行了全面对比分析，
涵盖产品功能、定价策略、用户评价和市场定位等维度。
## 2. 竞品概览
{self._format_competitor_overview(competitors)}
## 3. 产品功能对比
{self._format_feature_matrix(comparison.get('feature_matrix', []))}
## 4. 定价策略分析
{self._format_pricing(comparison.get('pricing_comparison', {}))}
## 5. 用户评价对比
{self._format_sentiment(comparison.get('sentiment_analysis', {}))}
## 6. 竞争定位分析
{self._format_position(comparison.get('competitive_position', {}))}
## 7. 策略建议
{self._format_recommendations(comparison.get('recommendations', []))}
---
*本报告由 AI 竞品分析系统自动生成*
"""
return report
def _format_competitor_overview(self, competitors):
sections = []
for comp in competitors:
sections.append(f"""### {comp.name}
- **官网**: {comp.website}
- **产品数量**: {len(comp.products)}
- **市场定位**: {comp.market_position or '待分析'}
- **核心优势**: {', '.join(comp.strengths[:3]) or '待分析'}
- **主要劣势**: {', '.join(comp.weaknesses[:3]) or '待分析'}
""")
return "\n".join(sections)
def _format_feature_matrix(self, matrix):
if not matrix:
return "功能矩阵数据收集中..."
# 表头
headers = list(matrix[0].keys())
table = "| " + " | ".join(headers) + " |\n"
table += "| " + " | ".join(["---"] * len(headers)) + " |\n"
for row in matrix:
cells = []
for h in headers:
val = row.get(h, "")
if isinstance(val, bool):
val = "✅" if val else "❌"
cells.append(str(val))
table += "| " + " | ".join(cells) + " |\n"
return table
def _format_pricing(self, pricing):
if not pricing:
return "定价数据收集中..."
return "定价分析内容"
def _format_sentiment(self, sentiment):
if not sentiment:
return "评价数据收集中..."
return "情感分析内容"
def _format_position(self, position):
if not position:
return "定位分析中..."
return "竞争定位内容"
def _format_recommendations(self, recommendations):
if not recommendations:
return "建议生成中..."
return "\n".join(f"{i+1}. {r}" for i, r in enumerate(recommendations))
13.3 运行竞品分析
async def run_competitor_analysis():
"""运行竞品分析示例"""
system = CompetitorAnalysisSystem()
result = await system.analyze(
our_product="Notion",
competitors=["Confluence", "Coda", "Roam Research", "Obsidian"]
)
print(f"\n✅ 竞品分析完成!")
print(f"报告路径: {result['path']}")
print(f"分析了 {len(result['competitors'])} 个竞争对手")
if __name__ == "__main__":
asyncio.run(run_competitor_analysis())
14. 常见问题与解决方案
Q1: Agent 执行任务时频繁超时怎么办？
解决方案：
# 1. 增加超时时间
config = AgentConfig(
name="my_agent",
max_iterations=20,
# 在步骤级别设置更长的超时
)
# 2. 将大任务拆分为更小的子任务
# 不好的做法：一步完成所有数据采集
# 好的做法：分批采集，每批处理5个数据源
# 3. 实现断点续传
class ResumableAgent(CustomAgent):
async def run_with_checkpoint(self, task: str) -> dict:
# 从上次中断的地方继续
checkpoint = self._load_checkpoint()
if checkpoint:
self.state = checkpoint
plan = self.state.plan
else:
plan = await self._plan(task)
self.state.plan = plan
for i, step in enumerate(plan[self.state.current_step:], self.state.current_step):
result = await self._execute_step(step)
self.state.advance()
self._save_checkpoint() # 每步保存检查点
return {"status": "completed", "results": self.state.context}
Q2: 浏览器自动化被目标网站检测和阻止怎么办？
解决方案：
class StealthBrowser:
"""反检测浏览器"""
def __init__(self):
self.config = {
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36",
"viewport": {"width": 1920, "height": 1080},
"locale": "zh-CN",
"timezone": "Asia/Shanghai"
}
async def create_stealth_page(self):
"""创建反检测页面"""
# 使用 playwright-stealth 或类似技术
page = await self.browser.new_page(
user_agent=self.config["user_agent"],
viewport=self.config["viewport"],
locale=self.config["locale"]
)
# 注入反检测脚本
await page.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {get: () => false});
Object.defineProperty(navigator, 'languages', {get: () => ['zh-CN', 'zh', 'en']});
""")
return page
async def human_like_delay(self, min_ms=500, max_ms=2000):
"""模拟人类操作延迟"""
import random
delay = random.randint(min_ms, max_ms) / 1000
await asyncio.sleep(delay)
Q3: 数据分析结果不准确如何改进？
解决方案：
class DataQualityChecker:
"""数据质量检查器"""
def check_quality(self, df: pd.DataFrame) -> dict:
"""全面的数据质量检查"""
issues = []
# 1. 检查缺失值
missing = df.isnull().sum()
for col, count in missing.items():
if count > 0:
pct = count / len(df) * 100
issues.append({
"type": "missing_values",
"column": col,
"count": int(count),
"percentage": round(pct, 2),
"severity": "high" if pct > 30 else "medium" if pct > 10 else "low"
})
# 2. 检查异常值
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
q1 = df[col].quantile(0.25)
q3 = df[col].quantile(0.75)
iqr = q3 - q1
outliers = ((df[col] < q1 - 3*iqr) | (df[col] > q3 + 3*iqr)).sum()
if outliers > 0:
issues.append({
"type": "outliers",
"column": col,
"count": int(outliers),
"severity": "medium"
})
# 3. 检查数据一致性
for col in df.select_dtypes(include=['object']).columns:
# 检查是否有前后空格等格式问题
if df[col].str.contains(r'^\s|\s$', regex=True).any():
issues.append({
"type": "formatting",
"column": col,
"description": "存在前后空格",
"severity": "low"
})
return {
"total_issues": len(issues),
"high_severity": len([i for i in issues if i["severity"] == "high"]),
"issues": issues,
"quality_score": max(0, 100 - len(issues) * 5)
}
Q4: 多 Agent 协作时出现死锁或资源竞争怎么办？
解决方案：
class AgentResourceManager:
"""Agent 资源管理器"""
def __init__(self):
self.locks = {}
self.semaphores = {}
async def acquire_resource(self, agent_id: str, resource: str, timeout: int = 30):
"""获取资源锁"""
if resource not in self.locks:
self.locks[resource] = asyncio.Lock()
try:
await asyncio.wait_for(self.locks[resource].acquire(), timeout=timeout)
return True
except asyncio.TimeoutError:
raise ResourceBusyError(f"资源 {resource} 被占用，等待超时")
def release_resource(self, agent_id: str, resource: str):
"""释放资源锁"""
if resource in self.locks:
self.locks[resource].release()
Q5: 如何优化 Agent 的执行效率？
优化策略：
class EfficientAgent(CustomAgent):
"""优化效率的 Agent"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cache = {}
self.batch_size = 10
async def cached_execute(self, tool: str, params: dict) -> dict:
"""带缓存的工具执行"""
cache_key = f"{tool}:{json.dumps(params, sort_keys=True)}"
if cache_key in self.cache:
return self.cache[cache_key]
result = self.tools.execute(tool, **params)
self.cache[cache_key] = result
return result
async def batch_execute(self, tasks: list) -> list:
"""批量执行相似任务"""
results = []
for i in range(0, len(tasks), self.batch_size):
batch = tasks[i:i+self.batch_size]
batch_results = await asyncio.gather(*[
self._execute_step(task) for task in batch
])
results.extend(batch_results)
return results
15. 进阶资源与社区
15.1 推荐学习路径
初级阶段（1-2周）
├── 理解 Agent 基本概念
├── 掌握单工具调用
├── 完成简单的自动化任务
└── 学习任务分解基础
中级阶段（2-4周）
├── 掌握多工具组合使用
├── 学习浏览器自动化
├── 实现数据分析流水线
└── 构建自定义工具
高级阶段（4-8周）
├── 设计多 Agent 协作系统
├── 实现企业级工作流
├── 优化 Agent 执行效率
└── 构建生产级应用
15.2 关键技术栈
- 编程语言：Python 3.10+
- 异步框架：asyncio, aiohttp
- 浏览器自动化：Playwright, Selenium
- 数据处理：Pandas, NumPy
- 可视化：Matplotlib, Plotly
- AI/LLM：OpenAI API, LangChain
- 任务队列：Celery, Redis Queue
15.3 最佳实践总结
- 任务分解要适度：每个子任务应该是可独立验证的最小单元
- 工具描述要精确：Agent 依赖工具描述来选择正确工具
- 错误处理要完善：每个工具调用都应有异常处理和重试机制
- 上下文管理要谨慎：避免上下文过长导致信息丢失
- 安全边界要明确：限制 Agent 的操作权限，避免危险操作
- 日志记录要详细：记录每一步操作，便于调试和审计
总结
Manus AI 通用 Agent 平台代表了 AI 从"对话"到"行动"的范式转变。通过本教程，你已经掌握了：
- ✅ Agent 架构与设计哲学
- ✅ 任务分解与动态规划
- ✅ 工具链集成与自定义工具开发
- ✅ 浏览器自动化与数据采集
- ✅ 代码执行与数据分析
- ✅ 多模态内容理解
- ✅ 自定义 Agent 与多 Agent 协作
- ✅ 企业级工作流设计
- ✅ 两个完整的实战项目
Agent 技术正在快速演进，持续学习和实践是掌握这一技术的关键。建议从简单的自动化任务开始，逐步构建更复杂的 Agent 系统，在实践中积累经验。
本教程内容基于公开技术和最佳实践整理，仅供学习参考。
