AI Agent 工程师必懂的三件事:Agent 架构、Prompt 工程与 MCP 协议
一份不绑定平台的通用技术指南:Agent 循环与多智能体编排、上下文传递与错误传播、Few-shot 与结构化输出、控制误报,以及 MCP 协议的结构、消息格式和一个最小 Server 实现。
English version: Three things every AI agent engineer should know: agent architecture, prompt engineering, and MCP
这篇指南讲构建可靠 AI 应用的三块底层知识:Agent 架构设计、Prompt Engineering 原理、MCP 协议。它们不属于任何一家模型厂商,换 OpenAI、Anthropic 还是 Gemini 都用得上,换 LangChain、LlamaIndex 等框架也一样。学会原理,不被特定工具绑架。
第一部分:Agent 架构设计
一个 Agent 本质上是:让 LLM 反复“思考、调用工具、看结果、再思考”,直到任务完成。搞懂这个循环的结构,是构建可靠 AI 应用的第一步。
1.1 核心机制:Agentic Loop
传统程序调用 LLM 是一次性的:发一个请求,拿一个回答,结束。而 Agent 是循环式的。模型每次回答可能是“我需要调用一个工具”,程序执行完工具后把结果喂回去,继续下一轮。
你的程序 LLM 模型
发送消息 ──────────────────▶ 思考...
(含工具定义) ↓
stop_reason = "tool_use"
◀────── 返回 tool_call ───── "我需要调用 search_web"
↓
执行工具(真正运行代码)
↓
把结果加入对话 ──────────────▶ 继续思考...
↓
stop_reason = "end_turn"
◀────── 返回最终答案 ──────── "根据搜索结果,答案是..."
关键:循环在 stop_reason = "end_turn" 时才停止
控制流的正确写法,任何 LLM SDK 都适用这个结构:
def run_agent(messages, tools):
while True:
response = llm.complete(messages=messages, tools=tools)
# 任务完成,退出循环
if response.stop_reason == "end_turn":
return response.content
# 需要调用工具,继续循环
if response.stop_reason == "tool_use":
tool_results = []
for tool_call in response.tool_calls:
result = execute_tool(tool_call.name, tool_call.input)
tool_results.append({
"tool_call_id": tool_call.id,
"content": result
})
# 把工具结果追加到对话历史
messages.append(response) # assistant 回复
messages.append(tool_results) # tool 结果
# 继续下一轮
常见陷阱:不要用“解析文本判断是否结束”,比如检查回答里有没有“任务完成”这几个字。应该依赖
stop_reason。LLM 的输出是概率性的,解析文本会有随机失败率。
1.2 多智能体编排:分工协作
当任务太复杂时,一个 Agent 搞不定。不是因为它“不够聪明”,而是上下文窗口有限,同时做太多事会让注意力分散、出错率上升。解决方案是多个 Agent 分工。
经典模式是 Hub & Spoke(主从架构):
┌──────────────┐
│ Coordinator │ ← 主控 Agent:任务分解、调度、聚合结果
└──────┬───────┘
┌────────────────┼────────────────┐
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ SearchAgent │ │ AnalyzeAgent │ │ WriteAgent │
│ (搜索网络) │ │ (分析文档) │ │ (撰写报告) │
└──────────────┘ └──────────────┘ └──────────────┘
子 Agent 之间不直接通信,所有协调都经过 Coordinator
好处:可观测性强、错误处理统一、信息流可控
关键原则:子 Agent 没有记忆。 每个子 Agent 只知道你在本次调用时传给它的内容。父 Agent 的对话历史不会自动传递给子 Agent,你必须显式地把需要的上下文包含在子 Agent 的提示词里。
# 错误做法:子 Agent 拿不到任何上下文
result = spawn_agent(
agent="WriteAgent",
prompt="请根据研究结果写一份报告" # 什么是"研究结果"?
)
# 正确做法:显式传入所有需要的上下文
result = spawn_agent(
agent="WriteAgent",
prompt=f"""请写一份关于"AI 对创意行业影响"的报告。
以下是已收集的研究结论:
{search_results} # 搜索 Agent 的输出
以下是关键文献摘要:
{analysis_results} # 分析 Agent 的输出
要求:……"""
)
并行还是串行,取决于任务之间有没有依赖:
| 场景 | 推荐方式 | 原因 |
|---|---|---|
| 搜索多个独立话题 | 并行 | 各任务互不依赖,并行可减少总延迟 |
| 先搜索再分析 | 串行 | 分析依赖搜索结果,必须按序执行 |
| 检查多个文件 | 并行 | 文件独立,同时处理效率最高 |
| 身份验证后再执行操作 | 串行 | 安全关键,必须保证顺序 |
1.3 上下文传递:信息如何流动
上下文(Context)是 LLM 的“工作记忆”。它看不到的东西,就当不存在。管理好上下文是 Agent 工程的核心难题。常见的四种策略:
- 追加式历史:每轮对话把上下文累积追加。简单,但 token 越来越多,成本上升。
- 渐进式压缩:定期把旧历史压缩成摘要。省 token,但数字、日期容易在压缩中丢失。
- 关键事实提取:把重要数据(订单号、金额、时间)单独提取成结构化块,始终保留在上下文顶部。
- 工具输出裁剪:API 返回 40 个字段,但只有 5 个有用。不裁剪会浪费大量 token 并干扰模型注意力。
还有一个“Lost in the Middle”问题:研究表明,LLM 对上下文头部和尾部的内容注意力最强,中间部分容易被忽略。设计 prompt 时按这个顺序安排:
system_prompt = """
[最重要的指令和规则] ← 放最前面,不会被遗忘
--- 背景信息 ---
{background_context} ← 放中间(相对次要)
--- 关键事实(始终保留)---
客户ID: {customer_id} ← 关键数据也放到尾部附近
订单: {order_details}
--- 当前任务 ---
{current_task} ← 放最后,模型处理时印象最深
"""
1.4 错误传播:失败时怎么办
现实中的工具调用会失败:网络超时、权限不足、数据不存在。错误处理的设计决定了一个 Agent 系统是否真正可靠。先分类:
| 类型 | 例子 | 正确处理 |
|---|---|---|
| 瞬时错误 | 网络超时、服务临时不可用 | 本地重试(指数退避),重试后仍失败再上报 |
| 验证错误 | 输入格式不对、缺少必填字段 | 不重试,直接上报,说明原因 |
| 业务错误 | 余额不足、超出权限 | 不重试,返回用户友好说明 |
| 权限错误 | 未授权访问 | 不重试,可能需要升级到人工处理 |
工具应该返回结构化的错误,不依赖特定框架:
def search_customer(customer_id):
try:
result = db.query(customer_id)
return { "success": True, "data": result }
except TimeoutError:
return {
"success": False,
"error_type": "transient", # 错误类别
"is_retryable": True, # 可以重试
"message": "数据库连接超时,请稍后重试"
}
except PermissionError:
return {
"success": False,
"error_type": "permission",
"is_retryable": False,
"message": "无权访问该客户记录,需要人工审核"
}
设计原则:子 Agent 应该先尝试在本地恢复(如自动重试瞬时错误),只有本地无法处理时才把错误上报给 Coordinator。上报时必须包含:失败类型 + 已尝试的操作 + 部分结果(如有)。这样 Coordinator 才能做出正确的恢复决策。
三个会让系统难以调试的反模式:
- 静默吃掉错误:工具出错了,但返回空列表假装成功。下游 Agent 以为没有数据,做出错误决策。
- 遇到任何错误就终止整个流程:一个子 Agent 的搜索超时,结果整个研究任务失败。正确做法是带着部分结果继续,并在最终输出中标注哪些信息有缺口。
- 返回模糊错误信息:
"操作失败"对 LLM 毫无帮助,它不知道该重试、换策略还是上报。
第二部分:Prompt Engineering 原理
Prompt 工程不是“玄学”,而是有规律可循的工程实践。这一章讲三个核心技术:让模型学会举一反三的 Few-shot、让输出可被程序解析的 Structured Output,以及减少误报提高精确度的方法。
2.1 Few-shot Prompting:用例子代替指令
当你用语言描述一个规则,模型可能有多种理解方式。但当你给出 2 到 4 个具体的输入/输出例子,模型可以直接学习模式本身,理解歧义大幅减少。
为什么有效:LLM 在预训练阶段见过无数“例子、模式、应用”的文本结构,它天然擅长从示例中归纳规律并推广到新情况。相比抽象描述,具体例子提供的信息密度更高。
仅有指令(输出不稳定):
SYSTEM:分析代码审查发现,输出格式要专业,按严重程度分类,给出修改建议。
OUTPUT:这段代码有一些问题需要注意。首先,SQL 查询存在安全隐患……(有时是列表,有时是段落,格式各异)
Few-shot(输出格式稳定):
分析代码,输出格式严格如下示例:
示例输入:query = "SELECT * FROM users WHERE id = " + user_input
示例输出:
🔴 [HIGH] SQL 注入漏洞
位置:第 3 行
问题:用户输入直接拼接到 SQL 语句
修复:使用参数化查询 WHERE id = ?
---
示例输入:def process(): pass # TODO: implement
示例输出:
🟡 [MEDIUM] 未完成的实现
位置:第 1 行
问题:函数体为空,含有 TODO 注释
修复:实现具体逻辑或移除占位符
Few-shot 的最佳实践:
- 示例数量 2 到 4 个。太少模式不够稳定,太多占用 token 且边际收益递减。
- 覆盖边界情况。不只展示正常情况,还要展示“这种情况不需要报告”“这种情况是误报”。
- 包含负例。告诉模型哪些情况虽然看起来像问题但其实是正常模式,这是降低误报最直接的手段。
- 格式示例就是格式约束。你展示什么格式,模型就会模仿什么格式,不需要额外的格式说明。
2.2 结构化输出:让 LLM 的输出可被程序解析
LLM 默认输出自然语言文本。但在自动化流水线中,你需要的是可被代码解析的 JSON 或其他结构化数据,不能让程序去“理解”自然语言。三种方法,可靠性递增。
方法 A:提示词约束,最简单,不够可靠。
# 告诉模型只输出 JSON,但无法保证
prompt = """请以 JSON 格式输出分析结果:
{"severity": "high/medium/low", "issue": "...", "fix": "..."}
只输出 JSON,不要其他文字。"""
# 问题:模型可能输出 ```json ... ``` 或前面加一句话
# 需要额外的清洗代码,仍有小概率格式错误
方法 B:JSON Schema 约束,更可靠。
# 大多数现代 LLM API 支持 response_format 参数
response = llm.complete(
prompt="分析这段代码的安全问题",
response_format={
"type": "json_schema",
"schema": {
"type": "object",
"properties": {
"severity": {"type": "string", "enum": ["high", "medium", "low"]},
"issues": {"type": "array", "items": {"type": "string"}},
"explanation": {"type": "string"}
},
"required": ["severity", "issues"]
}
}
)
# 输出保证符合 schema 的语法,但语义仍可能有误
方法 C:Tool Use 强制结构化,最可靠。把“提取结构化数据”伪装成一个“工具调用”。这样模型会主动生成符合工具参数 schema 的 JSON,而不是生成自然语言再转换。
# 定义一个"提取工具",实际上只是为了强制结构化输出
tools = [{
"name": "extract_code_issues",
"description": "提取代码中发现的安全和质量问题",
"parameters": {
"type": "object",
"properties": {
"issues": {
"type": "array",
"items": {
"type": "object",
"properties": {
"severity": {"enum": ["high", "medium", "low"]},
"line": {"type": "integer"},
"description": {"type": "string"},
"fix": {"type": "string"}
},
"required": ["severity", "description"]
}
}
}
}
}]
response = llm.complete(
prompt="分析这段代码",
tools=tools,
tool_choice="any" # 强制必须调用工具
)
# 从 tool_call 的 arguments 里取数据,格式绝对正确
Schema 设计技巧:
| 情况 | 设计方式 | 原因 |
|---|---|---|
| 文档里不一定有这个信息 | 字段设为 nullable(可选) | 否则模型会编造数据来满足 required |
| 类别固定但可能有新类型 | enum 里加一个 “other” 加一个描述字段 | 避免模型强行归类到不合适的已知类别 |
| 数值需要验证一致性 | 同时提取 calculated 和 stated 两个字段 | 可以对比发现语义错误,比如明细不等于总计 |
2.3 控制误报:提高精确度
对于代码审查、安全检测、内容审核等场景,误报比漏报危害更大。它会让用户失去对系统的信任,最终忽略所有警告,包括真正重要的那些。误报的三个根本原因:
- 标准模糊:“要保守一点”“只报高可信度”这类指令太抽象,模型无法准确执行。
- 过于宽泛:“检查所有潜在问题”导致模型把所有不确定的情况都报出来。
- 缺少负例:没有示例说明哪些情况是可接受的,模型没有参照。
解决方案是用具体标准替代模糊指令。
模糊指令(误报率高):
检查代码中的问题,要保守,只报高置信度的发现,避免误报。
具体标准(误报率低):
报告下列情况(明确要报):
• 代码行为与注释说明相矛盾
• 未处理的异常路径会导致数据丢失
• SQL / 命令注入风险
不要报告下列情况(明确不报):
• 代码风格偏好(缩进、命名风格等)
• 项目内部约定的模式(即使看起来非常规)
• 不影响功能的冗余代码
严重程度定义:
HIGH:可被利用的安全漏洞,例如 eval(user_input)
MEDIUM:有错误路径会导致数据损坏,例如未加事务的批量写入
LOW:不影响正确性的改进建议
验证与重试循环。对于结构化提取,模型有时会因为格式问题输出错误。正确做法是把错误反馈给模型让它自我修正,而不是直接报错:
def extract_with_retry(document, max_retries=2):
messages = [{"role": "user", "content": document}]
for attempt in range(max_retries):
result = llm.complete(messages, tools=[extract_tool])
extracted = result.tool_calls[0].arguments
errors = validate(extracted)
if not errors:
return extracted # 验证通过
# 验证失败:把错误反馈给模型
messages.append({"role": "assistant", "content": result})
messages.append({
"role": "user",
"content": f"提取结果有以下问题,请修正:\n{errors}"
})
raise ExtractionError("多次重试后仍无法提取正确结果")
什么时候重试无效:重试只能解决格式错误,也就是模型知道信息但输出结构不对。如果信息在原始文档中根本不存在,再多重试也没用。应该把字段设为 nullable,让模型返回 null 而不是编造。
第三部分:MCP 协议
MCP(Model Context Protocol)是一个让 AI 模型调用外部工具和数据源的开放标准协议。就像 USB 统一了外设接口,MCP 试图统一“AI 模型连接外部世界”的方式:一次接入,各家模型都能用。
3.1 MCP 是什么:AI 的“USB 接口”
在 MCP 出现之前,每家 AI 服务调用工具的方式各不相同,工具开发者需要为每家平台单独适配。
【没有 MCP 时】
OpenAI ──自定义接口──▶ 你的数据库工具(为 OpenAI 写的版本)
Anthropic ──自定义接口──▶ 你的数据库工具(为 Anthropic 写的版本)
Gemini ──自定义接口──▶ 你的数据库工具(为 Gemini 写的版本)
同一个工具,维护 N 份代码
【有了 MCP 后】
OpenAI ─┐
Anthropic ─┼──── MCP 标准协议 ────▶ 你的数据库工具(一份代码)
Gemini ─┘
写一次,所有支持 MCP 的模型都能用
MCP 是 Anthropic 于 2024 年发布的开放协议,已被多家 AI 工具和平台支持,包括 Cursor、Windsurf、Zed 等编辑器,以及各类 AI 助手。
3.2 MCP 的三层结构
┌──────────────────────┐
│ MCP Host(宿主) │ ← AI 应用本体(如 Claude Desktop、Cursor)
│ AI 模型运行在这里 │ 负责发起连接、路由请求
└──────────┬───────────┘
│ MCP 标准协议(JSON-RPC over stdio / HTTP)
┌──────────┴───────────┐
│ MCP Client(客户端) │ ← Host 内置,管理与 Server 的连接
└──────────┬───────────┘
│
┌───────┴─────────────────┐
▼ ▼
┌──────────────┐ ┌──────────────┐
│ MCP Server A │ │ MCP Server B │
│ (文件系统) │ │ (数据库) │
│ • Tools │ │ • Tools │
│ • Resources │ │ • Resources │
│ • Prompts │ │ • Prompts │
└──────────────┘ └──────────────┘
每个 MCP Server 暴露三类能力:
| 能力类型 | 作用 | 例子 |
|---|---|---|
| Tools(工具) | 模型可以调用的函数,有输入输出,会产生副作用 | create_file、send_email、run_sql |
| Resources(资源) | 只读的内容,供模型读取参考,不产生副作用 | 文档内容、数据库 schema、配置文件 |
| Prompts(提示词) | 预定义的提示词模板,可带参数 | “代码审查模板”、“周报生成模板” |
3.3 通信协议:MCP 消息格式
MCP 基于 JSON-RPC 2.0,通信方式可以是 stdio(标准输入/输出,适合本地进程)或 HTTP + SSE(适合远程服务)。连接建立的握手流程:
Client Server
initialize ─────────────────────────────▶
(声明自己支持的协议版本)
检查版本兼容性
◀─────────────── initialized
(返回支持的能力列表)
notifications/initialized ──────────────▶
(确认握手完成)
═══════════════ 正式通信开始 ════════════════
tools/list ─────────────────────────────▶
◀─────────────── [工具列表 + schema]
tools/call ─────────────────────────────▶
{name: "search_db", arguments: {...}}
◀─────────────── {content: [...], isError: false}
工具调用的消息长这样:
// 请求:调用工具
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "search_customer",
"arguments": { "customer_id": "C-12345" }
}
}
// 响应:成功
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{ "type": "text", "text": "{\"name\": \"张三\", \"email\": \"zhang@example.com\"}" }
],
"isError": false
}
}
// 响应:失败(注意 isError 标志)
{
"result": {
"content": [{ "type": "text", "text": "客户不存在" }],
"isError": true
}
}
失败用 isError 标志表达,而不是抛 HTTP 错误。这样模型能感知到失败并决定下一步。
3.4 实现一个 MCP Server
以下是一个最简单的 MCP Server 结构,以 Python 为例,展示核心概念:
# pip install mcp
from mcp.server import Server
from mcp.types import Tool, TextContent
app = Server("my-database-server")
# ① 声明工具列表(模型连接时会查询这个)
@app.list_tools()
async def list_tools():
return [
Tool(
name="search_customer",
# 描述非常重要!模型靠这个判断什么时候调用这个工具
description="""在客户数据库中搜索客户信息。
接受客户ID(如 C-12345)或邮箱地址。
返回:姓名、邮箱、注册日期、账户状态。
注意:这个工具只读,不会修改任何数据。""",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "客户ID 或 邮箱地址"}
},
"required": ["query"]
}
)
]
# ② 实现工具逻辑
@app.call_tool()
async def call_tool(name, arguments):
if name == "search_customer":
try:
result = query_database(arguments["query"])
return [TextContent(type="text", text=json.dumps(result))]
except Exception as e:
# 返回错误,使用 isError=True 而不是抛异常
return [TextContent(type="text", text=str(e))], True
# ③ 启动 Server(stdio 模式,适合本地工具)
if __name__ == "__main__":
import mcp.server.stdio
mcp.server.stdio.run(app)
最容易忽视的细节是工具描述。 描述是模型决定“要不要调用这个工具”的主要依据。描述越详细、边界越清晰,模型的选择就越准确。一个只写“搜索客户”的工具,和一个写清楚输入格式、返回字段、适用场景、边界情况的工具,实际使用效果天差地别。写描述时对照这个清单:
- 这个工具做什么? 用一句话说清楚目的。
- 什么时候用它? 区别于类似工具的适用场景。
- 输入格式是什么? 举例说明,比如“接受 C-12345 格式的 ID”。
- 返回什么? 列出关键字段,说明可能的空值情况。
- 有什么副作用? 是只读还是会修改数据?
3.5 什么时候用 MCP,什么时候直接调 API
| 场景 | 推荐 | 原因 |
|---|---|---|
| 工具需要被多个 AI 应用共用 | MCP Server | 写一次,多处使用;标准化便于维护 |
| 团队内部特定业务流程 | MCP Server | 可以共享给团队所有人使用的 AI 工具 |
| 简单的一次性脚本 | 直接调 API | 搭 MCP 架构的成本高于收益 |
| 该外部系统已有社区 MCP Server | 用社区现成的 | GitHub、Jira、Slack 等已有高质量实现 |
实用建议:先查有没有现成的社区 MCP Server(github.com/modelcontextprotocol/servers),数据库、Git、邮件、日历等标准化工具几乎都有现成实现。只有业务独特的场景才需要自己写。
3.6 MCP 核心要点
- 开放标准:基于 JSON-RPC,不绑定任何特定 AI 厂商,理论上任何模型都可支持。
- 三类能力:Tools(执行)、Resources(读取)、Prompts(模板),覆盖大多数集成需求。
- 描述驱动:工具描述的质量直接决定模型是否能正确调用。这是最容易被忽视的关键。
- isError 标志:错误通过 isError 字段返回,而非 HTTP 状态码,让 AI 能感知并处理失败。
关于本指南
本文聚焦于通用工程原理,所有概念和代码结构适用于 OpenAI、Anthropic、Google Gemini 等主流 LLM 平台,以及 LangChain、LlamaIndex 等开源框架。原版是一个静态文档站,2026 年 9 月整理成这篇文章。English version.