提示工程到Agent工程的进化Prompt Engineering已死Agent Engineering当立作者钟伊人 | 日期2026-07-29 | Week5 总结与趋势判断模块一范式转移的底层逻辑从静态提示到动态推理2023年是提示工程Prompt Engineering的黄金时代。工程师们耗费大量时间雕琢每一个提示词。Few-shot、Chain-of-Thought、ReAct等技巧层出不穷。但进入2025-2026年局面发生了根本性变化。大模型的上下文窗口从4K扩展到2M token。模型本身已具备强大的指令遵循能力。精心设计的提示词带来的边际收益急剧递减。核心观点提示工程并未消亡而是被吸收进了更宏大的Agent工程范式。维度提示工程时代Agent工程时代核心对象单条提示词自主推理循环上下文管理手动构造动态检索与压缩工具调用无或硬编码动态发现与编排错误处理重新提示自我修正循环评估方式人工评判自动化轨迹评估# 提示工程典型代码旧范式 def old_style_qa(question: str) - str: prompt f请按照以下步骤思考问题 步骤1理解问题 步骤2列出关键信息 步骤3逐步推理 步骤4给出答案 问题{question} return llm.generate(prompt) # Agent工程典型代码新范式 from typing import List, Dict, Any from dataclasses import dataclass from enum import Enum class AgentState(Enum): IDLE idle REASONING reasoning ACTING acting REFLECTING reflecting FINISHED finished dataclass class AgentStep: thought: str action: str | None observation: str | None state: AgentState class AgentEngine: 生产级Agent引擎核心框架 支持推理、行动、观察、反思的完整循环 def __init__( self, llm_client, max_iterations: int 15, tools: List[Any] | None None ): self.llm llm_client self.max_iterations max_iterations self.tools tools or [] self.history: List[AgentStep] [] def run(self, task: str) - Dict[str, Any]: 执行Agent推理循环 state AgentState.REASONING for i in range(self.max_iterations): if state AgentState.FINISHED: break step self._execute_step(task, state) self.history.append(step) state self._transition_state(step) return self._compile_result() def _execute_step(self, task: str, state: AgentState) - AgentStep: 执行单个推理步骤 context self._build_context(task) if state AgentState.REASONING: thought self.llm.generate( self._reasoning_prompt(context) ) action self._parse_action(thought) return AgentStep( thoughtthought, actionaction, observationNone, stateAgentState.ACTING ) elif state AgentState.ACTING: # 执行工具调用 observation self._execute_action(self.history[-1].action) return AgentStep( thought, actionNone, observationobservation, stateAgentState.REFLECTING ) elif state AgentState.REFLECTING: # 反思是否需要继续或可以结束 reflection self.llm.generate( self._reflection_prompt(context) ) if FINAL_ANSWER in reflection: return AgentStep( thoughtreflection, actionNone, observationNone, stateAgentState.FINISHED ) else: return AgentStep( thoughtreflection, actionNone, observationNone, stateAgentState.REASONING ) return AgentStep(, None, None, state) def _build_context(self, task: str) - str: 动态构建上下文支持长轨迹压缩 steps_desc [] for i, step in enumerate(self.history): desc f步骤{i1}:\n if step.thought: desc f 思考{step.thought}\n if step.action: desc f 行动{step.action}\n if step.observation: desc f 观察{step.observation[:200]}\n steps_desc.append(desc) return f任务{task}\n\n \n.join(steps_desc) def _transition_state(self, step: AgentStep) - AgentState: return step.state def _parse_action(self, thought: str) - str | None: 解析LLM输出中的工具调用意图 import re pattern rACTION:\s*(\w)\((.*?)\) match re.search(pattern, thought) if match: return f{match.group(1)}({match.group(2)}) return None def _execute_action(self, action: str) - str: 执行工具调用生产级实现 for tool in self.tools: if tool.name in action: return tool.execute(action) return f未知工具: {action} def _reasoning_prompt(self, context: str) - str: return f{context}\n\n请思考下一步行动格式THOUGHT: ... ACTION: tool_name(args) def _reflection_prompt(self, context: str) - str: return f{context}\n\n评估当前进展决定继续推理或输出最终答案格式REFLECTION: ... FINAL_ANSWER: ... 或 CONTINUE def _compile_result(self) - Dict[str, Any]: return { success: self.history[-1].state AgentState.FINISHED, steps: len(self.history), trajectory: [ {thought: s.thought, action: s.action, observation: s.observation} for s in self.history ] }MermaidAgent工程的核心推理循环模块二Agent工程的核心技术栈工具调用Tool Calling的标准化Agent工程的首要任务是工具的标准化接入。Function Calling已从GPT的专属功能变为行业标准。2026年的主流框架均支持OpenAI兼容的工具调用格式。# 生产级工具定义规范兼容OpenAI/Anthropic/Google from pydantic import BaseModel, Field, validator from typing import List, Optional, Union import json class ToolParameter(BaseModel): 工具参数定义JSON Schema子集 type: str # string, integer, number, boolean, array, object description: str enum: Optional[List[str]] None items: Optional[ToolParameter] None # 用于array类型 required: Optional[List[str]] None # 用于object类型 class ToolDefinition(BaseModel): 统一工具定义兼容多厂商LLM name: str Field(..., min_length1, max_length64) description: str Field(..., min_length10) parameters: dict # JSON Schema格式 validator(name) def name_must_be_valid_python(cls, v): import re if not re.match(r^[a-zA-Z_][a-zA-Z0-9_]*$, v): raise ValueError(工具名称必须是有效的Python标识符) return v def to_openai_schema(self) - dict: 转换为OpenAI Function Calling格式 return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters } } def to_anthropic_schema(self) - dict: 转换为Anthropic Tool Use格式 return { name: self.name, description: self.description, input_schema: self.parameters } # 示例生产级搜索引擎工具 class SearchTool: def __init__(self, api_key: str, base_url: str): self.api_key api_key self.base_url base_url self.definition ToolDefinition( nameweb_search, description使用搜索引擎查询最新信息。输入应为简洁的搜索查询。, parameters{ type: object, properties: { query: { type: string, description: 搜索查询字符串不超过100字 }, max_results: { type: integer, description: 返回结果数量默认5, default: 5 } }, required: [query] } ) def execute(self, args: str) - str: 执行搜索生产级实现 import requests try: params json.loads(args) if isinstance(args, str) else args query params.get(query, ) max_results params.get(max_results, 5) # 实际生产环境中应调用真实搜索API response requests.get( f{self.base_url}/search, params{q: query, count: max_results}, headers{Authorization: fBearer {self.api_key}}, timeout10 ) response.raise_for_status() results response.json().get(results, []) # 格式化输出 formatted [] for i, r in enumerate(results[:max_results], 1): formatted.append( f[{i}] {r[title]}\n{r[url]}\n{r[snippet]} ) return \n\n.join(formatted) if formatted else 未找到相关结果 except requests.Timeout: return 搜索超时请稍后重试 except requests.RequestException as e: return f搜索失败{str(e)} except Exception as e: return f工具执行异常{str(e)}记忆管理从Context到MemoryAgent工程最关键的挑战之一是记忆管理。LLM的上下文窗口虽大但成本与延迟随之增加。优秀的Agent系统需要多层记忆架构。# 生产级多层记忆管理系统 from abc import ABC, abstractmethod from typing import List, Tuple import numpy as np class MemoryStore(ABC): 记忆存储抽象接口 abstractmethod def add(self, key: str, value: str, metadata: dict | None None): pass abstractmethod def search(self, query: str, top_k: int 5) - List[Tuple[str, float]]: pass abstractmethod def compact(self, max_tokens: int) - str: 压缩记忆到指定token数 pass class WorkingMemory: 工作记忆最近N轮对话直接放入上下文 容量~4K tokens适合大多数场景 def __init__(self, max_turns: int 10): self.max_turns max_turns self.turns: List[dict] [] def add_turn(self, role: str, content: str): self.turns.append({role: role, content: content}) if len(self.turns) self.max_turns: self.turns.pop(0) def get_context(self) - str: return \n.join( f{t[role]}: {t[content]} for t in self.turns ) class EpisodicMemory(MemoryStore): 情节记忆向量数据库存储的历史交互 容量~100K tokens按需检索 def __init__(self, embedding_model, vector_db): self.embedding_model embedding_model self.vector_db vector_db def add(self, key: str, value: str, metadata: dict | None None): embedding self.embedding_model.encode(value) self.vector_db.upsert( idkey, vectorembedding.tolist(), metadata{**(metadata or {}), content: value} ) def search(self, query: str, top_k: int 5) - List[Tuple[str, float]]: query_embedding self.embedding_model.encode(query) results self.vector_db.search( vectorquery_embedding.tolist(), top_ktop_k ) return [(r[metadata][content], r[score]) for r in results] class AgentMemoryManager: 统一记忆管理器 自动在三级记忆间调度信息 def __init__(self, working: WorkingMemory, episodic: EpisodicMemory | None None): self.working working self.episodic episodic def build_context_for_llm(self, current_query: str) - str: 为LLM调用构建最优上下文 parts [] # 1. 工作记忆必须 parts.append(## 最近对话\n self.working.get_context()) # 2. 情节记忆按需检索 if self.episodic: relevant self.episodic.search(current_query, top_k3) if relevant: parts.append(## 相关历史\n \n---\n.join( f[相关度:{score:.2f}] {content} for content, score in relevant )) return \n\n.join(parts)模块三生产环境中的Agent工程实践错误恢复与自我修正生产级Agent必须具备强大的错误处理能力。单纯的重试不足以应对复杂任务。Agent需要能够识别错误类型并采取不同恢复策略。# 生产级Agent错误恢复框架 from enum import Enum from typing import Callable, Any class ErrorType(Enum): TRANSIENT transient # 暂时性错误网络超时等 PERMANENT permanent # 永久性错误无效参数等 RATE_LIMIT rate_limit # 速率限制 CONTEXT_FULL context_full # 上下文溢出 UNKNOWN unknown # 未知错误 class ErrorRecoveryStrategy(Enum): RETRY retry # 简单重试 BACKOFF_RETRY backoff # 指数退避重试 SIMPLIFY simplify # 简化任务后重试 DECOMPOSE decompose # 分解任务 ASK_USER ask_user # 询问用户 ABORT abort # 放弃任务 class AgentWithRecovery: 具备错误恢复能力的生产级Agent def __init__(self, base_agent: AgentEngine): self.agent base_agent self.error_counts: dict[str, int] {} self.max_retries 3 def run_with_recovery(self, task: str) - Dict[str, Any]: attempt 0 last_error None while attempt self.max_retries: try: result self.agent.run(task) if result.get(success): return result # 分析失败原因 error_type self._classify_error(result) strategy self._choose_recovery_strategy(error_type, attempt) if strategy ErrorRecoveryStrategy.ABORT: break task self._apply_recovery(task, strategy, result) attempt 1 except Exception as e: last_error e error_type self._classify_exception(e) if error_type ErrorType.TRANSIENT and attempt self.max_retries - 1: import time time.sleep(2 ** attempt) # 指数退避 attempt 1 continue raise return { success: False, error: str(last_error), attempts: attempt } def _classify_error(self, result: dict) - ErrorType: 根据Agent执行结果分类错误 trajectory result.get(trajectory, []) for step in reversed(trajectory): obs step.get(observation, ) if not obs: continue if timeout in obs.lower() or 超时 in obs: return ErrorType.TRANSIENT if rate limit in obs.lower() or 限流 in obs: return ErrorType.RATE_LIMIT if context length in obs.lower() or 上下文 in obs: return ErrorType.CONTEXT_FULL if invalid in obs.lower() or 无效 in obs: return ErrorType.PERMANENT return ErrorType.UNKNOWN def _classify_exception(self, e: Exception) - ErrorType: 根据异常类型分类错误 from requests.exceptions import Timeout, ConnectionError if isinstance(e, (Timeout, ConnectionError)): return ErrorType.TRANSIENT return ErrorType.UNKNOWN def _choose_recovery_strategy(self, error_type: ErrorType, attempt: int) - ErrorRecoveryStrategy: 根据错误类型选择恢复策略 if error_type ErrorType.TRANSIENT: return ErrorRecoveryStrategy.BACKOFF_RETRY elif error_type ErrorType.RATE_LIMIT: return ErrorRecoveryStrategy.BACKOFF_RETRY elif error_type ErrorType.CONTEXT_FULL: return ErrorRecoveryStrategy.SIMPLIFY elif error_type ErrorType.PERMANENT and attempt 0: return ErrorRecoveryStrategy.DECOMPOSE else: return ErrorRecoveryStrategy.ABORT def _apply_recovery(self, task: str, strategy: ErrorRecoveryStrategy, result: dict) - str: 应用恢复策略可能修改任务 if strategy ErrorRecoveryStrategy.SIMPLIFY: # 让LLM简化任务 prompt f以下任务过于复杂导致失败请将其简化为更小的子任务只输出第一个子任务\n{task} return self.agent.llm.generate(prompt) elif strategy ErrorRecoveryStrategy.DECOMPOSE: prompt f将以下任务分解为3个独立的子任务每行一个\n{task} subtasks self.agent.llm.generate(prompt).split(\n) return subtasks[0] if subtasks else task return task评估体系从输出质量到轨迹质量提示工程时代评估主要看最终输出质量。Agent工程时代评估必须覆盖整个推理轨迹。# Agent轨迹评估框架 from dataclasses import dataclass from typing import List, Callable dataclass class TrajectoryMetric: name: str description: str evaluate: Callable[[List[AgentStep]], float] class AgentEvaluator: Agent执行质量的多维度评估器 def __init__(self): self.metrics: List[TrajectoryMetric] [] self._register_default_metrics() def _register_default_metrics(self): self.metrics.append(TrajectoryMetric( nameefficiency, description完成任务所需的步骤数越少越好, evaluatelambda steps: max(0, 1.0 - len(steps) / 20) )) self.metrics.append(TrajectoryMetric( nametool_accuracy, description工具调用的准确率, evaluateself._calc_tool_accuracy )) self.metrics.append(TrajectoryMetric( namereasoning_coherence, description推理链的连贯性, evaluateself._calc_reasoning_coherence )) def _calc_tool_accuracy(self, steps: List[AgentStep]) - float: 计算工具调用准确率 action_steps [s for s in steps if s.action] if not action_steps: return 1.0 successful sum( 1 for s in action_steps if s.observation and 失败 not in s.observation and error not in s.observation.lower() ) return successful / len(action_steps) def _calc_reasoning_coherence(self, steps: List[AgentStep]) - float: 计算推理连贯性基于步骤间引用关系 if len(steps) 2: return 1.0 coherent_transitions 0 for i in range(1, len(steps)): prev steps[i-1] curr steps[i] # 检查当前步骤是否引用了前一步的观察结果 if curr.thought and prev.observation: if any( keyword in curr.thought for keyword in [根据, 基于, 根据上一步, based on, according to] ): coherent_transitions 1 return coherent_transitions / (len(steps) - 1) def evaluate_trajectory(self, steps: List[AgentStep]) - dict: 对完整轨迹进行多维度评估 scores {} for metric in self.metrics: try: scores[metric.name] metric.evaluate(steps) except Exception: scores[metric.name] 0.0 # 综合得分加权平均 weights {efficiency: 0.2, tool_accuracy: 0.4, reasoning_coherence: 0.4} overall sum(scores.get(k, 0) * v for k, v in weights.items()) return { overall_score: round(overall, 3), dimension_scores: scores, total_steps: len(steps), passed: overall 0.6 }模块四Prompt Engineering真的死了吗辩证看待吸收而非替代断言Prompt Engineering已死是一种吸引眼球的表述。更准确的说法是提示工程已不再是独立学科而是Agent工程的基础组件。现在的提示工程工作集中在以下几个方向系统提示词System Prompt的工程化设计定义Agent的角色、能力和约束设计工具调用的输出格式规范注入领域知识和安全规则Few-shot示例的动态选择根据任务类型检索最相关的示例示例的压缩与去重上下文管理的提示策略长对话的场景保持技术记忆压缩的提示设计# 现代Agent的系统提示词工程化模板 SYSTEM_PROMPT_TEMPLATE # 角色定义 你是一个专业的{role}具备以下核心能力 {capabilities} # 工具使用规范 你可以使用以下工具来完成任务 {tool_descriptions} # 推理格式要求 每次推理必须严格按照以下格式输出 THOUGHT: 你的思考过程 ACTION: 工具名(参数) 或 无无需工具时 当任务完成时在THOUGHT中包含FINAL_ANSWER:然后直接给出答案。 # 错误恢复 - 如果工具调用失败分析原因并重试最多重试3次。 - 如果任务过于复杂将其分解为子任务逐步完成。 # 安全约束 {safety_constraints} # 输出要求 - 使用简体中文回答 - 代码片段必须包含完整错误处理 - 不要编造信息 def build_system_prompt(role: str, tools: List[ToolDefinition], **kwargs) - str: 动态构建系统提示词工程化方法 tool_descriptions \n.join( f- {t.name}: {t.description}\n 参数{json.dumps(t.parameters, ensure_asciiFalse)} for t in tools ) return SYSTEM_PROMPT_TEMPLATE.format( rolerole, capabilitieskwargs.get(capabilities, 通用问题解决能力), tool_descriptionstool_descriptions, safety_constraintskwargs.get(safety_constraints, 遵守法律法规不生成有害内容) )Mermaid提示工程与Agent工程的关系模块五趋势判断与行动建议2026年下半年Agent工程的关键趋势趋势1多Agent协作成为标配单个Agent的能力上限已显现。复杂任务需要多个专用Agent协作完成。Orchestrator-Worker模式正在标准化。# 多Agent协作框架核心代码 class AgentOrchestrator: 多Agent编排器 支持动态任务分配和结果聚合 def __init__(self): self.agents: dict[str, AgentEngine] {} self.shared_memory [] def register_agent(self, name: str, agent: AgentEngine): self.agents[name] agent def orchestrate(self, task: str) - dict: 编排多Agent协作完成任务 使用LLM动态决定任务分配 # 第一步任务分解 plan self._decompose_task(task) results {} for step in plan[steps]: agent_name step[assignee] subtask step[task] if agent_name not in self.agents: results[step[id]] {error: fAgent {agent_name} 未注册} continue # 执行子任务传入共享记忆 agent self.agents[agent_name] context self._build_shared_context(step[depends_on], results) result agent.run(f{subtask}\n\n参考信息\n{context}) results[step[id]] result self.shared_memory.append({ step_id: step[id], result: result }) return self._aggregate_results(results) def _decompose_task(self, task: str) - dict: 使用LLM将任务分解为子任务计划 返回格式{steps: [{id: , task: , assignee: , depends_on: []}]} agent_descriptions \n.join( f- {name}: {getattr(agent, description, 通用Agent)} for name, agent in self.agents.items() ) prompt f将以下任务分解为子任务并分配给最合适的Agent。 可用Agent {agent_descriptions} 任务{task} 输出JSON格式包含steps数组每个元素有id、task、assignee、depends_on字段。 # 实际实现中调用LLM return {steps: []} # 简化 def _build_shared_context(self, depends_on: List[str], results: dict) - str: 构建共享上下文汇聚依赖步骤的结果 context_parts [] for dep_id in (depends_on or []): if dep_id in results: res results[dep_id] context_parts.append(f步骤{dep_id}的结果{res}) return \n.join(context_parts) if context_parts else 无 def _aggregate_results(self, results: dict) - dict: 聚合所有子任务结果 return { success: all(r.get(success) for r in results.values()), subtask_results: results, summary: f完成了{len(results)}个子任务 }趋势2本地化Agent框架崛起云端LLM的延迟和成本促使本地化部署。2026年本地运行70B参数模型已成为可能。Agent框架正在适配本地模型的特性。趋势3Agent专用评估基准成熟AgentBench、AgentEval等基准持续迭代。企业开始建立自己的Agent评估流水线。离线评估与在线监控并重。行动建议对于已经在做提示工程的开发者立即开始将现有提示词封装为Tool构建第一个Agent原型重点学习强化学习中的ReAct/Reflexion等推理框架工具建设建立自己的工具库这是Agent工程的核心资产评估先行在构建第一个Agent时就设计评估指标对于团队负责人重新定义岗位提示工程师→Agent工程师技术选型LangChain vs LlamaIndex vs 自研框架的决策树成本控制Agent的Token消耗是提示工程的10-100倍需要精细管理纯技术总结Agent工程是以自主推理循环为核心的新型范式吸收而非替代提示工程核心技术栈工具标准化OpenAI兼容格式、多层记忆管理工作/情节/语义、错误恢复分类策略模式生产级实现要点状态机管理推理循环、指数退避重试、轨迹评估效率/准确率/连贯性三维度多Agent协作是下一个主战场Orchestrator-Worker模式正在标准化行动建议将提示词封装为Tool建立工具库设计评估流水线
提示工程到Agent工程的进化：Prompt Engineering已死，Agent Engineering当立？
提示工程到Agent工程的进化Prompt Engineering已死Agent Engineering当立作者钟伊人 | 日期2026-07-29 | Week5 总结与趋势判断模块一范式转移的底层逻辑从静态提示到动态推理2023年是提示工程Prompt Engineering的黄金时代。工程师们耗费大量时间雕琢每一个提示词。Few-shot、Chain-of-Thought、ReAct等技巧层出不穷。但进入2025-2026年局面发生了根本性变化。大模型的上下文窗口从4K扩展到2M token。模型本身已具备强大的指令遵循能力。精心设计的提示词带来的边际收益急剧递减。核心观点提示工程并未消亡而是被吸收进了更宏大的Agent工程范式。维度提示工程时代Agent工程时代核心对象单条提示词自主推理循环上下文管理手动构造动态检索与压缩工具调用无或硬编码动态发现与编排错误处理重新提示自我修正循环评估方式人工评判自动化轨迹评估# 提示工程典型代码旧范式 def old_style_qa(question: str) - str: prompt f请按照以下步骤思考问题 步骤1理解问题 步骤2列出关键信息 步骤3逐步推理 步骤4给出答案 问题{question} return llm.generate(prompt) # Agent工程典型代码新范式 from typing import List, Dict, Any from dataclasses import dataclass from enum import Enum class AgentState(Enum): IDLE idle REASONING reasoning ACTING acting REFLECTING reflecting FINISHED finished dataclass class AgentStep: thought: str action: str | None observation: str | None state: AgentState class AgentEngine: 生产级Agent引擎核心框架 支持推理、行动、观察、反思的完整循环 def __init__( self, llm_client, max_iterations: int 15, tools: List[Any] | None None ): self.llm llm_client self.max_iterations max_iterations self.tools tools or [] self.history: List[AgentStep] [] def run(self, task: str) - Dict[str, Any]: 执行Agent推理循环 state AgentState.REASONING for i in range(self.max_iterations): if state AgentState.FINISHED: break step self._execute_step(task, state) self.history.append(step) state self._transition_state(step) return self._compile_result() def _execute_step(self, task: str, state: AgentState) - AgentStep: 执行单个推理步骤 context self._build_context(task) if state AgentState.REASONING: thought self.llm.generate( self._reasoning_prompt(context) ) action self._parse_action(thought) return AgentStep( thoughtthought, actionaction, observationNone, stateAgentState.ACTING ) elif state AgentState.ACTING: # 执行工具调用 observation self._execute_action(self.history[-1].action) return AgentStep( thought, actionNone, observationobservation, stateAgentState.REFLECTING ) elif state AgentState.REFLECTING: # 反思是否需要继续或可以结束 reflection self.llm.generate( self._reflection_prompt(context) ) if FINAL_ANSWER in reflection: return AgentStep( thoughtreflection, actionNone, observationNone, stateAgentState.FINISHED ) else: return AgentStep( thoughtreflection, actionNone, observationNone, stateAgentState.REASONING ) return AgentStep(, None, None, state) def _build_context(self, task: str) - str: 动态构建上下文支持长轨迹压缩 steps_desc [] for i, step in enumerate(self.history): desc f步骤{i1}:\n if step.thought: desc f 思考{step.thought}\n if step.action: desc f 行动{step.action}\n if step.observation: desc f 观察{step.observation[:200]}\n steps_desc.append(desc) return f任务{task}\n\n \n.join(steps_desc) def _transition_state(self, step: AgentStep) - AgentState: return step.state def _parse_action(self, thought: str) - str | None: 解析LLM输出中的工具调用意图 import re pattern rACTION:\s*(\w)\((.*?)\) match re.search(pattern, thought) if match: return f{match.group(1)}({match.group(2)}) return None def _execute_action(self, action: str) - str: 执行工具调用生产级实现 for tool in self.tools: if tool.name in action: return tool.execute(action) return f未知工具: {action} def _reasoning_prompt(self, context: str) - str: return f{context}\n\n请思考下一步行动格式THOUGHT: ... ACTION: tool_name(args) def _reflection_prompt(self, context: str) - str: return f{context}\n\n评估当前进展决定继续推理或输出最终答案格式REFLECTION: ... FINAL_ANSWER: ... 或 CONTINUE def _compile_result(self) - Dict[str, Any]: return { success: self.history[-1].state AgentState.FINISHED, steps: len(self.history), trajectory: [ {thought: s.thought, action: s.action, observation: s.observation} for s in self.history ] }MermaidAgent工程的核心推理循环模块二Agent工程的核心技术栈工具调用Tool Calling的标准化Agent工程的首要任务是工具的标准化接入。Function Calling已从GPT的专属功能变为行业标准。2026年的主流框架均支持OpenAI兼容的工具调用格式。# 生产级工具定义规范兼容OpenAI/Anthropic/Google from pydantic import BaseModel, Field, validator from typing import List, Optional, Union import json class ToolParameter(BaseModel): 工具参数定义JSON Schema子集 type: str # string, integer, number, boolean, array, object description: str enum: Optional[List[str]] None items: Optional[ToolParameter] None # 用于array类型 required: Optional[List[str]] None # 用于object类型 class ToolDefinition(BaseModel): 统一工具定义兼容多厂商LLM name: str Field(..., min_length1, max_length64) description: str Field(..., min_length10) parameters: dict # JSON Schema格式 validator(name) def name_must_be_valid_python(cls, v): import re if not re.match(r^[a-zA-Z_][a-zA-Z0-9_]*$, v): raise ValueError(工具名称必须是有效的Python标识符) return v def to_openai_schema(self) - dict: 转换为OpenAI Function Calling格式 return { type: function, function: { name: self.name, description: self.description, parameters: self.parameters } } def to_anthropic_schema(self) - dict: 转换为Anthropic Tool Use格式 return { name: self.name, description: self.description, input_schema: self.parameters } # 示例生产级搜索引擎工具 class SearchTool: def __init__(self, api_key: str, base_url: str): self.api_key api_key self.base_url base_url self.definition ToolDefinition( nameweb_search, description使用搜索引擎查询最新信息。输入应为简洁的搜索查询。, parameters{ type: object, properties: { query: { type: string, description: 搜索查询字符串不超过100字 }, max_results: { type: integer, description: 返回结果数量默认5, default: 5 } }, required: [query] } ) def execute(self, args: str) - str: 执行搜索生产级实现 import requests try: params json.loads(args) if isinstance(args, str) else args query params.get(query, ) max_results params.get(max_results, 5) # 实际生产环境中应调用真实搜索API response requests.get( f{self.base_url}/search, params{q: query, count: max_results}, headers{Authorization: fBearer {self.api_key}}, timeout10 ) response.raise_for_status() results response.json().get(results, []) # 格式化输出 formatted [] for i, r in enumerate(results[:max_results], 1): formatted.append( f[{i}] {r[title]}\n{r[url]}\n{r[snippet]} ) return \n\n.join(formatted) if formatted else 未找到相关结果 except requests.Timeout: return 搜索超时请稍后重试 except requests.RequestException as e: return f搜索失败{str(e)} except Exception as e: return f工具执行异常{str(e)}记忆管理从Context到MemoryAgent工程最关键的挑战之一是记忆管理。LLM的上下文窗口虽大但成本与延迟随之增加。优秀的Agent系统需要多层记忆架构。# 生产级多层记忆管理系统 from abc import ABC, abstractmethod from typing import List, Tuple import numpy as np class MemoryStore(ABC): 记忆存储抽象接口 abstractmethod def add(self, key: str, value: str, metadata: dict | None None): pass abstractmethod def search(self, query: str, top_k: int 5) - List[Tuple[str, float]]: pass abstractmethod def compact(self, max_tokens: int) - str: 压缩记忆到指定token数 pass class WorkingMemory: 工作记忆最近N轮对话直接放入上下文 容量~4K tokens适合大多数场景 def __init__(self, max_turns: int 10): self.max_turns max_turns self.turns: List[dict] [] def add_turn(self, role: str, content: str): self.turns.append({role: role, content: content}) if len(self.turns) self.max_turns: self.turns.pop(0) def get_context(self) - str: return \n.join( f{t[role]}: {t[content]} for t in self.turns ) class EpisodicMemory(MemoryStore): 情节记忆向量数据库存储的历史交互 容量~100K tokens按需检索 def __init__(self, embedding_model, vector_db): self.embedding_model embedding_model self.vector_db vector_db def add(self, key: str, value: str, metadata: dict | None None): embedding self.embedding_model.encode(value) self.vector_db.upsert( idkey, vectorembedding.tolist(), metadata{**(metadata or {}), content: value} ) def search(self, query: str, top_k: int 5) - List[Tuple[str, float]]: query_embedding self.embedding_model.encode(query) results self.vector_db.search( vectorquery_embedding.tolist(), top_ktop_k ) return [(r[metadata][content], r[score]) for r in results] class AgentMemoryManager: 统一记忆管理器 自动在三级记忆间调度信息 def __init__(self, working: WorkingMemory, episodic: EpisodicMemory | None None): self.working working self.episodic episodic def build_context_for_llm(self, current_query: str) - str: 为LLM调用构建最优上下文 parts [] # 1. 工作记忆必须 parts.append(## 最近对话\n self.working.get_context()) # 2. 情节记忆按需检索 if self.episodic: relevant self.episodic.search(current_query, top_k3) if relevant: parts.append(## 相关历史\n \n---\n.join( f[相关度:{score:.2f}] {content} for content, score in relevant )) return \n\n.join(parts)模块三生产环境中的Agent工程实践错误恢复与自我修正生产级Agent必须具备强大的错误处理能力。单纯的重试不足以应对复杂任务。Agent需要能够识别错误类型并采取不同恢复策略。# 生产级Agent错误恢复框架 from enum import Enum from typing import Callable, Any class ErrorType(Enum): TRANSIENT transient # 暂时性错误网络超时等 PERMANENT permanent # 永久性错误无效参数等 RATE_LIMIT rate_limit # 速率限制 CONTEXT_FULL context_full # 上下文溢出 UNKNOWN unknown # 未知错误 class ErrorRecoveryStrategy(Enum): RETRY retry # 简单重试 BACKOFF_RETRY backoff # 指数退避重试 SIMPLIFY simplify # 简化任务后重试 DECOMPOSE decompose # 分解任务 ASK_USER ask_user # 询问用户 ABORT abort # 放弃任务 class AgentWithRecovery: 具备错误恢复能力的生产级Agent def __init__(self, base_agent: AgentEngine): self.agent base_agent self.error_counts: dict[str, int] {} self.max_retries 3 def run_with_recovery(self, task: str) - Dict[str, Any]: attempt 0 last_error None while attempt self.max_retries: try: result self.agent.run(task) if result.get(success): return result # 分析失败原因 error_type self._classify_error(result) strategy self._choose_recovery_strategy(error_type, attempt) if strategy ErrorRecoveryStrategy.ABORT: break task self._apply_recovery(task, strategy, result) attempt 1 except Exception as e: last_error e error_type self._classify_exception(e) if error_type ErrorType.TRANSIENT and attempt self.max_retries - 1: import time time.sleep(2 ** attempt) # 指数退避 attempt 1 continue raise return { success: False, error: str(last_error), attempts: attempt } def _classify_error(self, result: dict) - ErrorType: 根据Agent执行结果分类错误 trajectory result.get(trajectory, []) for step in reversed(trajectory): obs step.get(observation, ) if not obs: continue if timeout in obs.lower() or 超时 in obs: return ErrorType.TRANSIENT if rate limit in obs.lower() or 限流 in obs: return ErrorType.RATE_LIMIT if context length in obs.lower() or 上下文 in obs: return ErrorType.CONTEXT_FULL if invalid in obs.lower() or 无效 in obs: return ErrorType.PERMANENT return ErrorType.UNKNOWN def _classify_exception(self, e: Exception) - ErrorType: 根据异常类型分类错误 from requests.exceptions import Timeout, ConnectionError if isinstance(e, (Timeout, ConnectionError)): return ErrorType.TRANSIENT return ErrorType.UNKNOWN def _choose_recovery_strategy(self, error_type: ErrorType, attempt: int) - ErrorRecoveryStrategy: 根据错误类型选择恢复策略 if error_type ErrorType.TRANSIENT: return ErrorRecoveryStrategy.BACKOFF_RETRY elif error_type ErrorType.RATE_LIMIT: return ErrorRecoveryStrategy.BACKOFF_RETRY elif error_type ErrorType.CONTEXT_FULL: return ErrorRecoveryStrategy.SIMPLIFY elif error_type ErrorType.PERMANENT and attempt 0: return ErrorRecoveryStrategy.DECOMPOSE else: return ErrorRecoveryStrategy.ABORT def _apply_recovery(self, task: str, strategy: ErrorRecoveryStrategy, result: dict) - str: 应用恢复策略可能修改任务 if strategy ErrorRecoveryStrategy.SIMPLIFY: # 让LLM简化任务 prompt f以下任务过于复杂导致失败请将其简化为更小的子任务只输出第一个子任务\n{task} return self.agent.llm.generate(prompt) elif strategy ErrorRecoveryStrategy.DECOMPOSE: prompt f将以下任务分解为3个独立的子任务每行一个\n{task} subtasks self.agent.llm.generate(prompt).split(\n) return subtasks[0] if subtasks else task return task评估体系从输出质量到轨迹质量提示工程时代评估主要看最终输出质量。Agent工程时代评估必须覆盖整个推理轨迹。# Agent轨迹评估框架 from dataclasses import dataclass from typing import List, Callable dataclass class TrajectoryMetric: name: str description: str evaluate: Callable[[List[AgentStep]], float] class AgentEvaluator: Agent执行质量的多维度评估器 def __init__(self): self.metrics: List[TrajectoryMetric] [] self._register_default_metrics() def _register_default_metrics(self): self.metrics.append(TrajectoryMetric( nameefficiency, description完成任务所需的步骤数越少越好, evaluatelambda steps: max(0, 1.0 - len(steps) / 20) )) self.metrics.append(TrajectoryMetric( nametool_accuracy, description工具调用的准确率, evaluateself._calc_tool_accuracy )) self.metrics.append(TrajectoryMetric( namereasoning_coherence, description推理链的连贯性, evaluateself._calc_reasoning_coherence )) def _calc_tool_accuracy(self, steps: List[AgentStep]) - float: 计算工具调用准确率 action_steps [s for s in steps if s.action] if not action_steps: return 1.0 successful sum( 1 for s in action_steps if s.observation and 失败 not in s.observation and error not in s.observation.lower() ) return successful / len(action_steps) def _calc_reasoning_coherence(self, steps: List[AgentStep]) - float: 计算推理连贯性基于步骤间引用关系 if len(steps) 2: return 1.0 coherent_transitions 0 for i in range(1, len(steps)): prev steps[i-1] curr steps[i] # 检查当前步骤是否引用了前一步的观察结果 if curr.thought and prev.observation: if any( keyword in curr.thought for keyword in [根据, 基于, 根据上一步, based on, according to] ): coherent_transitions 1 return coherent_transitions / (len(steps) - 1) def evaluate_trajectory(self, steps: List[AgentStep]) - dict: 对完整轨迹进行多维度评估 scores {} for metric in self.metrics: try: scores[metric.name] metric.evaluate(steps) except Exception: scores[metric.name] 0.0 # 综合得分加权平均 weights {efficiency: 0.2, tool_accuracy: 0.4, reasoning_coherence: 0.4} overall sum(scores.get(k, 0) * v for k, v in weights.items()) return { overall_score: round(overall, 3), dimension_scores: scores, total_steps: len(steps), passed: overall 0.6 }模块四Prompt Engineering真的死了吗辩证看待吸收而非替代断言Prompt Engineering已死是一种吸引眼球的表述。更准确的说法是提示工程已不再是独立学科而是Agent工程的基础组件。现在的提示工程工作集中在以下几个方向系统提示词System Prompt的工程化设计定义Agent的角色、能力和约束设计工具调用的输出格式规范注入领域知识和安全规则Few-shot示例的动态选择根据任务类型检索最相关的示例示例的压缩与去重上下文管理的提示策略长对话的场景保持技术记忆压缩的提示设计# 现代Agent的系统提示词工程化模板 SYSTEM_PROMPT_TEMPLATE # 角色定义 你是一个专业的{role}具备以下核心能力 {capabilities} # 工具使用规范 你可以使用以下工具来完成任务 {tool_descriptions} # 推理格式要求 每次推理必须严格按照以下格式输出 THOUGHT: 你的思考过程 ACTION: 工具名(参数) 或 无无需工具时 当任务完成时在THOUGHT中包含FINAL_ANSWER:然后直接给出答案。 # 错误恢复 - 如果工具调用失败分析原因并重试最多重试3次。 - 如果任务过于复杂将其分解为子任务逐步完成。 # 安全约束 {safety_constraints} # 输出要求 - 使用简体中文回答 - 代码片段必须包含完整错误处理 - 不要编造信息 def build_system_prompt(role: str, tools: List[ToolDefinition], **kwargs) - str: 动态构建系统提示词工程化方法 tool_descriptions \n.join( f- {t.name}: {t.description}\n 参数{json.dumps(t.parameters, ensure_asciiFalse)} for t in tools ) return SYSTEM_PROMPT_TEMPLATE.format( rolerole, capabilitieskwargs.get(capabilities, 通用问题解决能力), tool_descriptionstool_descriptions, safety_constraintskwargs.get(safety_constraints, 遵守法律法规不生成有害内容) )Mermaid提示工程与Agent工程的关系模块五趋势判断与行动建议2026年下半年Agent工程的关键趋势趋势1多Agent协作成为标配单个Agent的能力上限已显现。复杂任务需要多个专用Agent协作完成。Orchestrator-Worker模式正在标准化。# 多Agent协作框架核心代码 class AgentOrchestrator: 多Agent编排器 支持动态任务分配和结果聚合 def __init__(self): self.agents: dict[str, AgentEngine] {} self.shared_memory [] def register_agent(self, name: str, agent: AgentEngine): self.agents[name] agent def orchestrate(self, task: str) - dict: 编排多Agent协作完成任务 使用LLM动态决定任务分配 # 第一步任务分解 plan self._decompose_task(task) results {} for step in plan[steps]: agent_name step[assignee] subtask step[task] if agent_name not in self.agents: results[step[id]] {error: fAgent {agent_name} 未注册} continue # 执行子任务传入共享记忆 agent self.agents[agent_name] context self._build_shared_context(step[depends_on], results) result agent.run(f{subtask}\n\n参考信息\n{context}) results[step[id]] result self.shared_memory.append({ step_id: step[id], result: result }) return self._aggregate_results(results) def _decompose_task(self, task: str) - dict: 使用LLM将任务分解为子任务计划 返回格式{steps: [{id: , task: , assignee: , depends_on: []}]} agent_descriptions \n.join( f- {name}: {getattr(agent, description, 通用Agent)} for name, agent in self.agents.items() ) prompt f将以下任务分解为子任务并分配给最合适的Agent。 可用Agent {agent_descriptions} 任务{task} 输出JSON格式包含steps数组每个元素有id、task、assignee、depends_on字段。 # 实际实现中调用LLM return {steps: []} # 简化 def _build_shared_context(self, depends_on: List[str], results: dict) - str: 构建共享上下文汇聚依赖步骤的结果 context_parts [] for dep_id in (depends_on or []): if dep_id in results: res results[dep_id] context_parts.append(f步骤{dep_id}的结果{res}) return \n.join(context_parts) if context_parts else 无 def _aggregate_results(self, results: dict) - dict: 聚合所有子任务结果 return { success: all(r.get(success) for r in results.values()), subtask_results: results, summary: f完成了{len(results)}个子任务 }趋势2本地化Agent框架崛起云端LLM的延迟和成本促使本地化部署。2026年本地运行70B参数模型已成为可能。Agent框架正在适配本地模型的特性。趋势3Agent专用评估基准成熟AgentBench、AgentEval等基准持续迭代。企业开始建立自己的Agent评估流水线。离线评估与在线监控并重。行动建议对于已经在做提示工程的开发者立即开始将现有提示词封装为Tool构建第一个Agent原型重点学习强化学习中的ReAct/Reflexion等推理框架工具建设建立自己的工具库这是Agent工程的核心资产评估先行在构建第一个Agent时就设计评估指标对于团队负责人重新定义岗位提示工程师→Agent工程师技术选型LangChain vs LlamaIndex vs 自研框架的决策树成本控制Agent的Token消耗是提示工程的10-100倍需要精细管理纯技术总结Agent工程是以自主推理循环为核心的新型范式吸收而非替代提示工程核心技术栈工具标准化OpenAI兼容格式、多层记忆管理工作/情节/语义、错误恢复分类策略模式生产级实现要点状态机管理推理循环、指数退避重试、轨迹评估效率/准确率/连贯性三维度多Agent协作是下一个主战场Orchestrator-Worker模式正在标准化行动建议将提示词封装为Tool建立工具库设计评估流水线
