7 天前，你可能连 Agent 是什么都不确定。现在，你已经从零构建了一个完整的 Agent 系统。

---

今天的目标

1. 把 7 天的代码整合成一个结构清晰的完整项目
2. 学会如何扩展和定制自己的 Agent
3. 了解 Agent 技术的前沿方向

---

完整项目结构

ai_agent/
├── main.py              # 入口文件
├── config.py            # 配置文件
├── requirements.txt     # 依赖
├── agent/
│   ├── __init__.py
│   ├── core.py          # Agent 核心逻辑
│   └── memory.py        # 记忆系统
└── tools/
    ├── __init__.py
    ├── registry.py      # 工具注册表
    ├── calculator.py    # 数学计算器
    ├── file_tools.py    # 文件操作
    └── search.py        # 网页搜索

---

核心文件

config.py — 配置

# config.py

import os

# API 配置
API_KEY = os.getenv("OPENAI_API_KEY", "你的API Key")
BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.mimo.ai/v1")
MODEL = os.getenv("OPENAI_MODEL", "mimo-v2-flash")

# Agent 配置
MAX_ITERATIONS = 10       # Agent Loop 最大迭代次数
MAX_WORKING_MEMORY = 10   # 工作记忆最大条数
SUMMARY_THRESHOLD = 8     # 触发摘要的记忆条数

tools/registry.py — 工具注册表

# tools/registry.py

"""工具注册表：管理所有工具的注册和查找"""

import inspect
from typing import Callable, Dict, Any


class ToolRegistry:
    """工具注册中心"""

    def __init__(self):
        self._tools: Dict[str, Dict[str, Any]] = {}

    def register(self, name: str, description: str, parameters: Dict[str, str], function: Callable):
        """注册一个工具"""
        self._tools[name] = {
            "name": name,
            "description": description,
            "parameters": parameters,
            "function": function,
        }

    def get(self, name: str) -> Dict[str, Any]:
        """获取工具"""
        return self._tools.get(name)

    def execute(self, name: str, params: dict) -> str:
        """执行工具"""
        tool = self._tools.get(name)
        if not tool:
            return f"未知工具：{name}"

        func = tool["function"]
        sig = inspect.signature(func)
        kwargs = {n: params[n] for n in sig.parameters if n in params}

        try:
            return func(**kwargs)
        except Exception as e:
            return f"工具执行错误：{str(e)}"

    def build_prompt(self) -> str:
        """构建工具描述文本"""
        lines = ["你可以使用以下工具：\n"]
        for name, tool in self._tools.items():
            params = ", ".join(
                f"{k}: {v}" for k, v in tool["parameters"].items()
            ) if tool["parameters"] else "无"
            lines.append(f"{name} - {tool['description']}")
            lines.append(f"  参数：{params}")
            lines.append("")
        return "\n".join(lines)


# 全局实例
registry = ToolRegistry()

tools/calculator.py — 计算器

# tools/calculator.py

import math
import re


def calculator(expression: str) -> str:
    """安全的数学计算器"""
    allowed_names = {
        "abs": abs, "round": round,
        "sin": math.sin, "cos": math.cos, "tan": math.tan,
        "sqrt": math.sqrt, "log": math.log, "log10": math.log10,
        "pi": math.pi, "e": math.e,
        "pow": pow, "min": min, "max": max,
    }

    if not re.match(r'^[\d\s\+\-\*\/\.\(\),a-zA-Z]+$', expression):
        return "错误：表达式包含不允许的字符"

    try:
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return str(result)
    except Exception as e:
        return f"计算错误：{str(e)}"

tools/file_tools.py — 文件操作

# tools/file_tools.py

import os


def read_file(file_path: str) -> str:
    """读取文件"""
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        return f"错误：文件 '{file_path}' 不存在"
    except Exception as e:
        return f"读取错误：{str(e)}"


def write_file(file_path: str, content: str) -> str:
    """写入文件"""
    try:
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(content)
        return f"成功写入文件 '{file_path}'，共 {len(content)} 字符"
    except Exception as e:
        return f"写入错误：{str(e)}"


def list_files(directory: str = ".") -> str:
    """列出目录中的文件"""
    try:
        entries = os.listdir(directory)
        result = []
        for entry in entries:
            path = os.path.join(directory, entry)
            if os.path.isfile(path):
                size = os.path.getsize(path)
                result.append(f"{entry} ({size} bytes)")
            elif os.path.isdir(path):
                result.append(f"{entry}/")
        return "\n".join(result) if result else "目录为空"
    except Exception as e:
        return f"列出文件错误：{str(e)}"

tools/search.py — 网页搜索

# tools/search.py

import requests


def web_search(query: str) -> str:
    """搜索网页"""
    try:
        url = "https://api.duckduckgo.com/"
        params = {"q": query, "format": "json", "no_html": 1, "skip_disambig": 1}
        response = requests.get(url, params=params, timeout=5)
        data = response.json()
        results = []
        if data.get("Abstract"):
            results.append(data["Abstract"])
        if data.get("RelatedTopics"):
            for topic in data["RelatedTopics"][:3]:
                if isinstance(topic, dict) and "Text" in topic:
                    results.append(topic["Text"])
        return "\n".join(results) if results else f"关于 '{query}' 暂无详细信息"
    except requests.RequestException:
        return f"搜索 '{query}' 时网络错误"

agent/memory.py — 记忆系统

# agent/memory.py

import json
import os
from openai import OpenAI
import config


class Memory:
    """Agent 的记忆系统"""

    def __init__(self):
        self.working_memory = []
        self.long_term_memory = []
        self.context_file = "agent_memory.json"

    def add(self, role: str, content: str):
        """添加对话"""
        self.working_memory.append({"role": role, "content": content})
        if len(self.working_memory) > config.SUMMARY_THRESHOLD:
            self._summarize()

    def _summarize(self):
        """压缩较早的对话为摘要"""
        if len(self.working_memory) <= 4:
            return

        to_summarize = self.working_memory[:-4]
        self.working_memory = self.working_memory[-4:]

        client = OpenAI(api_key=config.API_KEY, base_url=config.BASE_URL)
        conversation = "\n".join(
            f"{m['role']}: {m['content']}" for m in to_summarize
        )
        response = client.chat.completions.create(
            model=config.MODEL,
            messages=[
                {"role": "system", "content": "用中文简洁地总结对话的关键信息。"},
                {"role": "user", "content": conversation},
            ],
            temperature=0,
        )
        self.long_term_memory.append(response.choices[0].message.content.strip())

    def get_context(self) -> list:
        """获取上下文"""
        messages = []
        if self.long_term_memory:
            summary = "\n".join(self.long_term_memory)
            messages.append({"role": "system", "content": f"历史摘要：\n{summary}"})
        messages.extend(self.working_memory)
        return messages

    def save(self):
        """持久化"""
        with open(self.context_file, "w", encoding="utf-8") as f:
            json.dump({
                "working": self.working_memory,
                "long_term": self.long_term_memory,
            }, f, ensure_ascii=False, indent=2)

    def load(self):
        """加载"""
        if os.path.exists(self.context_file):
            with open(self.context_file, "r", encoding="utf-8") as f:
                data = json.load(f)
                self.working_memory = data.get("working", [])
                self.long_term_memory = data.get("long_term", [])

    def clear(self):
        """清空"""
        self.working_memory = []
        self.long_term_memory = []
        if os.path.exists(self.context_file):
            os.remove(self.context_file)

agent/core.py — Agent 核心

# agent/core.py

"""Agent 的核心逻辑：观察 → 思考 → 行动 循环"""

import json
from openai import OpenAI
import config
from tools.registry import registry
from agent.memory import Memory


class Agent:
    """AI Agent 核心类"""

    def __init__(self):
        self.client = OpenAI(api_key=config.API_KEY, base_url=config.BASE_URL)
        self.memory = Memory()
        self.memory.load()

    def think(self, user_input: str) -> dict:
        """AI 思考并决策"""
        tools_prompt = registry.build_prompt()
        context = self.memory.get_context()

        system_prompt = f"""你是一个智能助手，可以使用工具完成任务。

{tools_prompt}

每一步只调用一个工具。根据结果决定下一步。

JSON 格式：
- 工具调用：{{"thought": "思考", "action": "tool", "tool": "工具名", "params": {{}}}}
- 最终回答：{{"thought": "思考", "action": "final", "response": "回答"}}

只输出 JSON。"""

        messages = [{"role": "system", "content": system_prompt}]
        messages.extend(context)
        messages.append({"role": "user", "content": user_input})

        response = self.client.chat.completions.create(
            model=config.MODEL,
            messages=messages,
            temperature=0,
        )

        reply = response.choices[0].message.content.strip()
        try:
            if reply.startswith("```"):
                reply = reply.split("\n", 1)[1]
                reply = reply.rsplit("```", 1)[0]
            return json.loads(reply)
        except json.JSONDecodeError:
            return {"thought": "解析失败", "action": "final", "response": reply}

    def run(self, user_input: str) -> str:
        """运行 Agent Loop"""
        self.memory.add("user", user_input)

        print(f"\n{'='*50}")
        print(f"你：{user_input}")
        print(f"{'='*50}")

        for i in range(1, config.MAX_ITERATIONS + 1):
            print(f"\n[步骤 {i}/{config.MAX_ITERATIONS}]")

            decision = self.think(user_input)
            thought = decision.get("thought", "")
            action = decision.get("action", "final")

            if thought:
                print(f"[思考]: {thought}")

            if action == "final":
                response = decision.get("response", "无法生成回答。")
                print(f"\nAgent：{response}")
                self.memory.add("assistant", response)
                return response

            if action == "tool":
                tool_name = decision.get("tool", "")
                params = decision.get("params", {})

                print(f"[调用工具]: {tool_name}")
                result = registry.execute(tool_name, params)
                print(f"[结果]: {result[:200]}{'...' if len(result) > 200 else ''}")

        print("\n[达到最大迭代次数]")
        return "任务比较复杂，请尝试简化需求。"

main.py — 入口

# main.py

"""AI Agent 主程序"""

from tools.registry import registry
from tools.calculator import calculator
from tools.file_tools import read_file, write_file, list_files
from tools.search import web_search
from agent.core import Agent


def register_tools():
    """注册所有工具"""
    registry.register(
        name="calculator",
        description="计算数学表达式",
        parameters={"expression": "数学表达式"},
        function=calculator,
    )
    registry.register(
        name="read_file",
        description="读取指定文件的内容",
        parameters={"file_path": "文件路径"},
        function=read_file,
    )
    registry.register(
        name="write_file",
        description="将内容写入指定文件",
        parameters={"file_path": "文件路径", "content": "要写入的内容"},
        function=write_file,
    )
    registry.register(
        name="list_files",
        description="列出当前目录中的所有文件和文件夹",
        parameters={"directory": "目录路径"},
        function=list_files,
    )
    registry.register(
        name="web_search",
        description="搜索互联网获取信息",
        parameters={"query": "搜索关键词"},
        function=web_search,
    )


def main():
    register_tools()
    agent = Agent()

    print("=" * 50)
    print("    我的 AI Agent 系统 v1.0")
    print("=" * 50)
    print("命令：")
    print("  直接输入  → 对话模式（有记忆，自动使用工具）")
    print("  /clear    → 清除对话记忆")
    print("  /memory   → 查看记忆状态")
    print("  /quit     → 退出")
    print("=" * 50)

    while True:
        user_input = input("\n你：").strip()
        if not user_input:
            continue

        if user_input == "/quit":
            agent.memory.save()
            print("记忆已保存。再见！")
            break
        elif user_input == "/clear":
            agent.memory.clear()
            print("记忆已清除。")
            continue
        elif user_input == "/memory":
            print(f"工作记忆：{len(agent.memory.working_memory)} 条")
            print(f"长期记忆：{len(agent.memory.long_term_memory)} 条")
            continue

        agent.run(user_input)


if __name__ == "__main__":
    main()

requirements.txt

openai>=1.0.0
requests>=2.28.0

---

如何扩展：添加自己的工具

只需要三步：

# 1. 写工具函数
def my_weather(city: str) -> str:
    """查询天气"""
    # 这里调用真实的天气 API
    return f"{city}：晴，25°C"

# 2. 在 main.py 中注册
registry.register(
    name="weather",
    description="查询指定城市的天气",
    parameters={"city": "城市名称"},
    function=my_weather,
)

# 3. 完成！AI 会自动学会使用这个工具

AI 不需要重新训练。它会通过 description 理解这个工具能做什么，并在合适的时候调用它。

---

7 天学习回顾

天数	学到了什么	关键概念
Day 1	Agent = 观察 → 思考 → 行动	意图识别、注册表
Day 2	工具的三要素	description、参数、安全执行
Day 3	多步骤执行	工具链、结果传递
Day 4	记忆系统	工作记忆、长期记忆、摘要
Day 5	Agent Loop	自主决策、停止条件、错误恢复
Day 6	Plan-and-Execute	规划器、执行器、依赖管理
Day 7	完整项目	代码组织、扩展方法

---

Agent 技术的前沿方向

你现在掌握的是 Agent 的基础。接下来，你可以关注这些方向：

1. RAG（检索增强生成）

当知识库很大时，不能把所有内容都塞给 AI。RAG 让 Agent 只检索相关的内容：

用户问题 → 向量搜索相关文档 → 只把相关文档给 AI

实现思路：用 sentence-transformers 把文档转成向量，用 faiss 做相似度搜索。

2. 多 Agent 协作

让多个 Agent 分工合作：

用户需求 → 调度 Agent → 分配给专门的 Agent → 汇总结果

实现思路：每个 Agent 有自己的工具集和专长，由一个中央 Agent 调度。

3. 自我反思（Self-Reflection）

让 Agent 能评估自己的输出质量：

执行 → 检查结果 → 如果不满意，重试或调整

实现思路：在 Agent Loop 中加入"检查"步骤，让 AI 评估自己的回答。

4. 工具学习（Tool Learning）

让 Agent 能自己创建工具：

用户需求 → Agent 发现没有合适的工具 → 自己写一个

实现思路：让 Agent 能生成并执行 Python 代码。

---

下一步：从这里开始

如果你想深入 Agent

1. 读论文：ReAct、Plan-and-Execute、Toolformer
2. 用框架：现在回去看 LangChain / AutoGen，你会发现"啊，原来它封装的是这个"
3. 做项目：找一个真实需求，用 Agent 解决它

如果你想做产品

1. 加 Web 界面：用 FastAPI + Vue/React 做一个 Web UI
2. 加更多工具：文件操作、数据库查询、API 调用
3. 加安全控制：限制工具权限、审核 AI 输出

如果你想学 AI

1. 理解 LLM：学习 Transformer 架构
2. 学习微调：用自己的数据训练模型
3. 研究 Agent：这是一个快速发展的领域

---

最后的话

7 天前，你可能觉得 Agent 是一个神秘的黑盒。

现在你知道了：

• Agent 不是魔法，它就是一个循环：观察 → 思考 → 行动
• AI 不是万能的，它需要工具来做事，需要记忆来保持上下文
• 框架不是必须的，理解原理比会用工具更重要

你从零手搓了一个 Agent 系统。每一行代码你都知道为什么在那里。

这就是真正的理解。

接下来，去用你学到的知识做点什么吧。写一个帮你整理文件的 Agent，做一个帮你搜索信息的工具，或者只是把今天学到的讲给朋友听。

学习最好的方式，就是去做。

---

附录：常见问题

Q: API 调用太贵怎么办？

• 使用更便宜的模型（如 mimo-v2-flash）
• 减少 MAX_ITERATIONS
• 缓存 AI 的回复

Q: AI 返回的 JSON 格式不对怎么办？

当前代码已经有基本的容错处理。更稳健的方案是用 OpenAI 的 Function Calling 功能（需要支持该功能的模型）。

Q: 如何支持中文以外的语言？

修改 config.py 中的 MODEL，换成支持目标语言的模型即可。

Q: 如何持久化对话历史？

当前代码使用 JSON 文件保存。生产环境可以换成数据库（SQLite、Redis 等）。

---

恭喜你完成了《7天从零手搓 AI Agent》系列！

如果你觉得这个系列有帮助，欢迎分享给更多人。有问题或建议，欢迎交流。

→ 回到第一篇：为什么这样学