在日常生活中、工作中从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统一的工具调用协议层MCPModel Context Protocol模型上下文协议是由Anthropic 公司在2024 年 11 月 25 日开源发布其核心定位是为大模型与工具之间提供一套统一的、上下文感知的通信协议。它向下兼容各类CLI工具、API服务、系统能力向上给Agent提供标准化的工具调用接口解决了工具碎片化、适配成本高、调用不稳定的核心问题。其核心能力价值体现在三个方面标准化的工具定义与交互规范MCP定义了统一的工具元数据格式、参数校验规则、返回结构与错误处理机制Agent无需针对单个工具做定制化适配只需遵循MCP协议即可完成调用大幅提升工具调用的成功率与稳定性原生的上下文感知能力不同于传统工具调用的静态参数传递MCP可以将Agent的会话上下文、历史执行记录、用户权限等信息同步给工具让工具可以结合上下文动态调整执行逻辑实现更智能的交互分层的安全与权限管控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的感知-规划-行动-记忆-反思全流程其核心体系分为五大模块上下文采集与标准化层 这是整个体系的入口负责采集Agent全流程的所有上下文信息包括用户输入、工具执行结果、多Agent交互信息、外部系统数据、环境反馈等。同时完成信息的标准化处理包括格式统一、无效信息过滤、语义结构化确保所有进入系统的信息都能被Agent的各个模块精准识别。上下文存储与记忆管理层 这是整个体系的“记忆中枢”负责上下文的分层存储与生命周期管理。它将上下文分为三类短期记忆当前会话的实时上下文支撑当前任务的规划与执行中期记忆任务全流程的执行记录、中间结果、反馈信息支撑任务的复盘与反思长期记忆用户历史偏好、业务知识、行业数据、历史任务沉淀的经验支撑Agent的长期能力迭代。 通常会结合向量数据库、关系型数据库、时序数据库实现热数据、温数据、冷数据的分层管理兼顾召回效率与存储成本。上下文路由与分发层 这是整个体系的“信息调度中枢”核心目标是“把正确的信息在正确的时间发给正确的模块”。比如在规划阶段给大模型传入任务目标、历史执行记录、业务规则等核心上下文在行动阶段给对应工具传入参数所需的专属上下文在多Agent协作时给不同的子Agent分发其任务所需的上下文同时隔离无关信息避免信息冗余。它解决了上下文“该给谁、不该给谁”的问题避免信息泛滥导致的系统混乱。上下文压缩与优化层 这是解决大模型上下文窗口限制、提升推理效率的核心模块。随着任务的推进上下文会持续膨胀很容易超出大模型的窗口限制导致核心信息丢失、幻觉频发。 该模块通过语义摘要、冗余信息剔除、RAG精准召回、窗口分层管理等技术在完整保留核心语义的前提下对上下文进行动态压缩与优化。同时基于任务阶段动态调整上下文的内容只给大模型传入当前阶段所需的核心信息大幅提升大模型的推理效率与输出准确性。上下文安全与合规层 这是整个体系的“安全闸门”负责全流程的上下文安全管控。包括用户隐私信息脱敏、企业机密数据权限管控、敏感内容过滤、合规审计日志留存等。它确保不同的Agent、工具、用户只能访问其权限范围内的上下文信息从根源上避免数据泄露、合规风险等问题。Context Engineer的核心价值就是为Agent的整个能力闭环搭建一套高效、可控的信息流转体系。它就像人体的神经系统把感知、规划、行动、记忆、反思的各个环节连接起来确保所有信息都能精准、高效地流转让整个Agent系统稳定、可控地运行。实现个人助理Agent开始之前先明确下Agent具备的基本功能个人助理Agent主要用于采用聊天的方式收集并整理用户任务并可以对任务进行整理总结包含以下基本功能支持MCP配置支持SKILL交互方式采用终端聊天的方式技术栈选择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_allowedTrue) 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, providerDeepSeekProvider()) current_system_prompt system_prompt # 构造Agent agent Agent( model, system_promptcurrent_system_prompt, deps_typeAgentDeps, toolsetsmcp_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( successFalse, messagef无效的优先级: {priority}有效值为 high, medium, low, ) task Task(contentcontent, prioritypriority_enum) ctx.deps.task_store.add_task(task) return AddTaskResult( successTrue, messagef任务已添加{content}, task_idtask.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(keylambda t: (t.priority.value, t.created_at)) return TaskListResult( successTrue, tasks[t.model_dump() for t in tasks], totallen(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(keylambda 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(successTrue, reportreport) return agent重点关注create_agent方法使用pydantic-ai实例化一个Agent非常方便主要包括两个步骤构造模型连接实例这一步主要是对模型提供商sdk或api的封装pydantic-ai内部提供了对市面上几乎所有主流的模型提供商的支持这里我们选择Deepseek的deepseek-v4-pro来作为我们的底层模型。构造智能体通过实例化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], argsconfig[args], timeout10, cwdconfig[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, encodingutf-8) as f: data json.load(f) return process_mcp_config(data[mcp_servers])MCP加载解析文件主要包括两个方法load_mcp_config方法用于读取解析mcp配置文件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, providerDeepSeekProvider()) current_system_prompt system_prompt agent Agent( model, system_promptcurrent_system_prompt, deps_typeAgentDeps, toolsetsmcp_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内容将该内容拼接到系统提示词后面随用户输入的内容一起发送给大模型。针对上面的逻辑有两个问题待解决如何判断用户输入的内容和哪个SKILL的描述更吻合如果用户输入的内容和每个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, encodingutf-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( nameskill_name, descriptiondescription, contentmarkdown_content.strip(), file_pathfile_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_k1): 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这个类主要功能包括add_skill对SKILL元数据进行向量化build_index为所有SKILL向量化后的向量集合构建索引用于后续的高效相似度检索search对用户内容进行向量化然后通过构建的索引执行相似度检索这里用到了两个工具分别是FAISSFacebook 开源的本地向量库支持超快相似度检索、意图检索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_storetask_store, skill_engineskill_engine) # 向模型发送消息时一并带入命中的SKILL内容 result agent.run_sync(user_input, depsdeps, instructionsinstructions) 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, providerDeepSeekProvider()) self.__compress_agent Agent( model, # 这里可以选择更便宜的小模型 instructions总结本次对话省略闲聊内容及无关话题。重点梳理任务要点与后续工作计划。, ) compress_content self.__compress_agent.run_sync(message_historymessages) 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_storetask_store, skill_engineskill_engine) # 将用户消息和历史消息一同发送给大模型通过message_history字段 result agent.run_sync(user_input, depsdeps, instructionsinstructions, message_historymessage_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)上下文管理的流程如下首先在Agent启动的时候构造上下文管理器实例然后在获取到用户消息之后发送给大模型之前获取历史消息记录然后将用户消息和历史消息记录一同发送给大模型模型响应消息以后判断所有消息的token量是否超过了设置的阈值这里设置的是1000如果超出了token阈值执行历史消息压缩将压缩后的消息更新到历史消息记录中供后续使用。总结本文首先简单介绍了一下Agent应用的三大基础设施MCP、SKILL、Context Engineer然后实现了一个简化版的个人助理Agent本次实现的项目只是为了起到抛砖引玉的作用因为实际生产环境可用的Agent的复杂度远远超过这个比如对于上下文管理中设计的记忆管理记忆分层、短期记忆、长期记忆、以RAG为核心的信息检索和过滤。再比如对于历史消息的压缩如何将成模型的解析成本如何优先移除低冗余、老旧信息保留核心推理链等等所以说一个生产环境可落地的Agent的开发是一个复杂的系统性工程限于篇幅本文的分享就到这里。这里给大家精心整理了一份全面的AI大模型学习资源包括AI大模型全套学习路线图从入门到实战、精品AI大模型学习书籍手册、视频教程、实战学习、面试题等资料免费分享扫码免费领取全部内容1. 成长路线图学习规划要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。这里我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。可以说是最科学最系统的学习成长路线。2. 大模型经典PDF书籍书籍和学习文档资料是学习大模型过程中必不可少的我们精选了一系列深入探讨大模型技术的书籍和学习文档它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。书籍含电子版PDF3. 大模型视频教程对于很多自学或者没有基础的同学来说书籍这些纯文字类的学习教材会觉得比较晦涩难以理解因此我们提供了丰富的大模型视频教程以动态、形象的方式展示技术概念帮助你更快、更轻松地掌握核心知识。4. 2026行业报告行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。5. 大模型项目实战学以致用当你的理论知识积累到一定程度就需要通过项目实战在实际操作中检验和巩固你所学到的知识同时为你找工作和职业发展打下坚实的基础。6. 大模型面试题面试不仅是技术的较量更需要充分的准备。在你已经掌握了大模型技术之后就需要开始准备面试我们将提供精心整理的大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。7. 资料领取全套内容免费抱走学 AI 不用再找第二份不管你是 0 基础想入门 AI 大模型还是有基础想冲刺大厂、了解行业趋势这份资料都能满足你现在只需按照提示操作就能免费领取扫码免费领取全部内容
)
