如何從 0 到 1 開發一個 AI Agent

![文章頂部.png](https://i.imgur.com/vxOt3QH.jpeg)![星炎 (1).png](https://i.imgur.com/h7HWb76.jpeg)

如何從 0 到 1 開發一個 AI Agent

引言

在日常生活中、工作中，從 openClaw 到各種 Code Agent，我們接觸了各式各樣的 AI Agent，這也是 AI 大模型在日常應用中最主要的應用形式。

早期的 AI 應用，大多停留在「提示詞調優」的階段——透過優化單次對話的 Prompt，讓大模型完成特定任務。但當我們試圖讓 AI 真正落地到複雜業務場景時，慢慢發現，只透過對 Prompt 的優化很快就遇到了瓶頸：工具呼叫頻繁出錯、多任務協同混亂、上下文溢出導致幻覺等問題層出不窮。

針對這些問題，慢慢衍生出一些標準的工程化基礎設施如 MCP、SKILLS 以及 Context Engineer，用於支撐 AI Agent 盡可能穩定運行，盡可能可靠地處理複雜業務場景的問題。

本文首先會對衍生出的標準工程化基礎設施進行簡要介紹，然後透過實作一個簡化版的個人助理 Agent，來看看 MCP、SKILLS 以及 Context Engineer 是如何支撐 Agent 穩定運行的。

前置介紹

開始實作 Agent 之前，首先對相關概念做一個基本介紹，由於本文最終目標是實作一個簡化版個人助理 Agent，不會對相關概念做深入解析，這裡只做簡單介紹，想了解更多，可自行查詢對應的文件。

1.1 MCP：統一的工具呼叫協定層

MCP（Model Context Protocol，模型上下文協定）是由 Anthropic 公司在 2024 年 11 月 25 日開源發布，其核心定位，是為大模型與工具之間提供一套統一的、上下文感知的通訊協定。它向下相容各類 CLI 工具、API 服務、系統能力，向上給 Agent 提供標準化的工具呼叫介面，解決了工具碎片化、適配成本高、呼叫不穩定的核心問題。

其核心能力價值體現在三個方面：

1. 標準化的工具定義與互動規範：MCP 定義了統一的工具中繼資料格式、參數校驗規則、回傳結構與錯誤處理機制，Agent 無需針對單個工具做客製化適配，只需遵循 MCP 協定即可完成呼叫，大幅提升工具呼叫的成功率與穩定性；
2. 原生的上下文感知能力：不同於傳統工具呼叫的靜態參數傳遞，MCP 可以將 Agent 的會話上下文、歷史執行記錄、使用者權限等資訊同步給工具，讓工具可以結合上下文動態調整執行邏輯，實現更智慧的互動；
3. 分層的安全與權限管控：MCP 內建了權限隔離、呼叫稽核、沙箱執行等能力，開發者可以精細化管控 Agent 對工具的存取權限，限制高風險操作，同時完整記錄所有呼叫日誌，解決了原生 CLI 的安全風險問題。

1.2 SKILLS：面向業務的工具鏈封裝

SKILLS 的本質，是將多個原子工具、固定的業務流程、Prompt 邏輯、錯誤處理機制打包封裝成可重用的業務能力單元。比如「使用者行為資料分析」技能，就可以封裝「SQL 資料查詢 → 資料清洗 → 指標計算 → 視覺化生成 → 報告輸出」的全流程，整合多個原子工具，內建對應的業務邏輯與例外處理方案。

它解決了 Agent 工程化中的兩個核心痛點：

• 解決工具呼叫的碎片化問題：無需讓大模型每次都從零拆解任務、逐個呼叫工具，大幅降低大模型的推理壓力，減少任務拆解錯誤與工具呼叫幻覺；
• 實現業務能力的可重用：封裝好的 SKILLS 可以在不同 Agent、不同場景中直接重用，大幅降低 Agent 的開發成本，同時保證業務邏輯的一致性。

二者的協同關係非常清晰：MCP 提供標準化的協定與管控層，SKILLS 基於前兩者完成面向業務的能力封裝，共同構成 Agent 完整的行動體系，讓 Agent 真正具備「動手做事」的能力。

1.3 Context Engineer——貫穿 Agent 全生命週期的神經系統

在 AI Agent 的落地實踐中，有一個被絕大多數人忽視的真相：80% 的 Agent 故障，都不是大模型能力不足，而是上下文管理混亂導致的。

不該傳給大模型的冗餘資訊塞滿了上下文視窗，導致核心資訊被截斷；該同步給工具/子 Agent 的關鍵上下文沒有傳遞，導致執行錯誤；歷史執行記錄沒有被有效召回，導致重複犯錯；敏感資訊沒有去識別化，導致資料外洩……基於這些問題，它需要一套完整的、貫穿 Agent 全生命週期的工程化體系——這就是 Context Engineer（上下文工程）。

1.3.1 Context Engineer 與 Prompt Engineering 的核心差異

在 AI 發展早期，更被人熟知的是 Prompt Engineer（提示詞工程），其目標是透過設計、優化、除錯 Prompt（提示詞），讓 AI 模型穩定輸出符合業務預期的結果。

很多人會把 Context Engineer 和 Prompt Engineering 混為一談，但二者有著本質的差異：

• Prompt Engineering 是單點的、單次的優化，針對的是單次大模型呼叫的提示詞，目標是讓大模型在單次呼叫中輸出符合預期的結果；
• Context Engineer 是體系化的、全生命週期的工程，針對的是 Agent 從啟動到任務結束的全流程上下文，目標是讓整個 Agent 系統的資訊流轉高效、準確、可控，支撐整個能力閉環的穩定運行。

如果說 Prompt Engineering 是「給 AI 說一句正確的話」，那麼 Context Engineer 就是「給 AI 搭建一套完整的資訊處理系統」。

1.3.2 Context Engineer 的核心模組與能力

Context Engineer 貫穿 Agent 的感知 - 規劃 - 行動 - 記憶 - 反思全流程，其核心體系分為五大模組：

1. 上下文採集與標準化層 這是整個體系的入口，負責採集 Agent 全流程的所有上下文資訊，包括使用者輸入、工具執行結果、多 Agent 互動資訊、外部系統資料、環境回饋等。同時完成資訊的標準化處理，包括格式統一、無效資訊過濾、語義結構化，確保所有進入系統的資訊，都能被 Agent 的各個模組精準識別。

2. 上下文儲存與記憶管理層 這是整個體系的「記憶中樞」，負責上下文的分層儲存與生命週期管理。它將上下文分為三類：

   • 短期記憶：當前會話的即時上下文，支撐當前任務的規劃與執行；
   • 中期記憶：任務全流程的執行記錄、中間結果、回饋資訊，支撐任務的回顧與反思；
   • 長期記憶：使用者歷史偏好、業務知識、產業資料、歷史任務沉澱的經驗，支撐 Agent 的長期能力迭代。 通常會結合向量資料庫、關聯式資料庫、時序資料庫，實現熱資料、溫資料、冷資料的分層管理，兼顧召回效率與儲存成本。
3. 上下文路由與分發層 這是整個體系的「資訊調度中樞」，核心目標是「把正確的資訊，在正確的時間，發給正確的模組」。比如在規劃階段，給大模型傳入任務目標、歷史執行記錄、業務規則等核心上下文；在行動階段，給對應工具傳入參數所需的專屬上下文；在多 Agent 協作時，給不同的子 Agent 分發其任務所需的上下文，同時隔離無關資訊，避免資訊冗餘。它解決了上下文「該給誰、不該給誰」的問題，避免資訊泛濫導致的系統混亂。
4. 上下文壓縮與優化層 這是解決大模型上下文視窗限制、提升推理效率的核心模組。隨著任務的推進，上下文會持續膨脹，很容易超出大模型的視窗限制，導致核心資訊丟失、幻覺頻發。該模組透過語義摘要、冗餘資訊剔除、RAG 精準召回、視窗分層管理等技術，在完整保留核心語義的前提下，對上下文進行動態壓縮與優化。同時基於任務階段，動態調整上下文的內容，只給大模型傳入當前階段所需的核心資訊，大幅提升大模型的推理效率與輸出準確性。
5. 上下文安全與合規層 這是整個體系的「安全閘門」，負責全流程的上下文安全管控。包括使用者隱私資訊去識別化、企業機密資料權限管控、敏感內容過濾、合規稽核日誌留存等。它確保不同的 Agent、工具、使用者，只能存取其權限範圍內的上下文資訊，從根源上避免資料外洩、合規風險等問題。

Context Engineer 的核心價值，就是為 Agent 的整個能力閉環搭建一套高效、可控的資訊流轉體系。它就像人體的神經系統，把感知、規劃、行動、記憶、反思的各個環節連接起來，確保所有資訊都能精準、高效地流轉，讓整個 Agent 系統穩定、可控地運行。

實作個人助理 Agent

開始之前，先明確一下 Agent 具備的基本功能，個人助理 Agent 主要用於採用聊天的方式收集並整理使用者任務，並可以對任務進行整理總結，包含以下基本功能：

1. 支援 MCP 配置
2. 支援 SKILL
3. 互動方式採用終端機聊天的方式

技術棧選擇 pydentic-AI 框架，選擇 Agent 框架原因是為了聚焦 Agent 核心實作邏輯，對於一些繁瑣的邊緣操作，比如：同模型提供商通訊的邏輯、以及 MCP 協定解析相關的邏輯均使用框架提供的封裝。

效果展示

Agent 啟動

MCP 工具查詢

任務建立

任務查詢

取得任務報告

實作解析

互動層

互動採用終端機聊天的方式，透過阻塞式 stdio 實現通訊，具體程式碼如下：

python 体验AI代码助手 代码解读复制代码class TUI:
    def __init__(self):
        self.running = True

    # 開始歡迎語
    def print_welcome(self) -> None:
        print("=" * 50)
        print("Task Assistant Agent 已就緒，請開始對話")
        print("輸入 'exit' 或 'quit' 退出程式")
        print("=" * 50)
        print()

    # 印出訊息內容
    def print_message(self, message: str) -> None:
        print(f"[助理]: {message}")

    # 取得使用者訊息
    def get_user_input(self) -> str:
        try:
            user_input = input("你: ").strip()
            return user_input
        except (EOFError, KeyboardInterrupt):
            return ""

    # 檢查是否是退出命令
    def should_exit(self, user_input: str) -> bool:
        return user_input.lower() in ["exit", "quit"]

    # 退出友善提示
    def print_goodbye(self) -> None:
        print()
        print("=" * 50)
        print("感謝使用，再見！")
        print("=" * 50)

具體使用方式如下：

python 体验AI代码助手 代码解读复制代码def run() -> None:
    tui = TUI()

    # 印出歡迎語
    tui.print_welcome()

    while tui.running:
        # 等待使用者輸入
        user_input = tui.get_user_input()

        if not user_input:
            continue

        if tui.should_exit(user_input):
            tui.running = False
            tui.print_goodbye()
            break

        try:
            # ...省略部分邏輯...

            # 印出訊息
            tui.print_message(response)

        except Exception as e:
            error_msg = f"處理請求時發生錯誤: {str(e)}"
            # 印出訊息
            tui.print_message(error_msg)

if __name__ == "__main__":
    run()

互動邏輯比較簡單，透過在 Agent 開始執行時，實例化 TUI，然後透過呼叫 tui.print_welcome() 印出歡迎語，再透過 tui.get_user_input() 等待使用者輸入訊息，內部透過讀取終端機標準輸入實現，while 死迴圈實現不間斷持續聊天的效果。

執行效果如下：

Agent 核心調度層

這部分主要包括 Agent 建立、系統提示詞、上下文管理、任務管理等核心功能，具體程式碼如下：

python 体验AI代码助手 代码解读复制代码from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.deepseek import DeepSeekProvider
from pydantic import BaseModel, ConfigDict
from typing import Optional

from load_skill import SkillInferenceEngine
from models import Task, Priority, Status
from store import TaskStore
from load_mcp import load_mcp_config

# 添加任務結果物件模型
class AddTaskResult(BaseModel):
    success: bool
    message: str
    task_id: Optional[str] = None

# 任務列表物件模型
class TaskListResult(BaseModel):
    success: bool
    tasks: list[dict]
    total: int

# 任務報告物件模型
class TaskReportResult(BaseModel):
    success: bool
    report: dict

# Agent 上下文依賴模型
class AgentDeps(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    task_store: TaskStore
    skill_engine: SkillInferenceEngine

# 系統提示詞
system_prompt = """你是一個任務助理，幫助使用者管理日常任務。

你有以下能力：
1. 新增任務：從使用者的輸入中提取任務內容和優先級，然後添加到任務列表
2. 查看任務列表：列出所有任務或按優先級/狀態篩選
3. 生成任務報告：按時間整理任務，提供統計資訊

新增任務時，預設優先級為 medium（中等），除非使用者明確指定 high（高）或 low（低）。

請用中文回覆。"""

# 建立 Agent 方法
def create_agent(
    mcp_config_path: str | None = None,
) -> Agent[AgentDeps]:
    # MCP 相關處理
    mcp_servers = []
    if mcp_config_path:
        mcp_servers = load_mcp_config(mcp_config_path)

    # 構造 deepseek 模型連接實例
    model = OpenAIChatModel("deepseek-v4-pro", provider=DeepSeekProvider())

    current_system_prompt = system_prompt

    # 構造 Agent
    agent = Agent(
        model,
        system_prompt=current_system_prompt,
        deps_type=AgentDeps,
        toolsets=mcp_servers,
    )

    @agent.tool
    def add_task(
        ctx: RunContext[AgentDeps], content: str, priority: str = "medium"
    ) -> AddTaskResult:
        """新增一個新任務"""
        try:
            priority_enum = Priority(priority.lower())
        except ValueError:
            return AddTaskResult(
                success=False,
                message=f"無效的優先級: {priority}，有效值為 high, medium, low",
            )

        task = Task(content=content, priority=priority_enum)
        ctx.deps.task_store.add_task(task)

        return AddTaskResult(
            success=True, message=f"任務已新增：{content}", task_id=task.id
        )

    @agent.tool
    def list_tasks(
        ctx: RunContext[AgentDeps],
        priority_filter: str | None = None,
        status_filter: str | None = None,
    ) -> TaskListResult:
        """列出所有任務"""
        tasks = ctx.deps.task_store.get_all_tasks()

        if priority_filter:
            try:
                priority_enum = Priority(priority_filter.lower())
                tasks = [t for t in tasks if t.priority == priority_enum]
            except ValueError:
                pass

        if status_filter:
            try:
                status_enum = Status(status_filter.lower())
                tasks = [t for t in tasks if t.status == status_enum]
            except ValueError:
                pass

        tasks.sort(key=lambda t: (t.priority.value, t.created_at))

        return TaskListResult(
            success=True, tasks=[t.model_dump() for t in tasks], total=len(tasks)
        )

    @agent.tool
    def get_task_report(ctx: RunContext[AgentDeps]) -> TaskReportResult:
        """生成按時間整理的任務報告"""
        report = ctx.deps.task_store.generate_report()

        all_tasks = ctx.deps.task_store.get_all_tasks()
        if all_tasks:
            all_tasks.sort(key=lambda t: t.created_at)
            report["timeline"] = {
                "earliest": all_tasks[0].created_at.isoformat() if all_tasks else None,
                "latest": all_tasks[-1].created_at.isoformat() if all_tasks else None,
            }

        return TaskReportResult(success=True, report=report)

    return agent

重點關注 create_agent 方法，使用 pydantic-ai 實例化一個 Agent 非常方便，主要包括兩個步驟：

1. 構造模型連接實例，這一步主要是對模型提供商 sdk 或 api 的封裝，pydantic-ai 內部提供了對市面上幾乎所有主流模型提供商的支援，這裡我們選擇 DeepSeek 的 deepseek-v4-pro 來作為我們的底層模型。
2. 構造智慧體，透過實例化 Agent 類時傳入上一步的連接實例，得到 Agent 實例，後續和模型的通訊，均基於該 Agent 實例。

除了基本的 Agent 實例建立，pydantic-ai 還支援內部的工具函式，透過 @agent.tool 裝飾器裝飾的方法會被註冊成工具函式，供模型在特定場景呼叫。本次開發的智慧體作為示範，註冊了 3 個方法，分別會在新增任務時、取得任務列表時、取得任務報告時呼叫。

MCP 的支援

前面提到過，MCP 是一個通訊協定，規定了用戶端和伺服器以什麼方式通訊，Agent 中對 MCP 功能的支援，主要指的是將 Agent 作為 MCP 用戶端，如果完全從零實作 MCP 用戶端，需要實作協定層（對 JSON-RPC 2.0 協定的編解碼）、傳輸層（分別對 stdio、sse、Streamable HTTP 的支援）、會話層（管理連線的生命週期）等等，可見還是很繁瑣了（有興趣的可以自行研究實作）。不過好在 pydantic-ai 框架提供了開箱即用的 MCP 用戶端實作，具體程式碼如下：

python 体验AI代码助手 代码解读复制代码from typing import List, TypedDict
import json
from pathlib import Path

from pydantic_ai.mcp import MCPServerStdio

# MCP 配置資料型別介面
class MCPConfig(TypedDict):
    command: str
    args: List[str]
    cwd: str

# 解析 mcp 配置
def process_mcp_config(list: list[MCPConfig]) -> list[MCPServerStdio]:
    configs = []
    for config in list:
        configs.append(MCPServerStdio(
            config['command'],
            args=config['args'],
            timeout=10,
            cwd=config['cwd']
        ))
    return configs

# 載入 mcp 配置檔
def load_mcp_config(config_path: str | Path) -> list[MCPServerStdio]:
    path = Path(config_path)
    if not path.exists():
        return process_mcp_config([])
    with open(path, "r", encoding="utf-8") as f:
        data = json.load(f)
        return process_mcp_config(data['mcp_servers'])

MCP 載入解析檔案主要包括兩個方法：

1. load_mcp_config 方法，用於讀取解析 mcp 配置檔
2. process_mcp_config 方法，針對每個 mcp 配置透過 MCPServerStdio 構建對應的客戶端，最終回傳一個 MCP 客戶端集合

具體使用方式如下：

python 体验AI代码助手 代码解读复制代码def create_agent(
    mcp_config_path: str | None = None,
) -> Agent[AgentDeps]:
    mcp_servers = []

    # 如果 mcp 配置路徑存在，載入 mcp 配置
    if mcp_config_path:
        mcp_servers = load_mcp_config(mcp_config_path)

    model = OpenAIChatModel("deepseek-v4-pro", provider=DeepSeekProvider())

    current_system_prompt = system_prompt

    agent = Agent(
        model,
        system_prompt=current_system_prompt,
        deps_type=AgentDeps,
        toolsets=mcp_servers,
    )

MCP 的載入位於 agent 檔案的 create_agent 方法中，在開始實例化 Agent 之前呼叫 load_mcp_config 載入 MCP 配置，然後在實例化 Agent 時，傳入取得的 MCP 配置（作為 toolsets 傳入）。

SKILL 的支援

開始實作 SKILL 功能支援的程式碼之前，先來梳理一下 SKILL 的解析流程，主要包含兩個階段，分別是 SKILL 載入階段 和 SKILL 執行階段。

SKILL 的載入比較簡單，主要是載入 skills 目錄中的 SKILL.md 檔案，然後依序提取 SKILL.md 檔案的 Frontmatter 欄位（本次實作只支援 name 和 description 欄位），然後構成 SKILL.md 的技能列表，列表中只包括 SKILL.md 的中繼資料。重點介紹下 SKILL 的執行階段。

眾所周知，SKILL 最核心的特點是按需載入，漸進式揭露，這也是其風頭強盛的原因之一。那麼該如何實作其漸進式揭露的功能呢？

核心邏輯其實很簡單，首先當取得使用者的輸入內容時，依序和收集到的 SKILL 中繼資料的描述內容做對比，如果匹配到了，說明命中了 SKILL，然後透過該 SKILL 的中繼資料資訊取得完整的 SKILL 內容，將該內容拼接到系統提示詞後面，隨使用者輸入的內容一起發送給大模型。

針對上面的邏輯，有兩個問題待解決：

1. 如何判斷使用者輸入的內容和哪個 SKILL 的描述更吻合？
2. 如果使用者輸入的內容和每個 SKILL 的描述差別都很大，但終究會找到一個相對來說更吻合一些的 SKILL 執行，對使用者來說也是不合理（如果都差別很大，就都不執行才對），這個場景該怎麼處理？

先回答第一個問題，判斷使用者輸入的內容和哪個 SKILL 的描述更吻合，答案是透過向量化來實現，將所有 SKILL 的名稱(name)+描述(description)轉化為向量，然後同樣將使用者輸入的內容也轉化為向量，將使用者內容向量和每個 SKILL 向量依序計算餘弦相似度，分數越高的 SKILL，說明和使用者內容越接近。

然後是第二個問題，也很簡單，直接設定一個閾值，只有超過這個閾值 SKILL 且相對來說分數最高的 SKILL 才算命中。接下來看下程式碼中的具體實作：

python 体验AI代码助手 代码解读复制代码def preprocess() -> SkillInferenceEngine:
    # 載入 skills
    skills = load_skills_from_directory("skills")

    # 建立向量庫
    store = SkillVectorStore()

    # 註冊 skills
    for skill in skills:
        store.add_skill(skill)

    # 建立檢索索引
    store.build_index()

    # 建立推理引擎
    engine = SkillInferenceEngine(store)

    return engine

首先看下 preprocess 方法，開始先透過 load_skills_from_directory 載入 skills 內容，其程式碼如下：

python 体验AI代码助手 代码解读复制代码def parse_frontmatter(markdown: str) -> tuple[dict, str]:
    """
    解析 markdown 檔案的 frontmatter 和內容

    Args:
        markdown: 原始 markdown 內容

    Returns:
        (frontmatter_dict, content_without_frontmatter)
    """
    match = FRONTMATTER_REGEX.match(markdown)
    if not match:
        return {}, markdown

    frontmatter_text = match.group(1)
    content = markdown[match.end():]

    frontmatter = {}
    for line in frontmatter_text.split('\n'):
        if ':' in line:
            key, _, value = line.partition(':')
            frontmatter[key.strip()] = value.strip()

    return frontmatter, content

def load_skill_from_file(file_path: str) -> Optional[Skill]:
    """
    從單個 SKILL.md 檔案載入技能

    Args:
        file_path: SKILL.md 檔案路徑

    Returns:
        Skill 物件，解析失敗返回 None
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
    except (IOError, OSError):
        return None

    frontmatter, markdown_content = parse_frontmatter(content)

    # 解析 name（目錄名作為 fallback）
    skill_name = frontmatter.get('name')
    if not skill_name:
        skill_name = Path(file_path).parent.name

    # 解析 description（取第一行作為 fallback）
    description = frontmatter.get('description')
    if not description:
        for line in markdown_content.split('\n'):
            line = line.strip()
            if line:
                if line.startswith('#'):
                    description = line.lstrip('#').strip()
                else:
                    description = line
                break
        if not description:
            description = skill_name

    return Skill(
        name=skill_name,
        description=description,
        content=markdown_content.strip(),
        file_path=file_path
    )

def load_skills_from_directory(skills_dir: str) -> list[Skill]:
    """
    從 skills 目錄載入所有技能

    目錄結構:
        skills/
            skill-name/
                SKILL.md
            another-skill/
                SKILL.md

    Args:
        skills_dir: skills 根目錄路徑

    Returns:
        Skill 物件列表
    """
    skills = []
    skills_path = Path(skills_dir)

    if not skills_path.exists():
        return skills

    for entry in skills_path.iterdir():
        if not entry.is_dir():
            continue

        skill_file = entry / 'SKILL.md'
        if not skill_file.exists():
            continue

        skill = load_skill_from_file(str(skill_file))
        if skill:
            skills.append(skill)

    return skills

整體 skills 的載入有三個方法構成，分別是

• load_skills_from_directory：從 skills 目錄載入所有單獨 skill 目錄
• load_skill_from_file：從每個單獨 skill 目錄載入解析 SKILL.md 檔案
• parse_frontmatter：解析 SKILL.md 檔案，提取 Frontmatter 中的 name 和 description，以及 SKILL 內容

這裡做了簡化處理，直接寫死從固定目錄中載入 SKILL，在真實場景中，還需要包含從專案根目錄、內部設定的全域目錄、內建 skill 目錄中載入。

然後是建立向量庫，程式碼如下：

python 体验AI代码助手 代码解读复制代码class SkillVectorStore:
    def __init__(self):
        self.model = SentenceTransformer('./all-MiniLM-L6-v2')  # 輕量語義模型
        self.skills: List[Skill] = []
        self.index = None
        self.embeddings = []

    # 把技能描述向量化
    def add_skill(self, skill: Skill):
        text = f"{skill.name} {skill.description}"
        emb = self.model.encode([text])[0]
        self.skills.append(skill)
        self.embeddings.append(emb)

    # 建立 FAISS 檢索庫
    def build_index(self):
        emb_np = np.array(self.embeddings).astype('float32')
        self.index = faiss.IndexFlatL2(emb_np.shape[1])
        self.index.add(emb_np)

    # 檢索最匹配的技能
    def search(self, query: str, top_k=1):
        query_emb = self.model.encode([query])
        D, I = self.index.search(np.array(query_emb).astype('float32'), top_k)
        idx = I[0][0]
        score = 1 - D[0][0] / 10  # 轉成相似度 0~1
        return self.skills[idx], score

這個類主要功能包括：

1. add_skill：對 SKILL 中繼資料進行向量化
2. build_index：為所有 SKILL 向量化後的向量集合建立索引，用於後續的高效率相似度檢索
3. search：對使用者內容進行向量化，然後透過建立的索引執行相似度檢索

這裡用到了兩個工具，分別是：

• FAISS：Facebook 開源的本地向量庫，支援超快相似度檢索、意圖檢索

• Sentence-Transformers：文字向量化模型，特點是輕量可本地執行，用於把文字變成向量

有關這兩個工具的詳細資訊，有興趣的自行研究。

向量庫實例建構好以後，遍歷 SKILL 中繼資料集合，依序向量化，完成後，為 SKILL 向量集合建立檢索索引，程式碼如下：

python 体验AI代码助手 代码解读复制代码def preprocess() -> SkillInferenceEngine:
    # 載入 skills
    skills = load_skills_from_directory("skills")

    # 建立向量庫
    store = SkillVectorStore()

    # 註冊 skills【這裡】
    for skill in skills:
        store.add_skill(skill)

    # 建立檢索索引【這裡】
    store.build_index()

    # 建立推理引擎
    engine = SkillInferenceEngine(store)

    return engine

最後建立推理引擎，推理引擎的程式碼如下：

python 体验AI代码助手 代码解读复制代码class SkillInferenceEngine:
    def __init__(self, vector_store: SkillVectorStore):
        self.vs = vector_store
        self.threshold = 0.65  # 閾值

    def infer(self, query: str) -> Optional[Skill]:
        skill, score = self.vs.search(query)
        if score >= self.threshold:
            print(f"✅ 命中技能: {skill.name} | 相似度: {score:.2f}")
            return skill
        else:
            print(f"❌ 無匹配技能 | 最高分: {score:.2f}")
            return None

推理引擎的程式碼主要就是透過向量庫實例的 search 方法做使用者內容向量和 SKILL 向量的相似度檢索，並設定基礎閾值，最終檢索出一個符合要求的 SKILL。

整個 preprocess 預處理方法在啟動 Agent 時執行，得到推理引擎，然後在每次使用者輸入內容時透過推理引擎執行 SKILL 搜尋，具體程式碼如下：

python 体验AI代码助手 代码解读复制代码def run() -> None:
    # ...省略部分代碼...

    # 預處理 skills
    skill_engine = preprocess()

    # ...省略部分代碼...

    while tui.running:
        user_input = tui.get_user_input()

        # ...省略部分代碼...

        try:
            # 透過推理引擎，提取命中 skill
            target_skill = skill_engine.infer(user_input)
            instructions = None
            if target_skill:
                print(f"命中技能：{target_skill.name}")
                instructions = target_skill.content

            deps = AgentDeps(task_store=task_store, skill_engine=skill_engine)
            # 向模型發送訊息時，一併帶入命中的 SKILL 內容
            result = agent.run_sync(user_input, deps=deps, instructions=instructions)

            response = result.output
            context.add_message("assistant", response)
            tui.print_message(response)

        except Exception as e:
            error_msg = f"處理請求時出錯: {str(e)}"
            context.add_message("assistant", error_msg)
            tui.print_message(error_msg)

從上面程式碼中可以看到，命中的 SKILL 內容是透過 instructions 攜帶發送給模型的。

instructions 在 pydantic-ai 框架被稱為指令，也可以理解為另一種 System Prompt，差別是 instructions 的內容不會存在於歷史訊息中，滿足 SKILL 內容用後即焚的要求。

Context Engineer 功能

前面的介紹中，Context Engineer 包含的內容很多，本專案是處於示範的目的，因此只會選擇其中一兩點實作，只實作以下功能：

• 上下文壓縮
• 上下文儲存

同樣的，pydantic-ai 對上下文的處理提供了很好的支援，具體實作程式碼如下：

python 体验AI代码助手 代码解读复制代码from openai import BaseModel
from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.deepseek import DeepSeekProvider

class ContextManager(BaseModel):
    __contexts: list[ModelMessage] = []
    __compress_agent: Agent = None

    def update_messages(self, messages: list[ModelMessage]):
        self.__contexts = messages

    def clear_messages(self):
        self.__contexts = []

    def get_messages(self):
        return self.__contexts

    def compress_messages_and_update(self, messages: list[ModelMessage]):
        if not self.__compress_agent:
            model = OpenAIChatModel("deepseek-v4-pro", provider=DeepSeekProvider())
            self.__compress_agent = Agent(
                model, # 這裡可以選擇更便宜的小模型
                instructions="""總結本次對話，省略閒聊內容及無關話題。重點梳理任務要點與後續工作計畫。""",
            )

        compress_content = self.__compress_agent.run_sync(message_history=messages)
        self.__contexts = compress_content

這裡實作了一個用於上下文管理的類，包含三個方法，分別是：

• update_messages：更新上下文訊息
• clear_messages：清空上下文訊息
• get_messages：取得上下文訊息
• compress_messages_and_update：壓縮上下文訊息並記錄

上下文管理類實例化的位置如下：

python 体验AI代码助手 代码解读复制代码def run() -> None:
    # ...省略部分代碼...

    # 上下文管理器在這裡實例化
    context_manager = ContextManager()

    tui = TUI()
    agent = create_agent(mcp_path)

    # ...省略部分代碼...

    # context = context_manager.create_context(IntentType.UNKNOWN)

    while tui.running:
        user_input = tui.get_user_input()

        # ...省略部分代碼...

        try:
            # ...省略部分代碼...

            # 在向模型發送訊息之前先載入歷史訊息
            message_history = context_manager.get_messages()

            deps = AgentDeps(task_store=task_store, skill_engine=skill_engine)
            # 將使用者訊息和歷史訊息一同發送給大模型（透過 message_history 欄位）
            result = agent.run_sync(user_input, deps=deps, instructions=instructions, message_history=message_history)

            tui.print_message(result.output)

             # 更新歷史訊息：如果歷史訊息的 token 數量大於 1000，執行訊息壓縮
            usage = result.usage()
            if usage.total_tokens > 1000:
                context_manager.compress_messages_and_update(result.all_messages)
            else:
                context_manager.update_messages(result.all_messages)

        except Exception as e:
            error_msg = f"處理請求時出錯: {str(e)}"
            tui.print_message(error_msg)

上下文管理的流程如下：

1. 首先在 Agent 啟動的時候構造上下文管理器實例
2. 然後在取得使用者訊息之後，發送給大模型之前，取得歷史訊息記錄
3. 然後將使用者訊息和歷史訊息記錄一同發送給大模型
4. 模型回應訊息以後，判斷所有訊息的 token 量是否超過了設定的閾值（這裡設定的是 1000）
5. 如果超出了 token 閾值，執行歷史訊息壓縮，將壓縮後的訊息更新到歷史訊息記錄中，供後續使用。

總結

本文首先簡單介紹了一下 Agent 應用的三大基礎設施 MCP、SKILL、Context Engineer，然後實作了一個簡化版的個人助理 Agent，本次實作的專案只是為了起到拋磚引玉的作用，因為實際生產環境可用的 Agent 的複雜度遠遠超過這個，比如：對於上下文管理中設計的記憶管理（記憶分層、短期記憶、長期記憶）、以 RAG 為核心的資訊檢索和過濾。再比如：對於歷史訊息的壓縮，如何將模型的解析成本，如何優先移除低冗餘、老舊資訊，保留核心推理鏈等等，所以說一個生產環境可落地的 Agent 的開發，是一個複雜的系統性工程，限於篇幅，本文的分享就到這裡。

原文出處：https://juejin.cn/post/7641133523776749608