核心内容摘要
Selaoban不止于表
文章介绍多Agent框架三大核心优化Prompt精准化提高工具调用成功率至95%LLM调用缓存减少70% API费用决策逻辑简化使简单问题直达答案。
这些优化使框架决策速度提升60%整体效率显著提高同时降低成本。
文章提供了完整代码实现和实测效果解决了Agent决策慢、花费高、逻辑绕的问题让多Agent框架从稳定跑变为飞着跑。
目前国内还是很缺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时代未来的就业机会在哪里答案就藏在大模型的浪潮里。
从ChatGPT、DeepSeek等日常工具到自然语言处理、计算机视觉、多模态等核心领域技术普惠化、应用垂直化与生态开源化正催生Prompt工程师、自然语言处理、计算机视觉工程师、大模型算法工程师、AI应用产品经理等AI岗位。
掌握大模型技能就是把握高薪未来。
那么普通人如何抓住大模型风口AI技术的普及对个人能力提出了新的要求在AI时代持续学习和适应新技术变得尤为重要。
无论是企业还是个人都需要不断更新知识体系提升与AI协作的能力以适应不断变化的工作环境。
因此这里给大家整理了一份《2026最新大模型全套学习资源》包括2026最新大模型学习路线、大模型书籍、视频教程、项目实战、最新行业报告、面试题、AI产品经理入门到精通等带你从零基础入门到精通快速掌握大模型技术由于篇幅有限有需要的小伙伴可以扫码获取
成长路线图学习规划要学习一门新的技术作为新手一定要先学习成长路线图方向不对努力白费。
这里我们为新手和想要进一步提升的专业人士准备了一份详细的学习成长路线图和规划。
大模型经典PDF书籍书籍和学习文档资料是学习大模型过程中必不可少的我们精选了一系列深入探讨大模型技术的书籍和学习文档它们由领域内的顶尖专家撰写内容全面、深入、详尽为你学习大模型提供坚实的理论基础。
书籍含电子版PDF
大模型视频教程对于很多自学或者没有基础的同学来说书籍这些纯文字类的学习教材会觉得比较晦涩难以理解因此我们提供了丰富的大模型视频教程以动态、形象的方式展示技术概念帮助你更快、更轻松地掌握核心知识。
大模型项目实战学以致用当你的理论知识积累到一定程度就需要通过项目实战在实际操作中检验和巩固你所学到的知识同时为你找工作和职业发展打下坚实的基础。
大模型行业报告行业分析主要包括对不同行业的现状、趋势、问题、机会等进行系统地调研和评估以了解哪些行业更适合引入大模型的技术和应用以及在哪些方面可以发挥大模型的优势。
大模型面试题面试不仅是技术的较量更需要充分的准备。
在你已经掌握了大模型技术之后就需要开始准备面试我们将提供精心整理的大模型面试题库涵盖当前面试中可能遇到的各种技术问题让你在面试中游刃有余。
为什么大家都在学AI大模型随着AI技术的发展企业对人才的需求从“单一技术”转向 “AI行业”双背景。
企业对人才的需求从“单一技术”转向 “AI行业”双背景。
金融AI、制造AI、医疗AI等跨界岗位薪资涨幅达30%-50%。
同时很多人面临优化裁员近期科技巨头英特尔裁员2万人传统岗位不断缩减因此转行AI势在必行这些资料有用吗这份资料由我们和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理现任上海殷泊信息科技CEO其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证服务航天科工、国家电网等1000企业以第一作者在IEEE Transactions发表论文50篇获NASA JPL火星探测系统强化学习专利等35项中美专利。
本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。
资料内容涵盖了从入门到进阶的各类视频教程和实战项目无论你是小白还是有些技术基础的技术人员这份资料都绝对能帮助你提升薪资待遇转行大模型岗位。
大模型全套学习资料已整理打包有需要的小伙伴可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】