第7章：实操——从零构建一个 Obsidian Agent

本章解决的核心问题：把你前六章学的所有东西——Agent 架构、RAG、工具调用、安全策略——变成一个真实的、能在你自己的 Obsidian vault 上运行的 Agent。

读完这一章，你会有一个能索引你全部笔记、理解你的知识库、帮你搜索和创建笔记的 AI Agent。所有代码可以直接运行。

7.1 我们要做什么

🎨 漫画图解：〈把 Obsidian 变成知识城市〉

先定义清楚这个 Agent 的能力边界：

flowchart LR
    subgraph Obsidian Agent
        T1[🔍 语义搜索
用自然语言搜笔记]
        T2[📖 阅读笔记
打开并理解任意笔记]
        T3[✍️ 创建笔记
自动生成带 frontmatter 的笔记]
        T4[🔗 发现关联
找到可能相关但未链接的笔记]
    end
    T1 --> User[👤 你]
    T2 --> User
    T3 --> User
    T4 --> User

它不能做什么（明确的能力边界）：

• 不能修改已有笔记（防止误删）
• 不能删除文件
• 不能访问网络（只在你本地 vault 内工作）
• 不能执行系统命令

技术架构

flowchart TB
    subgraph 离线阶段
        Vault[📂 Obsidian Vault
所有 .md 文件] --> Chunk[✂️ 切片]
        Chunk --> Embed[🔢 Embedding
OpenAI text-embedding-3-small]
        Embed --> Chroma[🗄️ Chroma 向量数据库]
    end

    subgraph 在线阶段
        You[👤 你输入问题] --> Loop{🔄 Agent 循环}
        Loop -->|需要搜索| Search[🔍 向量检索]
        Chroma -.-> Search
        Loop -->|需要读取| Read[📖 读文件]
        Loop -->|需要创建| Create[✍️ 写文件]
        Loop -->|能直接回答| Answer[💬 回复]
    end

你将用到的技术栈

费用预估：索引 1000 篇中等长度笔记 ≈ $0.05-0.10（一次性）。每次查询 ≈ $0.002-0.01（取决于检索到的上下文长度）。一个月的日常使用大概 $0.5-2。

7.2 环境准备

安装依赖

打开终端，创建项目文件夹，安装依赖：

# 创建项目目录（放在 vault 外面）
mkdir obsidian-agent
cd obsidian-agent

# 创建虚拟环境
python -m venv venv
# Windows:
venv\Scripts\activate
# Mac/Linux:
source venv/bin/activate

# 安装依赖
pip install chromadb openai python-frontmatter tiktoken pyyaml

设置 API Key

# Windows PowerShell:
$env:OPENAI_API_KEY = "sk-你的key"

# Mac/Linux:
export OPENAI_API_KEY="sk-你的key"

没有 OpenAI API Key？ 可以用兼容接口替代。Ollama + 本地模型（如 qwen2.5）、或者 DeepSeek、智谱等国内厂商的 API——只需替换 base_url 和模型名。本章末尾会给出替换方法。

确认 vault 路径

你需要知道你的 Obsidian vault 的绝对路径。在整个代码中，我们假设 vault 在 W:\obsidian\Work——你需要替换成你自己的路径。

7.3 第一步：索引你的 Obsidian Vault

🎨 漫画图解：〈索引车如何认识整个 Vault〉

这是 RAG 的「离线阶段」——把你的全部笔记变成可搜索的向量索引。

7.3.1 读取所有笔记

import os
import frontmatter
from pathlib import Path

VAULT_PATH = r"W:\obsidian\Work"  # 替换成你的 vault 路径

def load_all_notes(vault_path: str) -> list[dict]:
    """扫描 vault 中所有 .md 文件，返回笔记列表"""
    notes = []
    vault = Path(vault_path)

    for md_file in vault.rglob("*.md"):
        # 跳过 .obsidian 配置目录
        if ".obsidian" in md_file.parts:
            continue

        try:
            # 用 python-frontmatter 解析 YAML frontmatter + 正文
            post = frontmatter.load(str(md_file))

            notes.append({
                "path": str(md_file.relative_to(vault)),
                "title": post.get("title", md_file.stem),
                "tags": post.get("tags", []),
                "content": post.content,
                # 保留 frontmatter 中的其他有用字段
                "metadata": {k: v for k, v in post.metadata.items()
                             if k not in ("title", "tags")}
            })
        except Exception as e:
            print(f"⚠️ 跳过无法解析的文件：{md_file} —— {e}")

    return notes

# 测试
notes = load_all_notes(VAULT_PATH)
print(f"✅ 共找到 {len(notes)} 篇笔记")
for n in notes[:3]:
    print(f"  📄 {n['path']} —— 标题：{n['title']}")

7.3.2 文档切片

import tiktoken

def chunk_text(text: str, max_tokens: int = 400, overlap: int = 50) -> list[str]:
    """
    把长文本切成重叠的小段。

    参数：
    - max_tokens：每段最多多少 token
    - overlap：相邻段之间重叠多少 token

    返回：切片后的段落列表
    """
    encoding = tiktoken.get_encoding("cl100k_base")  # OpenAI 的分词器
    tokens = encoding.encode(text)

    chunks = []
    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunk_tokens = tokens[start:end]
        chunks.append(encoding.decode(chunk_tokens))
        start += (max_tokens - overlap)  # 下一段的起始位置，留 overlap 重叠

    return chunks

# 测试
sample_text = notes[0]["content"]
chunks = chunk_text(sample_text)
print(f"原文 {len(sample_text)} 字 → {len(chunks)} 个切片")
print(f"第1段前100字：{chunks[0][:100]}...")

为什么用 tiktoken 而不用字符数做切片？ 因为 LLM 是按 token 计费、按 token 计上下文的。用 token 数控制切片大小比用字符数精确得多。中文一个 token 约等于 1-2 个汉字，英文一个 token 约等于 0.75 个单词。

7.3.3 向量化并存入 Chroma

from openai import OpenAI
import chromadb
import uuid

# 初始化客户端
openai_client = OpenAI()  # 自动读取 OPENAI_API_KEY 环境变量
chroma_client = chromadb.PersistentClient(path="./obsidian_vectordb")

# 创建或获取集合
collection = chroma_client.get_or_create_collection(
    name="obsidian_notes",
    metadata={"hnsw:space": "cosine"}  # 用余弦相似度
)

def build_index(notes: list[dict], batch_size: int = 20):
    """把所有笔记切片、向量化、存入 Chroma"""
    total_chunks = 0

    for note in notes:
        # 跳过太短的笔记
        if len(note["content"].strip()) < 50:
            continue

        chunks = chunk_text(note["content"])

        for i, chunk in enumerate(chunks):
            try:
                # 调用 OpenAI Embedding API
                response = openai_client.embeddings.create(
                    model="text-embedding-3-small",
                    input=chunk
                )
                embedding = response.data[0].embedding

                # 存入 Chroma
                chunk_id = str(uuid.uuid4())
                collection.add(
                    embeddings=[embedding],
                    documents=[chunk],
                    ids=[chunk_id],
                    metadatas=[{
                        "source_path": note["path"],
                        "source_title": note["title"],
                        "chunk_index": i,
                        "total_chunks": len(chunks),
                        "tags": ", ".join(note["tags"]) if note["tags"] else ""
                    }]
                )
                total_chunks += 1

            except Exception as e:
                print(f"⚠️ 向量化失败：{note['path']} 第{i}段 —— {e}")

        if total_chunks % 50 == 0:
            print(f"  已索引 {total_chunks} 个切片...")

    print(f"✅ 索引完成！共 {total_chunks} 个切片，来自 {len(notes)} 篇笔记")

# 执行索引（这一行会花几分钟）
build_index(notes)

注意：调用 OpenAI Embedding API 是收费的。text-embedding-3-small 的价格是 $0.02 / 1M tokens。如果你的 vault 有 1000 篇笔记、每篇平均 2000 字（约 1500 token），总索引成本大约是：

几乎可以忽略不计。

7.4 第二步：定义 Agent 的工具

🎨 漫画图解：〈Obsidian Agent 的四件工具〉

回顾第4章——工具定义的质量决定 Agent 的表现。我们要定义四个工具：

工具一：语义搜索笔记

def search_notes(query: str, limit: int = 5) -> list[dict]:
    """
    在 Obsidian vault 中用自然语言搜索最相关的笔记段落。

    参数：
    - query：自然语言查询，例如「机器学习中的过拟合是什么意思」
    - limit：返回结果数量，默认 5，最大 10

    返回：相关段落的列表，每项包含内容、来源笔记、相似度分数
    """
    # 1. 把查询向量化
    query_embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    ).data[0].embedding

    # 2. 在 Chroma 中搜索
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=min(limit, 10),
        include=["documents", "metadatas", "distances"]
    )

    # 3. 整理结果
    formatted = []
    for i in range(len(results["documents"][0])):
        formatted.append({
            "content": results["documents"][0][i],
            "source": results["metadatas"][0][i]["source_path"],
            "title": results["metadatas"][0][i]["source_title"],
            "relevance": round(1 - results["distances"][0][i], 3)  # 距离 → 相似度
        })

    return formatted

工具二：读取一篇笔记的完整内容

def read_note(file_path: str) -> dict:
    """
    读取一篇笔记的完整内容。

    参数：
    - file_path：笔记的相对路径，例如 '蒸馏术/AI 科普/第1章：AI 到底是个啥.md'

    返回：笔记的标题、标签、完整正文
    """
    full_path = Path(VAULT_PATH) / file_path

    # 安全检查：确保路径在 vault 内
    if not str(full_path.resolve()).startswith(str(Path(VAULT_PATH).resolve())):
        return {"error": "路径越界——不能读取 vault 之外的文件"}

    if not full_path.exists():
        return {"error": f"文件不存在：{file_path}"}

    try:
        post = frontmatter.load(str(full_path))
        return {
            "title": post.get("title", full_path.stem),
            "tags": post.get("tags", []),
            "content": post.content,
            "path": file_path
        }
    except Exception as e:
        return {"error": f"读取失败：{e}"}

工具三：创建一篇新笔记

def create_note(file_path: str, title: str, content: str, tags: list[str] = None) -> dict:
    """
    在 vault 中创建一篇新的 Markdown 笔记。

    参数：
    - file_path：新笔记的相对路径，例如 'notes/新想法.md'
    - title：笔记标题
    - content：笔记正文（Markdown 格式）
    - tags：可选的标签列表

    返回：创建结果
    """
    full_path = Path(VAULT_PATH) / file_path

    # 安全检查
    if not str(full_path.resolve()).startswith(str(Path(VAULT_PATH).resolve())):
        return {"error": "路径越界——不能在 vault 之外创建文件"}

    if full_path.exists():
        return {"error": f"文件已存在：{file_path}。如需覆盖请手动删除后再试。"}

    # 组装 YAML frontmatter
    from datetime import date
    import yaml

    frontmatter_data = {
        "title": title,
        "date": str(date.today()),
        "tags": tags or [],
        "created_by": "Obsidian Agent"
    }

    yaml_header = yaml.dump(frontmatter_data, allow_unicode=True, sort_keys=False)

    # 写入文件
    full_content = f"---\n{yaml_header}---\n\n{content}\n"

    # 确保父目录存在
    full_path.parent.mkdir(parents=True, exist_ok=True)

    full_path.write_text(full_content, encoding="utf-8")

    return {
        "success": True,
        "path": file_path,
        "title": title,
        "message": f"笔记已创建：{file_path}"
    }

工具四：发现潜在关联

def find_connections(note_path: str, limit: int = 5) -> list[dict]:
    """
    找到与指定笔记内容最相似、但尚未被链接的其他笔记。

    参数：
    - note_path：源笔记的相对路径
    - limit：返回的关联笔记数量

    返回：关联笔记列表
    """
    # 读取目标笔记的全部内容
    note_data = read_note(note_path)
    if "error" in note_data:
        return [note_data]

    # 用全文做向量搜索（排除自己）
    query_embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=note_data["content"][:8000]  # 取前 8000 字符（Embedding API 有长度限制）
    ).data[0].embedding

    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=limit + 1,  # +1 因为可能命中自己
        include=["documents", "metadatas", "distances"]
    )

    connections = []
    for i in range(len(results["documents"][0])):
        source = results["metadatas"][0][i]["source_path"]
        if source == note_path:
            continue  # 跳过自己
        connections.append({
            "source": source,
            "title": results["metadatas"][0][i]["source_title"],
            "snippet": results["documents"][0][i][:200] + "...",
            "relevance": round(1 - results["distances"][0][i], 3)
        })

    return connections[:limit]

7.5 第三步：实现 Agent 循环

🎨 漫画图解：〈Agent 循环控制室〉

这是整个 Agent 的「心脏」。回顾第2章的 Sense-Think-Act 循环——现在把它变成代码。

7.5.1 工具定义（给 LLM 看的 JSON Schema）

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_notes",
            "description": "在 Obsidian vault 中搜索笔记内容。当你需要查找某个主题、概念或信息时使用。适合「介绍一下XX」「XX是什么」「找关于XX的笔记」这类问题。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "自然语言搜索查询，用中文描述你想找什么"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回结果数量，默认5",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_note",
            "description": "读取一篇笔记的完整内容。当用户想查看某篇笔记的全文时使用。比如「打开XX笔记」「看看XX那篇写了什么」。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "笔记的相对路径，例如 '蒸馏术/AI 科普/第1章：AI 到底是个啥.md'"
                    }
                },
                "required": ["file_path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_note",
            "description": "在 Obsidian vault 中创建一篇新笔记。当用户说「记下来」「创建笔记」「新建一篇」时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {
                        "type": "string",
                        "description": "新笔记的文件路径，例如 'notes/学习记录/Agent 学习心得.md'"
                    },
                    "title": {
                        "type": "string",
                        "description": "笔记标题"
                    },
                    "content": {
                        "type": "string",
                        "description": "笔记正文，Markdown 格式"
                    },
                    "tags": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "可选的标签列表"
                    }
                },
                "required": ["file_path", "title", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "find_connections",
            "description": "找到与指定笔记内容相关、但尚未链接的其他笔记。用于发现知识库中的潜在关联。",
            "parameters": {
                "type": "object",
                "properties": {
                    "note_path": {
                        "type": "string",
                        "description": "源笔记的相对路径"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回数量，默认5",
                        "default": 5
                    }
                },
                "required": ["note_path"]
            }
        }
    }
]

7.5.2 工具调度器

import json

# 工具名称 → 实际函数的映射
TOOL_MAP = {
    "search_notes": search_notes,
    "read_note": read_note,
    "create_note": create_note,
    "find_connections": find_connections
}

def execute_tool(tool_name: str, arguments: dict) -> str:
    """执行工具调用，返回格式化的结果字符串"""
    if tool_name not in TOOL_MAP:
        return f"❌ 未知工具：{tool_name}"

    try:
        result = TOOL_MAP[tool_name](**arguments)
        return json.dumps(result, ensure_ascii=False, indent=2)
    except Exception as e:
        return f"❌ 工具执行失败：{e}"

7.5.3 Agent 主循环

def run_agent(user_input: str, max_turns: int = 10, verbose: bool = True):
    """
    Obsidian Agent 的主循环。

    参数：
    - user_input：用户的问题或指令
    - max_turns：最大工具调用轮次（防止死循环）
    - verbose：是否打印详细的思考过程
    """

    # 初始化对话历史
    messages = [
        {
            "role": "system",
            "content": """你是 Obsidian Agent——一个运行在用户 Obsidian vault 中的 AI 助手。

你的能力：
1. 搜索笔记内容（search_notes）：找到用户 vault 中相关的笔记段落
2. 阅读笔记全文（read_note）：打开并阅读任意一篇笔记
3. 创建新笔记（create_note）：在 vault 中创建新的 Markdown 笔记
4. 发现笔记关联（find_connections）：找到可能相关但未链接的笔记

行为准则：
- 优先使用 search_notes 查找 vault 中的信息，而不是凭自己的训练数据回答
- 如果 vault 中有相关信息，基于 vault 内容回答并引用来源
- 如果 vault 中没有相关信息，诚实告知用户
- 创建笔记前，确保 file_path 以 .md 结尾且路径合理
- 用中文回答，保持清晰、简洁、有帮助
- 引用笔记时使用 标题 格式"""
        },
        {"role": "user", "content": user_input}
    ]

    turn = 0

    while turn < max_turns:
        turn += 1

        if verbose:
            print(f"\n{'='*50}")
            print(f"🔄 第 {turn} 轮")

        # 调用 LLM
        response = openai_client.chat.completions.create(
            model="gpt-4o-mini",  # 便宜、够用、Function Calling 稳定
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",  # LLM 自己决定是否调工具
            temperature=0.3      # 低温度 = 更确定性的行为
        )

        assistant_msg = response.choices[0].message

        # 情况1：LLM 决定调用工具
        if assistant_msg.tool_calls:
            # 先把 LLM 的工具调用请求加入对话历史
            messages.append({
                "role": "assistant",
                "content": assistant_msg.content,
                "tool_calls": [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {
                            "name": tc.function.name,
                            "arguments": tc.function.arguments
                        }
                    }
                    for tc in assistant_msg.tool_calls
                ]
            })

            for tc in assistant_msg.tool_calls:
                tool_name = tc.function.name
                arguments = json.loads(tc.function.arguments)

                if verbose:
                    print(f"🔧 调用工具：{tool_name}")
                    print(f"   参数：{json.dumps(arguments, ensure_ascii=False)}")

                # 执行工具
                result = execute_tool(tool_name, arguments)

                if verbose:
                    preview = result[:300] + "..." if len(result) > 300 else result
                    print(f"📥 结果：{preview}")

                # 把工具返回结果加入对话历史
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result
                })

            # 工具执行完毕，继续循环——LLM 会在下一轮看到结果并决定下一步

        # 情况2：LLM 直接回复（任务完成或无需工具）
        else:
            if verbose:
                print(f"💬 Agent 最终回复：")

            return assistant_msg.content

    # 超过最大轮次
    return "⚠️ Agent 达到最大轮次限制，任务可能未完成。请尝试简化你的请求。"


# ========== 运行 ==========

if __name__ == "__main__":
    # 先确保索引已构建（只需运行一次）
    # notes = load_all_notes(VAULT_PATH)
    # build_index(notes)

    print("🤖 Obsidian Agent 已就绪")
    print("你的 vault 中有 " + str(collection.count()) + " 个索引切片")
    print("输入 'quit' 退出\n")

    while True:
        user_input = input("\n👤 你：")
        if user_input.lower() in ("quit", "exit", "q"):
            print("👋 再见！")
            break

        response = run_agent(user_input)
        print(f"\n🤖 Agent：{response}")

7.6 第四步：安全与限制

🎨 漫画图解：〈Vault 城墙与安全门卫〉

Agent 能写文件，所以必须有安全机制。

7.6.1 路径越界防护

def safe_path(relative_path: str) -> Path:
    """
    确保路径在 vault 内。
    如果路径试图越界（例如 ../../etc/passwd），抛出异常。
    """
    vault = Path(VAULT_PATH).resolve()
    target = (vault / relative_path).resolve()

    if not str(target).startswith(str(vault)):
        raise ValueError(f"🚫 安全拦截：路径越界 —— {relative_path}")

    return target

把这个函数用在 read_note 和 create_note 的开头——它已经在我们的实现中了。

7.6.2 禁止操作名单

FORBIDDEN_PATTERNS = [
    ".obsidian",     # Obsidian 配置目录
    ".git",          # 版本控制
    ".trash",        # 系统回收站
    "node_modules",  # 依赖
]

def is_forbidden(path: str) -> bool:
    """检查路径是否包含禁止操作的模式"""
    return any(pattern in path for pattern in FORBIDDEN_PATTERNS)

7.6.3 最大轮次限制

run_agent 的 max_turns 参数已经实现了这个——默认 10 轮，超过即终止，防止死循环烧钱。

7.6.4 操作日志

import logging

logging.basicConfig(
    filename="agent_operations.log",
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s"
)

def log_operation(action: str, details: str):
    """记录每一次 Agent 操作"""
    logging.info(f"{action} | {details}")

在 execute_tool 中加上 log_operation(tool_name, str(arguments))，这样每次出问题你都能追溯。

7.7 第五步：运行你的 Agent

完整启动流程

# main.py —— 完整入口脚本

import os
from pathlib import Path

# ===== 配置 =====
VAULT_PATH = r"W:\obsidian\Work"  # ← 改成你的 vault 路径
API_KEY = os.getenv("OPENAI_API_KEY")

if not API_KEY:
    print("❌ 请先设置 OPENAI_API_KEY 环境变量")
    print("   PowerShell: $env:OPENAI_API_KEY = 'sk-...'")
    print("   Mac/Linux:  export OPENAI_API_KEY='sk-...'")
    exit(1)

# ===== 初始化 =====
from openai import OpenAI
import chromadb

openai_client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./obsidian_vectordb")
collection = chroma_client.get_or_create_collection(
    name="obsidian_notes",
    metadata={"hnsw:space": "cosine"}
)

# ===== 检查是否需要建索引 =====
if collection.count() == 0:
    print("📂 首次运行，正在索引 vault...")
    notes = load_all_notes(VAULT_PATH)
    build_index(notes)
else:
    print(f"📂 索引已存在，共 {collection.count()} 个切片")

# ===== 启动交互 =====
print("\n" + "="*50)
print("🤖 Obsidian Agent 已就绪！")
print("="*50)
print("你可以这样问我：")
print("  • 「什么是 RAG？」（会搜索你的笔记）")
print("  • 「帮我整理一下 AI 科普系列的核心概念」")
print("  • 「把我刚才说的这些记录到新笔记里」")
print("  • 「找一下和 [[某篇笔记]] 相关的其他笔记」")
print("\n输入 'quit' 退出\n")

while True:
    try:
        user_input = input("\n👤 你：")
        if user_input.lower() in ("quit", "exit", "q"):
            break

        response = run_agent(user_input, verbose=False)
        print(f"\n🤖 Agent：{response}")

    except KeyboardInterrupt:
        print("\n👋 再见！")
        break
    except Exception as e:
        print(f"\n❌ 出错了：{e}")

运行效果示例

👤 你：什么是RAG？

🤖 Agent：根据你 vault 中的笔记，RAG（检索增强生成）是一种给 LLM
装上外部知识库的技术。你的笔记 [[蒸馏术/AI Agent/第3章：RAG——给 Agent
装上外部知识库]] 中有详细解释：

简单来说，RAG 就像让 LLM 从「闭卷考试」变成「开卷考试」——它会先从
你的知识库里找到相关资料，再基于资料回答，而不是凭训练记忆回答。

核心流程：文档切片 → 向量化 → 存入向量数据库 → 用户提问时检索 →
把检索结果和问题一起给 LLM → 生成有据可查的答案。

需要我打开那篇笔记看完整内容吗？

👤 你：帮我总结一下我这周关于AI的学习，记录到新笔记里

🤖 Agent：好的，我先搜索你最近的笔记...（搜索中）

我找到了几篇本周创建的 AI 相关笔记：
1. [[蒸馏术/AI Agent/第1章：Agent 到底是什么]]
2. [[蒸馏术/AI Agent/第2章：大脑、手脚与记忆]]
3. [[蒸馏术/AI Agent/第3章：RAG——给 Agent 装上外部知识库]]

我已经创建了一篇总结笔记：[[weekly/AI Agent 学习总结-2026-07-22]]

内容包括：
- Agent 的核心定义（LLM + 行动循环）
- Agent 架构（大脑、规划、记忆、工具）
- RAG 的五个步骤
- 你下一步该读什么

打开看看？

7.8 完整代码

把以下所有代码保存为 main.py，放到 obsidian-agent/ 目录下，修改 VAULT_PATH 后直接 python main.py 即可运行。

"""
Obsidian Agent —— 一个运行在你本地 Obsidian vault 上的 AI 助手
前置依赖：pip install chromadb openai python-frontmatter tiktoken pyyaml
使用前：设置 OPENAI_API_KEY 环境变量
"""

import os
import json
import uuid
import logging
from pathlib import Path
from datetime import date

import frontmatter
import tiktoken
import yaml
from openai import OpenAI
import chromadb

# ============================================================
# 配置
# ============================================================

VAULT_PATH = r"W:\obsidian\Work"  # ← 改成你的 vault 路径
MAX_TURNS = 10                     # 最大工具调用轮次
CHUNK_SIZE = 400                   # 切片大小（token）
CHUNK_OVERLAP = 50                 # 切片重叠（token）
EMBEDDING_MODEL = "text-embedding-3-small"
LLM_MODEL = "gpt-4o-mini"

# ============================================================
# 日志
# ============================================================

logging.basicConfig(
    filename="agent_operations.log",
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s"
)

# ============================================================
# 客户端初始化
# ============================================================

openai_client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./obsidian_vectordb")
collection = chroma_client.get_or_create_collection(
    name="obsidian_notes",
    metadata={"hnsw:space": "cosine"}
)

# ============================================================
# 第一步：索引 Vault
# ============================================================

def load_all_notes(vault_path: str) -> list[dict]:
    """扫描 vault 中所有 .md 文件"""
    notes = []
    vault = Path(vault_path)
    for md_file in vault.rglob("*.md"):
        if ".obsidian" in md_file.parts:
            continue
        try:
            post = frontmatter.load(str(md_file))
            notes.append({
                "path": str(md_file.relative_to(vault)),
                "title": post.get("title", md_file.stem),
                "tags": post.get("tags", []),
                "content": post.content,
                "metadata": {k: v for k, v in post.metadata.items()
                             if k not in ("title", "tags")}
            })
        except Exception as e:
            print(f"⚠️ 跳过：{md_file} —— {e}")
    return notes


def chunk_text(text: str, max_tokens: int = CHUNK_SIZE,
               overlap: int = CHUNK_OVERLAP) -> list[str]:
    """把长文本切成重叠的小段"""
    encoding = tiktoken.get_encoding("cl100k_base")
    tokens = encoding.encode(text)
    chunks = []
    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunk_tokens = tokens[start:end]
        chunks.append(encoding.decode(chunk_tokens))
        start += (max_tokens - overlap)
    return chunks


def build_index(notes: list[dict]):
    """把所有笔记切片、向量化、存入 Chroma"""
    total = 0
    for note in notes:
        if len(note["content"].strip()) < 50:
            continue
        chunks = chunk_text(note["content"])
        for i, chunk in enumerate(chunks):
            try:
                response = openai_client.embeddings.create(
                    model=EMBEDDING_MODEL, input=chunk
                )
                embedding = response.data[0].embedding
                collection.add(
                    embeddings=[embedding],
                    documents=[chunk],
                    ids=[str(uuid.uuid4())],
                    metadatas=[{
                        "source_path": note["path"],
                        "source_title": note["title"],
                        "chunk_index": i,
                        "total_chunks": len(chunks),
                        "tags": ", ".join(note["tags"]) if note["tags"] else ""
                    }]
                )
                total += 1
            except Exception as e:
                print(f"⚠️ 向量化失败：{note['path']}#{i} —— {e}")
        if total % 50 == 0:
            print(f"  已索引 {total} 个切片...")
    print(f"✅ 索引完成！{total} 个切片，来自 {len(notes)} 篇笔记")


# ============================================================
# 第二步 + 第三步：工具定义 + Agent 循环
# ============================================================

def safe_path(relative_path: str) -> Path:
    """确保路径在 vault 内"""
    vault = Path(VAULT_PATH).resolve()
    target = (vault / relative_path).resolve()
    if not str(target).startswith(str(vault)):
        raise ValueError(f"🚫 路径越界：{relative_path}")
    return target


def search_notes(query: str, limit: int = 5) -> list[dict]:
    """语义搜索笔记"""
    query_embedding = openai_client.embeddings.create(
        model=EMBEDDING_MODEL, input=query
    ).data[0].embedding
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=min(limit, 10),
        include=["documents", "metadatas", "distances"]
    )
    formatted = []
    for i in range(len(results["documents"][0])):
        formatted.append({
            "content": results["documents"][0][i],
            "source": results["metadatas"][0][i]["source_path"],
            "title": results["metadatas"][0][i]["source_title"],
            "relevance": round(1 - results["distances"][0][i], 3)
        })
    return formatted


def read_note(file_path: str) -> dict:
    """读取一篇笔记"""
    full_path = safe_path(file_path)
    if not full_path.exists():
        return {"error": f"文件不存在：{file_path}"}
    try:
        post = frontmatter.load(str(full_path))
        return {
            "title": post.get("title", full_path.stem),
            "tags": post.get("tags", []),
            "content": post.content,
            "path": file_path
        }
    except Exception as e:
        return {"error": f"读取失败：{e}"}


def create_note(file_path: str, title: str, content: str,
                tags: list[str] = None) -> dict:
    """创建新笔记"""
    full_path = safe_path(file_path)
    if full_path.exists():
        return {"error": f"文件已存在：{file_path}"}
    fm = {
        "title": title,
        "date": str(date.today()),
        "tags": tags or [],
        "created_by": "Obsidian Agent"
    }
    yaml_header = yaml.dump(fm, allow_unicode=True, sort_keys=False)
    full_content = f"---\n{yaml_header}---\n\n{content}\n"
    full_path.parent.mkdir(parents=True, exist_ok=True)
    full_path.write_text(full_content, encoding="utf-8")
    logging.info(f"CREATE | {file_path}")
    return {"success": True, "path": file_path, "title": title}


def find_connections(note_path: str, limit: int = 5) -> list[dict]:
    """发现关联笔记"""
    note_data = read_note(note_path)
    if "error" in note_data:
        return [note_data]
    query_embedding = openai_client.embeddings.create(
        model=EMBEDDING_MODEL, input=note_data["content"][:8000]
    ).data[0].embedding
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=limit + 1,
        include=["documents", "metadatas", "distances"]
    )
    connections = []
    for i in range(len(results["documents"][0])):
        source = results["metadatas"][0][i]["source_path"]
        if source == note_path:
            continue
        connections.append({
            "source": source,
            "title": results["metadatas"][0][i]["source_title"],
            "snippet": results["documents"][0][i][:200] + "...",
            "relevance": round(1 - results["distances"][0][i], 3)
        })
    return connections[:limit]


TOOL_MAP = {
    "search_notes": search_notes,
    "read_note": read_note,
    "create_note": create_note,
    "find_connections": find_connections
}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_notes",
            "description": "在 Obsidian vault 中搜索笔记内容。当需要查找某个主题、概念或信息时使用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "自然语言搜索查询"},
                    "limit": {"type": "integer", "description": "返回数量，默认5", "default": 5}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_note",
            "description": "读取一篇笔记的完整内容。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "笔记的相对路径，例如 'notes/学习.md'"}
                },
                "required": ["file_path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_note",
            "description": "在 Obsidian vault 中创建新笔记。",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "文件路径，以 .md 结尾"},
                    "title": {"type": "string", "description": "笔记标题"},
                    "content": {"type": "string", "description": "Markdown 正文"},
                    "tags": {"type": "array", "items": {"type": "string"}, "description": "标签列表"}
                },
                "required": ["file_path", "title", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "find_connections",
            "description": "找到与指定笔记内容相关的其他笔记。",
            "parameters": {
                "type": "object",
                "properties": {
                    "note_path": {"type": "string", "description": "源笔记路径"},
                    "limit": {"type": "integer", "description": "返回数量", "default": 5}
                },
                "required": ["note_path"]
            }
        }
    }
]


def execute_tool(tool_name: str, arguments: dict) -> str:
    """执行工具并返回结果"""
    if tool_name not in TOOL_MAP:
        return json.dumps({"error": f"未知工具：{tool_name}"})
    try:
        result = TOOL_MAP[tool_name](**arguments)
        logging.info(f"TOOL | {tool_name} | {json.dumps(arguments, ensure_ascii=False)}")
        return json.dumps(result, ensure_ascii=False, indent=2)
    except Exception as e:
        logging.error(f"TOOL_ERROR | {tool_name} | {e}")
        return json.dumps({"error": str(e)})


def run_agent(user_input: str, max_turns: int = MAX_TURNS,
              verbose: bool = True) -> str:
    """Agent 主循环"""
    messages = [
        {
            "role": "system",
            "content": """你是 Obsidian Agent——运行在用户 Obsidian vault 中的 AI 助手。

你的能力：搜索笔记(search_notes)、阅读笔记(read_note)、创建笔记(create_note)、发现关联(find_connections)。

行为准则：
- 优先用 search_notes 查 vault，不要凭训练数据回答
- vault 中有相关信息就引用来源（用 标题 格式）
- vault 中没有就诚实告知
- 创建笔记时确保路径以 .md 结尾
- 用中文回答，保持清晰简洁"""
        },
        {"role": "user", "content": user_input}
    ]

    for turn in range(max_turns):
        if verbose:
            print(f"\n--- 第 {turn+1} 轮 ---")

        response = openai_client.chat.completions.create(
            model=LLM_MODEL,
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
            temperature=0.3
        )

        msg = response.choices[0].message

        if msg.tool_calls:
            messages.append({
                "role": "assistant",
                "content": msg.content,
                "tool_calls": [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {
                            "name": tc.function.name,
                            "arguments": tc.function.arguments
                        }
                    }
                    for tc in msg.tool_calls
                ]
            })

            for tc in msg.tool_calls:
                name = tc.function.name
                args = json.loads(tc.function.arguments)
                if verbose:
                    print(f"🔧 {name}({json.dumps(args, ensure_ascii=False)})")
                result = execute_tool(name, args)
                if verbose:
                    print(f"📥 {result[:200]}...")
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result
                })
        else:
            return msg.content

    return "⚠️ 达到最大轮次限制，请简化请求。"


# ============================================================
# 主程序
# ============================================================

if __name__ == "__main__":
    # 检查 API Key
    if not os.getenv("OPENAI_API_KEY"):
        print("❌ 请设置 OPENAI_API_KEY 环境变量")
        exit(1)

    # 建索引（首次运行）
    if collection.count() == 0:
        print("📂 首次运行，正在索引 vault...")
        notes = load_all_notes(VAULT_PATH)
        build_index(notes)
    else:
        print(f"📂 索引就绪：{collection.count()} 个切片")

    print("\n" + "="*50)
    print("🤖 Obsidian Agent 已就绪！")
    print("="*50)
    print("试试：「什么是 RAG？」「帮我整理 AI 科普的核心概念」")
    print("输入 quit 退出\n")

    while True:
        try:
            ui = input("\n👤 你：")
            if ui.lower() in ("quit", "exit", "q"):
                break
            resp = run_agent(ui, verbose=False)
            print(f"\n🤖 Agent：{resp}")
        except KeyboardInterrupt:
            print("\n👋")
            break
        except Exception as e:
            print(f"\n❌ {e}")

7.9 扩展方向

你已经有了一个能跑的 Obsidian Agent。下面是几个你可以自己加的扩展：

扩展一：使用本地模型（省钱 + 隐私）

把 OpenAI 替换成 Ollama 本地模型：

# 只需改两行
from openai import OpenAI

local_client = OpenAI(
    base_url="http://localhost:11434/v1",  # Ollama 默认地址
    api_key="ollama"                        # Ollama 不需要真实 key
)

# Embedding 也用本地的
# pip install sentence-transformers
from sentence_transformers import SentenceTransformer
local_embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")

扩展二：增量索引

每次 Agent 启动时只索引新增/修改的笔记，而不是每次都重建：

def incremental_index(vault_path: str):
    """只索引自上次运行以来新增或修改的笔记"""
    existing_paths = set(
        collection.get(include=["metadatas"])["metadatas"] or []
    )
    # 比较文件修改时间，只处理更新的...

扩展三：多 Agent 协作

把第6章学的多 Agent 模式用上——研究 Agent + 写作 Agent + 审校 Agent：

def multi_agent_report(topic: str):
    """三个 Agent 协作写报告"""
    research = run_agent(f"深度研究：{topic}，整理资料", verbose=False)
    draft = run_agent(f"基于以下研究，写一篇报告：\n{research}", verbose=False)
    review = run_agent(f"审校这篇报告，给出修改意见：\n{draft}", verbose=False)
    return f"研究报告\n{draft}\n\n审校意见\n{review}"

扩展四：MCP 接入

让你的 Obsidian Agent 通过 MCP 连接更多工具——Gmail、日历、GitHub：

# 使用 mcp Python SDK
from mcp import ClientSession, StdioServerParameters

# 连接到一个 MCP 文件系统服务
server_params = StdioServerParameters(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", VAULT_PATH]
)

🎉 恭喜

你从零构建了一个完整的 AI Agent。

回顾一下你做了什么：

1. 建了一个 RAG 知识库索引你的全部笔记（第3章的知识）
2. 定义了四个工具并写了 Function Calling 的 JSON Schema（第4章的知识）
3. 实现了 Agent 的 Sense-Think-Act 主循环（第2章的知识）
4. 加上了路径越界防护和操作日志（安全策略）
5. 整个系统只用了约 300 行 Python（架构简洁的力量）

更重要的是：你现在不只是「会用 AI」——你理解了 Agent 的全部底层机制。RAG 不再是一个模糊的概念，你可以给别人画出从「文档切片」到「向量检索」再到「增强生成」的完整流程图。Function Calling 不再神秘——你知道 LLM 只是「说出」它想调什么，真正执行的是宿主程序。

当有人跟你说「我们公司在用 Agent」的时候，你脑子里不再是「哦，就是那个 AI 对话机器人」——而是「你们的 Agent 用什么架构？ReAct 还是 Plan-and-Solve？RAG 用的是什么 Embedding 模型？向量数据库选的 Chroma 还是 Pinecone？工具调用走 MCP 吗？」

这就是从「AI 用户」到「AI 工程师」的转变。

本章小结