在日常生活中、工作中，从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，实现通信，具体代码如下：

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)

具体使用方式如下：

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创建、系统提示词、上下文管理、任务管理等核心功能，具体代码如下：

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客户端实现，具体代码如下：

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客户端集合

具体使用方式如下：

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才算命中。接下来看下代码中的具体实现：

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内容，其代码如下：

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内容

这里做了简化处理，直接写死从固定目录中加载SKLL，在真实场景中，还需要包含从项目根目录、内部设置的全局目录、内置skill目录中加载。

然后是创建向量库，代码如下：

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向量集合构建检索索引，代码如下：

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

最后构建推理引擎，推理引擎的代码如下：

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搜索，具体代码如下：

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对上下文的处理提供了很好的支持，具体实现代码如下：

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：压缩上下文消息并记录

上下文管理类实例化的位置如下：

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的开发，是一个复杂的系统性工程，限于篇幅，本文的分享就到这里。

这里给大家精心整理了一份全面的AI大模型学习资源，包括：AI大模型全套学习路线图（从入门到实战）、精品AI大模型学习书籍手册、视频教程、实战学习、面试题等，资料免费分享！

👇👇扫码免费领取全部内容👇👇

1. 成长路线图&学习规划

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

这里，我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。可以说是最科学最系统的学习成长路线。

2. 大模型经典PDF书籍

书籍和学习文档资料是学习大模型过程中必不可少的，我们精选了一系列深入探讨大模型技术的书籍和学习文档，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。（书籍含电子版PDF）

3. 大模型视频教程

对于很多自学或者没有基础的同学来说，书籍这些纯文字类的学习教材会觉得比较晦涩难以理解，因此，我们提供了丰富的大模型视频教程，以动态、形象的方式展示技术概念，帮助你更快、更轻松地掌握核心知识。

4. 2026行业报告

行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

5. 大模型项目实战

学以致用 ，当你的理论知识积累到一定程度，就需要通过项目实战，在实际操作中检验和巩固你所学到的知识，同时为你找工作和职业发展打下坚实的基础。

6. 大模型面试题

面试不仅是技术的较量，更需要充分的准备。

在你已经掌握了大模型技术之后，就需要开始准备面试，我们将提供精心整理的大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

7. 资料领取：全套内容免费抱走，学 AI 不用再找第二份

不管你是 0 基础想入门 AI 大模型，还是有基础想冲刺大厂、了解行业趋势，这份资料都能满足你！
现在只需按照提示操作，就能免费领取：

👇👇扫码免费领取全部内容👇👇