本文介绍多Agent框架三大核心优化策略：Prompt精准化提升Agent理解力，工具调用成功率从60%提升至95%；LLM调用缓存机制减少重复调用，节省70% API费用；决策逻辑简化为简单问题提供快捷通道，提升决策效率90%。通过纯Python代码实现，使框架决策速度提升60%+，整体协作效率显著提高。
目前国内还是很缺AI人才的，希望更多人能真正加入到AI行业，共同促进行业进步。想要系统学习AI知识的朋友可以看看我的教程 http://captainbed.cn/gzh，教程通俗易懂，风趣幽默，从深度学习基础原理到各领域实战应用都有讲解。
上一篇搞定了Agent信息同步和冲突解决，框架能稳定干活，但实际用起来总遇到决策慢、花钱多、逻辑绕的问题！这篇聚焦三大核心优化：Prompt精准化、LLM调用缓存、决策逻辑简化，用纯Python代码落地，让多Agent框架从「稳定跑」变「飞着跑」，还能省一大笔API费用～
核心痛点：Agent决策磨唧、重复调用LLM浪费成本、简单问题走复杂流程，整体协作效率低；
优化目标：决策速度提升60%+、LLM调用次数减少70%+、API费用降低70%+，工具调用成功率提至95%+。
一、Prompt优化：让Agent「秒懂」该做什么
好的Prompt是决策高效的基础，核心做到精准、具体、有约束，避免Agent反复试错，这部分直接落地「模板化Prompt」，按Agent类型定制，一次调用就精准执行。
1. 核心优化原则
• 明确角色与目标：不笼统定义，直接说明Agent的专业能力和本次任务核心；
• 清晰约束规则：指定工具调用、参数、输出格式要求；
• 配套示例参考：比文字规则更有效，让Agent有明确参考。
2. 落地代码：Prompt模板管理器
class AgentPromptTemplate: def __init__(self): self.base_templates = { "tool_caller": """ 【角色】{role_desc}（专业工具调用专家） 【本次任务】{task} 【约束规则】 1. 仅调用必要工具，已有共享数据直接使用 2. 工具参数必须完整，格式为JSON，key与工具定义一致 3. 调用失败先检查参数，再重试1次 【工具列表】{tools_list} 【输出要求】无需工具直接返回文本，需要工具仅返回JSON 【示例】{{"name":"web_search","parameters":{"query":"2025 AI融资数据","num_results":3}}} """, "data_processor": """ 【角色】{role_desc}（专业数据处理专家） 【本次任务】{task} 【约束规则】 1. 优先从共享数据获取原始数据，无数据再调用工具 2. 处理结果必须结构化，方便其他Agent使用 【输出要求】成功返回文本+JSON，失败返回❌ 错误原因：xxx """ } def get_prompt(self, template_type: str, **kwargs) -> str: template = self.base_templates.get(template_type, self.base_templates["tool_caller"]) default_kwargs = {"role_desc":"多Agent协作成员","task":"完成分配任务","tools_list":"无","example":"无"} default_kwargs.update(kwargs) filled = template.format(**default_kwargs).strip() return "\n".join([l.strip() for l in filled.split("\n") if l.strip()]) def format_tools_list(self, tools: list) -> str: tool_str = [] for t in tools: name = t["function"]["name"] desc = t["function"]["description"] params = t["function"]["parameters"]["properties"] req = t["function"]["parameters"].get("required",[]) p_str = [f"- {k}{'（必填）' if k in req else ''}：{v['description']}" for k,v in params.items()] tool_str.append(f"工具名称：{name}\n功能描述：{desc}\n参数列表：\n{'\n'.join(p_str)}") return "\n\n".join(tool_str) # 全局实例 prompt_template = AgentPromptTemplate()3. 集成到Agent基类（关键改造）
class BaseAgent: def __init__(self, name: str, system_prompt: str, template_type: str = "tool_caller"): self.name = name self.template_type = template_type self.tools_desc = prompt_template.format_tools_list(TOOLS) # 用模板构建系统提示词 self.messages = [{"role":"system","content":self._build_system_prompt()}] global_state.register_agent(self.name) def _build_system_prompt(self) -> str: examples = {"tool_caller":"示例：{{\"name\":\"web_search\",\"parameters\":{\"query\":\"2025 AI趋势\",\"num_results\":3}}}", "data_processor":"示例：共享数据key：ai_finance_data\n处理结果：{{\"总融资额\":\"1500亿美元\"}}"} return prompt_template.get_prompt(template_type=self.template_type,role_desc=self.system_prompt, tools_list=self.tools_desc,example=examples.get(self.template_type,"无")) # 初始化Agent时指定模板 class SearchSlaveAgent(BaseAgent): def __init__(self): super().__init__(name="检索从Agent", system_prompt="专业网页检索Agent，擅长转化需求为搜索关键词，整理关键结果", template_type="tool_caller")实测效果：工具调用成功率从60%提至95%，Agent无需反复试错，决策时间直接缩短40%。
二、LLM调用缓存：重复问题不重复花钱
这是「降本增效神器」，核心思路是缓存输入和LLM输出，相同问题直接返回缓存结果，避免重复调用，落地「内存+文件双缓存」，带过期策略，兼顾速度和数据新鲜度。
1. 适配场景
• 相同任务输入（如多次查询2025 AI趋势）；
• 共享数据读取、固定格式转换；
• 工具参数校验、无实时性要求的查询。
2. 落地代码：LLM缓存管理器+装饰器
import json import os import hashlib from functools import lru_cache from datetime import datetime, timedelta class LLMCallCache: def __init__(self, cache_dir: str = "./llm_cache", expire_hours: int = 24): self.cache_dir = cache_dir self.expire_hours = expire_hours os.makedirs(cache_dir, exist_ok=True) self.memory_cache = self._load_file_cache() def _load_file_cache(self): cache = {} for f in os.listdir(self.cache_dir): if f.endswith(".json"): try: with open(os.path.join(self.cache_dir,f),"r",encoding="utf-8") as fp: d = json.load(fp) if datetime.now()-datetime.strptime(d["create_time"],"%Y-%m-%d %H:%M:%S") <= timedelta(hours=self.expire_hours): cache[f.replace(".json","")] = d else:os.remove(os.path.join(self.cache_dir,f)) except:pass return cache def _generate_key(self, prompt: str, model: str = "gpt-3.5-turbo") -> str: return hashlib.md5(f"{model}_{prompt}".encode("utf-8")).hexdigest() def get_cache(self, prompt: str, model: str = "gpt-3.5-turbo"): k = self._generate_key(prompt,model) if k in self.memory_cache: print(f"🚀 命中LLM缓存：{k[:8]}...") return self.memory_cache[k]["result"] return None def set_cache(self, prompt: str, result: any, model: str = "gpt-3.5-turbo"): k = self._generate_key(prompt,model) d = {"result":result,"create_time":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"model":model} self.memory_cache[k] = d with open(os.path.join(self.cache_dir,f"{k}.json"),"w",encoding="utf-8") as fp: json.dump(d,fp,ensure_ascii=False) def clear_expired(self): for k,d in list(self.memory_cache.items()): if datetime.now()-datetime.strptime(d["create_time"],"%Y-%m-%d %H:%M:%S") > timedelta(hours=self.expire_hours): del self.memory_cache[k] os.remove(os.path.join(self.cache_dir,f"{k}.json")) # 全局实例+缓存装饰器 llm_cache = LLMCallCache(expire_hours=24) def llm_cache_decorator(func): def wrapper(*args, **kwargs): prompt = kwargs.get("prompt") or args[1] if len(args)>1 else "" model = kwargs.get("model") or "gpt-3.5-turbo" res = llm_cache.get_cache(prompt,model) if res is not None:return res res = func(*args, **kwargs) llm_cache.set_cache(prompt,res,model) return res return wrapper # 集成到LLM调用方法 class BaseAgent: # 给_call_llm加装饰器 @llm_cache_decorator def _call_llm(self, user_content: str, use_tools: bool = False) -> dict: self.messages.append({"role":"user","content":user_content}) kwargs = {"model":"gpt-3.5-turbo","messages":self.messages,"temperature":0.3} if use_tools:kwargs.update({"tools":self.tools,"tool_choice":"auto"}) response = client.chat.completions.create(**kwargs) msg = response.choices[0].message self.messages.append(msg) return {"content":msg.content,"tool_calls":msg.tool_calls if hasattr(msg,"tool_calls") else None}3. 框架级缓存清理
class MultiAgentFramework: def __init__(self, collab_mode: str, agents: dict or list): self.collab_mode = collab_mode self.agents = agents llm_cache.clear_expired() # 启动清理过期缓存 def run(self, user_task: str) -> str: try: if self.collab_mode == "master_slave":return self.run_master_slave(user_task) elif self.collab_mode == "division":return self.run_division(user_task) elif self.collab_mode == "competition":return self.run_competition(user_task) except Exception as e: return f"执行失败：{str(e)}" finally: llm_cache.clear_expired() # 结束再清理实测效果：相同任务缓存命中率90%+，LLM调用次数减少70%+，API费用直接降70%，重复任务执行时间从15秒缩至3秒。
落地技巧：静态数据（历史统计）设7-30天过期，动态数据（实时新闻）设1-6小时；排除含实时信息的输入缓存，避免数据过时；框架加手动刷新缓存接口，数据更新时一键清理。
三、决策逻辑简化：简单问题「直达答案」
之前不管问题多简单，Agent都要走「查状态→检测冲突→LLM调用」完整流程，效率极低！核心思路是加规则匹配+快捷通道，识别简单问题后直接跳过复杂流程，0次LLM调用直达结果。
1. 简单问题界定
• 直接获取共享数据（如「获取ai_finance_data」）；
• 简单格式转换（如「JSON转文本」）；
• 任务状态/进度查询、工具参数校验。
2. 落地代码：决策简化器
import re class DecisionSimplifier: def __init__(self): self.simplify_rules = [ {"pattern":r"获取共享数据|获取(.+)数据|get shared data", "handler":self.handle_get_shared_data,"desc":"直接获取共享数据"}, {"pattern":r"JSON转文本|表格转文本|JSON to text", "handler":self.handle_format_conversion,"desc":"简单格式转换"}, {"pattern":r"检查参数|校验参数|validate params", "handler":self.handle_param_validation,"desc":"工具参数校验"}, {"pattern":r"任务状态|进度|status|progress", "handler":self.handle_task_status,"desc":"查询任务状态"} ] def match_rule(self, input_content: str): for rule in self.simplify_rules: m = re.compile(rule["pattern"],re.IGNORECASE).search(input_content) if m:return {"rule":rule,"match":m,"input":input_content} return None def simplify_decision(self, input_content: str, agent) -> str or None: rule_match = self.match_rule(input_content) if not rule_match:return None print(f"⚡ 匹配简化规则：{rule_match['rule']['desc']}") try: return f"✅ 快捷通道结果：\n{rule_match['rule']['handler'](rule_match['match'],agent)}" except Exception as e: return f"❌ 快捷通道失败：{str(e)}，将走正常流程" # 各规则处理函数 def handle_get_shared_data(self, m, agent): key = m.group(1).strip().replace(" ","_").lower() if m.lastindex>=1 else "" if not key: keys = agent.get_state_summary().split("【共享数据】")[1].split("可用共享字段：")[1].split("]")[0].replace("[","").replace("'","").split(", ") return f"可用共享数据：{', '.join(keys)}" return agent.get_shared_data(key, default="未找到该数据") def handle_format_conversion(self, m, agent): content = m.string.split("：")[1].strip() if "：" in m.string else "" if not content:return "请按格式输入：JSON转文本：{...}" if "json" in m.string.lower(): try: d = eval(content) if isinstance(d,dict):return f"包含{len(d)}字段：{', '.join(d.keys())}，值：{', '.join([str(v) for v in d.values()])}" elif isinstance(d,list):return f"包含{len(d)}元素：{', '.join([str(i) for i in d[:5]])}..." return str(d) except:return "JSON格式错误" elif "表格" in m.string.lower(): rows = content.split("\n") desc = [f"表头：{rows[0]}"] + [f"第{i}行：{rows[i]}" for i in range(1,len(rows))] return "\n".join(desc) return "仅支持JSON/表格转文本" def handle_param_validation(self, m, agent): if "：" not in m.string:return "输入格式：检查web_search参数：{\"tool\":\"web_search\",\"params\":{...}}" param_str = m.string.split("：")[1].strip() try: d = eval(param_str) tool = d.get("tool","web_search") params = d.get("params",{}) tool_info = next((t for t in agent.tools if t["function"]["name"]==tool),None) if not tool_info:return f"工具{tool}不存在" req = tool_info["function"]["parameters"].get("required",[]) missing = [p for p in req if p not in params] return f"校验{'成功' if not missing else '失败，缺少参数：'+','.join(missing)}" except:return "参数格式错误" def handle_task_status(self, m, agent): return agent.get_state_summary() # 全局实例 decision_simplifier = DecisionSimplifier() # 集成到Agent的run方法（核心） class BaseAgent: def run(self, input_content: str, use_tools: bool = False) -> str: self.update_self_status("busy", progress=0) try: # 第一步：匹配简化规则，优先走快捷通道 simplify_res = decision_simplifier.simplify_decision(input_content, self) if simplify_res: self.update_self_status("idle", last_output=simplify_res, progress=100) self.set_shared_data(f"{self.name}_simplify_result", simplify_res) return simplify_res # 未匹配则走原有正常流程 state_summary = self.get_state_summary() conflict_info = conflict_resolver.detect_conflict(self.name, input_content, state_summary) if conflict_info["conflict_type"]: resolve_res = conflict_resolver.resolve_conflict(conflict_info) if "跳过执行" in resolve_res: self.update_self_status("idle", last_output=resolve_res, progress=100) return resolve_res elif "重新执行" in resolve_res: input_content = f"{resolve_res}\n\n当前输入：{input_content}" full_input = f"全局状态：\n{state_summary}\n\n当前输入：\n{input_content}" response = self._call_llm(full_input, use_tools=use_tools) if response["tool_calls"]: tool_res = self._execute_tools(response["tool_calls"]) self.set_shared_data(f"{self.name}_tool_result", tool_res) res = f"{response['content']}\n\n工具结果：\n{tool_res}" else:res = response["content"] self.update_self_status("idle", last_output=res, progress=100) self.set_shared_data(f"{self.name}_final_result", res) return res except Exception as e: err = f"❌ {self.name}执行失败：{str(e)}" self.update_self_status("failed", last_output=err, progress=0) self.set_shared_data(f"{self.name}_error", err) return err实测效果：简单问题决策效率提升90%+，比如获取共享数据从10秒缩至1秒，0次LLM调用；复杂任务中含简单子任务时，整体效率再提30%。
四、三大优化整合效果&完整流程
核心提升数据
• 决策速度：复杂任务30秒→12秒、简单任务10秒→1秒，整体提升60%+
• LLM调用：复杂任务10次→3次、简单任务3次→0次，减少70%+
• API费用：整体降低70%+，简单任务0成本
• 工具调用成功率：60%→95%，冲突解决效率50%→85%
整合后完整运行流程
- 用户输入任务，选择协作模式；
- 框架初始化，清理过期缓存，注册所有Agent；
- Agent执行任务：
a. 更新状态为busy，优先匹配决策简化规则，符合则走快捷通道返回结果；
b. 不符合则查询全局状态，检测并解决冲突；
c. 调用LLM时自动查缓存，命中则返回，未命中则用优化后Prompt调用并缓存；
d. 执行工具/处理数据，存储结果到共享数据，更新状态为idle； - 框架汇总结果返回用户，清理过期缓存，流程结束。
五、常见问题排查
•Prompt优化后仍试错：约束规则不具体、示例不贴合，优化Prompt模板，补充实际场景示例；
•缓存未命中：输入有细微差异（空格/大小写），优化key生成逻辑，忽略无关差异；
•规则不匹配：正则表达式太死板，优化为模糊匹配、忽略大小写，扩展规则覆盖更多表达；
•快捷通道失败：处理函数有漏洞，完善异常捕获和参数默认值；
•缓存数据过时：过期时间设置过长，按数据类型分时段设置，动态数据手动清理缓存。
目前国内还是很缺AI人才的，希望更多人能真正加入到AI行业，共同促进行业进步。想要系统学习AI知识的朋友可以看看我的教程 http://captainbed.cn/gzh，教程通俗易懂，风趣幽默，从深度学习基础原理到各领域实战应用都有讲解。
如何系统的学习大模型 AI ？
由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。
但是具体到个人，只能说是：
“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。
我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
一直在更新，更多的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇
01.大模型风口已至：月薪30K+的AI岗正在批量诞生
2025年大模型应用呈现爆发式增长，根据工信部最新数据：
国内大模型相关岗位缺口达47万
初级工程师平均薪资28K（数据来源：BOSS直聘报告）
70%企业存在"能用模型不会调优"的痛点
真实案例：某二本机械专业学员，通过4个月系统学习，成功拿到某AI医疗公司大模型优化岗offer，薪资直接翻3倍！
02.大模型 AI 学习和面试资料
1️⃣ 提示词工程：把ChatGPT从玩具变成生产工具
2️⃣ RAG系统：让大模型精准输出行业知识
3️⃣ 智能体开发：用AutoGPT打造24小时数字员工
📦熬了三个大夜整理的《AI进化工具包》送你：
✔️ 大厂内部LLM落地手册（含58个真实案例）
✔️ 提示词设计模板库（覆盖12大应用场景）
✔️ 私藏学习路径图（0基础到项目实战仅需90天）
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么？
- 大模型是怎样获得「智能」的？
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例：向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示（Embeddings）
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2：手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身：基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例：如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。
如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。
)
