AI Agent入门实战：从零构建一个自动化助手

人间凡尔赛

37人浏览 · 2026-07-22 22:13:39

人间凡尔赛 · 2026-07-22 22:13:39 发布

AI Agent入门实战：从零构建一个自动化助手

引言

2026年，AI Agent 已经从概念演示变成了生产力工具。各大公司都在部署 Agent 系统——从客服到代码审查，从数据分析到报告生成。但很多人对 Agent 的理解还停留在"调一个 API 就完事"的层面。

本文带你从零构建一个真正能用的 AI Agent：它能搜索资料、撰写报告、发送邮件。全程可运行，代码完整。

---

一、AI Agent 是什么？（3分钟理解核心概念）

简单说，AI Agent ≠ 大模型。

| 概念 | 定义 | 比喻 |

|------|------|------|

| LLM | 语言模型，只是"大脑" | 一个很聪明但没手没脚的人 |

| Agent | LLM + 工具 + 记忆 + 规划 | 一个有自己的工作台、工具箱、记事本的人 |

| Workflow | 固定的执行流程 | SOP 手册，按步骤操作 |

| Autonomous Agent | 自主决策执行 | 有自主权的高级员工 |

Agent 的核心能力链：

用户需求 → LLM 理解 → 拆解任务 → 调用工具 → 获取结果 → 整合输出
                             ↻（循环直到完成）

---

二、环境搭建

# 1. 创建项目
mkdir ai-agent-demo && cd ai-agent-demo
python3 -m venv venv
source venv/bin/activate

# 2. 安装依赖
pip install openai langchain langchain-community
pip install httpx python-dotenv beautifulsoup4

# 3. 配置环境变量
echo "OPENAI_API_KEY=sk-your-key" > .env

---

三、从零写一个最小 Agent

3.1 基础结构

# agent_core.py — 最小 Agent 实现
import json
import httpx
from openai import OpenAI
from datetime import datetime

class MiniAgent:
    """一个极简但功能完整的 AI Agent"""
    
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.messages = []
        self.tools = {}
        
    def register_tool(self, name: str, func: callable, description: str):
        """注册 Agent 可使用的工具"""
        self.tools[name] = {"func": func, "description": description}
        
    def run(self, user_input: str) -> str:
        """执行一次任务"""
        self.messages.append({"role": "user", "content": user_input})
        
        # 构建工具描述
        tool_descriptions = "\n".join([
            f"- {name}: {info['description']}"
            for name, info in self.tools.items()
        ])
        
        system_prompt = f"""你是 AI Agent 助手。当前时间：{datetime.now()}
可用工具：
{tool_descriptions}

当你需要调用工具时，请输出 JSON 格式：
{{"tool": "工具名", "args": {{"参数名": "参数值"}}}}

当你准备好最终回答时，直接输出自然语言即可。"""

        messages = [
            {"role": "system", "content": system_prompt},
            *self.messages
        ]
        
        max_iterations = 10
        for i in range(max_iterations):
            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                temperature=0.3,
            )
            
            content = response.choices[0].message.content
            
            # 尝试解析 JSON（工具调用）
            try:
                tool_call = json.loads(content)
                if "tool" in tool_call and "args" in tool_call:
                    tool_name = tool_call["tool"]
                    tool_args = tool_call["args"]
                    
                    if tool_name in self.tools:
                        result = self.tools[tool_name]["func"](**tool_args)
                        messages.append({
                            "role": "assistant", 
                            "content": f"[调用工具 {tool_name}，结果: {result}]"
                        })
                        continue
            except json.JSONDecodeError:
                pass
            
            # 非 JSON 格式，视为最终回答
            self.messages.append({"role": "assistant", "content": content})
            return content
        
        return "Agent 已达到最大迭代次数，请重新提问。"

3.2 注册真实工具

# tools.py — 注册 Agent 的实际能力
import httpx
from bs4 import BeautifulSoup

def web_search(query: str) -> list[dict]:
    """搜索网络获取信息"""
    url = f"https://api.duckduckgo.com/?q={query}&format=json"
    response = httpx.get(url, timeout=10)
    data = response.json()
    
    results = []
    for topic in data.get("RelatedTopics", [])[:3]:
        if "Text" in topic:
            results.append({
                "title": topic.get("Text", "")[:100],
                "url": topic.get("FirstURL", "")
            })
    return results

def scrape_webpage(url: str) -> str:
    """抓取网页内容"""
    try:
        response = httpx.get(url, timeout=15)
        soup = BeautifulSoup(response.text, "html.parser")
        # 移除脚本和样式
        for tag in soup(["script", "style", "nav", "footer"]):
            tag.decompose()
        text = soup.get_text(separator="\n", strip=True)
        return text[:2000]  # 限制长度
    except Exception as e:
        return f"抓取失败: {e}"

def write_report(title: str, content: str) -> str:
    """将报告保存到本地文件"""
    filename = f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(f"# {title}\n\n")
        f.write(content)
    return f"报告已保存到 {filename}"

3.3 组装并运行

# main.py — 组装 Agent 并运行
from dotenv import load_dotenv
import os
from agent_core import MiniAgent
from tools import web_search, scrape_webpage, write_report
from datetime import datetime

load_dotenv()

# 初始化 Agent
agent = MiniAgent(api_key=os.getenv("OPENAI_API_KEY"))

# 注册工具
agent.register_tool("web_search", web_search, "在互联网上搜索信息，参数: query(str)")
agent.register_tool("scrape_webpage", scrape_webpage, "抓取指定 URL 的网页内容，参数: url(str)")
agent.register_tool("write_report", write_report, "将报告内容写入本地文件，参数: title(str), content(str)")

# 执行任务
result = agent.run("请帮我查一下2026年AI Agent行业的最新趋势，然后整理成一份500字左右的报告")
print(result)

运行效果：

$ python main.py
[调用工具 web_search，结果: {...}]
[调用工具 scrape_webpage，结果: {...}]
[调用工具 write_report，结果: "报告已保存到 report_20260721_143022.md"]

已生成报告：2026年AI Agent行业趋势分析...

---

四、用 Dify 搭建无代码 Agent

如果你的团队没有 Python 基础，Dify 是最快的选择。

Dify Agent 搭建步骤：

1. 登录 Dify → 创建空白应用 → 选择"Agent"

2. 配置 LLM：选择 GPT-4o 或 Claude Sonnet

3. 添加工具：

• **Web Search**（内置搜索引擎）

• **Code Interpreter**（执行 Python 代码）

• **自定义 API 工具**（连接你的第三方服务）

4. 设置 Prompt：

   你是一个技术研究助手。当用户要求你研究一个主题时：
   1. 先用搜索引擎查找相关资料
   2. 阅读找到的关键页面
   3. 整理成结构化报告
   4. 如有需要，用 Python 处理数据生成图表

5. 发布为 API → 集成到你的 Slack/飞书/网页

---

五、进阶：集成邮件发送能力

# 添加邮件工具
import smtplib
from email.mime.text import MIMEText

def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = "agent@yourdomain.com"
    msg["To"] = to
    
    with smtplib.SMTP("smtp.gmail.com", 587) as server:
        server.starttls()
        server.login("your@email.com", "your-password")
        server.send_message(msg)
    return f"邮件已发送到 {to}"

# 注册邮件工具
agent.register_tool("send_email", send_email, "发送邮件，参数: to(str), subject(str), body(str)")

现在你的 Agent 可以这样工作：

你：每天早上9点，搜索AI行业新闻，整理简报并发送到我的邮箱

当然，定时任务需要配合 cron 或 n8n 来实现调度。

---

六、CrewAI 多 Agent 协作示例

当你需要多个角色协作时，CrewAI 是最成熟的框架。

# crew_demo.py — 多 Agent 协作
from crewai import Agent, Task, Crew, Process

# 定义三个角色
researcher = Agent(
    role="研究员",
    goal="查找最前沿的技术资料",
    backstory="你是一个资深技术研究员，擅长从海量信息中找到关键内容。",
    tools=[web_search, scrape_webpage],
    llm="gpt-4o"
)

writer = Agent(
    role="写手",
    goal="将复杂的技术信息写成通俗易懂的文章",
    backstory="你是一个技术写手，擅长安抚读者的技术焦虑。",
    llm="gpt-4o"
)

reviewer = Agent(
    role="审查员",
    goal="检查报告的事实准确性和逻辑完整性",
    backstory="你是一个严谨的审查员，不允许任何事实错误和数据错误。",
    llm="gpt-4o"
)

# 分配任务
research_task = Task(
    description="研究2026年AI Agent在企业中的应用现状",
    agent=researcher,
    expected_output="一份包含3-5个关键趋势的研究笔记"
)

write_task = Task(
    description="基于研究笔记，写一篇2000字左右的分析文章",
    agent=writer,
    expected_output="一篇结构完整的技术文章"
)

review_task = Task(
    description="审查文章，确保所有数据准确、逻辑自洽",
    agent=reviewer,
    expected_output="审查通过或修改建议"
)

# 组成 Crew 并运行
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential  # 按顺序执行
)

result = crew.kickoff()
print(result)

---

七、部署建议

| 方式 | 成本 | 适用场景 |

|------|------|---------|

| 本地运行 | GPU 费用 | 开发调试 |

| VPS + Docker | ~$10/月 | 生产环境 |

| Vercel Edge | Free | 轻量 API |

| Dify Cloud | Free-$59/月 | 无代码场景 |

推荐生产部署（Docker Compose）：

version: '3.8'
services:
  agent-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - SENDGRID_API_KEY=${SENDGRID_API_KEY}
    volumes:
      - ./reports:/app/reports
    restart: unless-stopped

---