核心内容摘要
桃传媒：不止于“桃”，开启内容新纪元
本文介绍多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类型定制一次调用就精准执行。
核心优化原则• 明确角色与目标不笼统定义直接说明Agent的专业能力和本次任务核心• 清晰约束规则指定工具调用、参数、输出格式要求• 配套示例参考比文字规则更有效让Agent有明确参考。
落地代码Prompt模板管理器class AgentPromptTemplate: def __init__(self): self.base_templates { tool_caller: 【角色】{role_desc}专业工具调用专家 【本次任务】{task} 【约束规则】
仅调用必要工具已有共享数据直接使用
工具参数必须完整格式为JSONkey与工具定义一致
调用失败先检查参数再重试1次 【工具列表】{tools_list} 【输出要求】无需工具直接返回文本需要工具仅返回JSON 【示例】} , data_processor: 【角色】{role_desc}专业数据处理专家 【本次任务】{task} 【约束规则】
优先从共享数据获取原始数据无数据再调用工具
处理结果必须结构化方便其他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()
集成到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:示例}, data_processor:示例共享数据keyai_finance_data\n处理结果} return prompt_template.get_prompt(template_typeself.template_type,role_descself.system_prompt, tools_listself.tools_desc,exampleexamples.get(self.template_type,无)) # 初始化Agent时指定模板 class SearchSlaveAgent(BaseAgent): def __init__(self): super().__init__(name检索从Agent, system_prompt专业网页检索Agent擅长转化需求为搜索关键词整理关键结果, template_typetool_caller)实测效果工具调用成功率从60%提至95%Agent无需反复试错决策时间直接缩短40%。
LLM调用缓存重复问题不重复花钱这是「降本增效神器」核心思路是缓存输入和LLM输出相同问题直接返回缓存结果避免重复调用落地「内存文件双缓存」带过期策略兼顾速度和数据新鲜度。
适配场景• 相同任务输入如多次查询2025 AI趋势• 共享数据读取、固定格式转换• 工具参数校验、无实时性要求的查询。
落地代码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
: self.cache_dir cache_dir self.expire_hours expire_hours os.makedirs(cache_dir, exist_okTrue) 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,encodingutf-
as fp: d json.load(fp) if datetime.now()-datetime.strptime(d[create_time],%Y-%m-%d %H:%M:%S) timedelta(hoursself.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-
5-turbo) - str: return hashlib.md5(f{model}_{prompt}.encode(utf-
).hexdigest() def get_cache(self, prompt: str, model: str gpt-
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-
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,encodingutf-
as fp: json.dump(d,fp,ensure_asciiFalse) 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(hoursself.expire_hours): del self.memory_cache[k] os.remove(os.path.join(self.cache_dir,f{k}.json)) # 全局实例缓存装饰器 llm_cache LLMCallCache(expire_hours
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-
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-
5-turbo,messages:self.messages,temperature:
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}
框架级缓存清理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秒。
落地技巧静态数据历史统计设
天过期动态数据实时新闻设
小时排除含实时信息的输入缓存避免数据过时框架加手动刷新缓存接口数据更新时一键清理。
决策逻辑简化简单问题「直达答案」之前不管问题多简单Agent都要走「查状态→检测冲突→LLM调用」完整流程效率极低核心思路是加规则匹配快捷通道识别简单问题后直接跳过复杂流程0次LLM调用直达结果。
简单问题界定• 直接获取共享数据如「获取ai_finance_data」• 简单格式转换如「JSON转文本」• 任务状态/进度查询、工具参数校验。
落地代码决策简化器import re class DecisionSimplifier: def __init__(self): self.simplify_rules [ {pattern:r获取共享数据|获取(.)数据|get shared data, handler:self.handle_get_shared_data,desc:直接获取共享数据}, {pattern:rJSON转文本|表格转文本|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(
.strip().replace( ,_).lower() if m.lastindex1 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
try: # 第一步匹配简化规则优先走快捷通道 simplify_res decision_simplifier.simplify_decision(input_content, self) if simplify_res: self.update_self_status(idle, last_outputsimplify_res, progress
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_outputresolve_res, progress
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_toolsuse_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_outputres, progress
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_outputerr, progress
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%整合后完整运行流程用户输入任务选择协作模式框架初始化清理过期缓存注册所有AgentAgent执行任务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%免费】
大模型风口已至月薪30K的AI岗正在批量诞生2025年大模型应用呈现爆发式增长根据工信部最新数据国内大模型相关岗位缺口达47万初级工程师平均薪资28K数据来源BOSS直聘报告70%企业存在能用模型不会调优的痛点真实案例某二本机械专业学员通过4个月系统学习成功拿到某AI医疗公司大模型优化岗offer薪资直接翻3倍
大模型 AI 学习和面试资料1️⃣ 提示词工程把ChatGPT从玩具变成生产工具2️⃣ RAG系统让大模型精准输出行业知识3️⃣ 智能体开发用AutoGPT打造24小时数字员工熬了三个大夜整理的《AI进化工具包》送你✔️ 大厂内部LLM落地手册含58个真实案例✔️ 提示词设计模板库覆盖12大应用场景✔️ 私藏学习路径图0基础到项目实战仅需90天第一阶段10天初阶应用该阶段让大家对大模型 AI有一个最前沿的认识对大模型 AI 的理解超过 95% 的人可以在相关讨论时发表高级、不跟风、又接地气的见解别人只会和 AI 聊天而你能调教 AI并能用代码将大模型和业务衔接。
大模型 AI 能干什么大模型是怎样获得「智能」的用好 AI 的核心心法大模型应用业务架构大模型应用技术架构代码示例向 GPT-
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天内完成所有的任务那你堪称天才。
然而如果你能完成
% 的内容你就已经开始具备成为一名大模型 AI 的正确特征了。
这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】