提示工程架构师进阶:Agentic AI创新应用的高级设计技巧
想象你是一家科技公司的"AI指挥官",过去指挥AI做事时,需要像教3岁孩子搭积木一样:“第一步拿起红色方块,第二步放在蓝色方块上面,第三步…”——这就是传统LLM应用的"被动响应模式"。而现在,你希望AI能像经验丰富的项目经理:接到"组织一场技术峰会"的任务后,能自己拆解目标(确定主题、邀请嘉宾、安排场地)、协调资源(联系供应商、分配团队任务)、处理突发问题(嘉宾临时请假时自动调整议程)——这就是
提示工程架构师进阶:Agentic AI创新应用的高级设计技巧
关键词:Agentic AI、提示工程架构师、智能体设计、多智能体协作、自主决策机制、工具调用策略、动态提示优化
摘要:随着大语言模型(LLM)技术的飞速发展,Agentic AI(智能体AI)已成为推动AI应用从"被动响应"向"主动决策"升级的核心力量。本文专为希望进阶的提示工程架构师打造,系统讲解Agentic AI创新应用的高级设计技巧。通过生活化比喻、可视化架构图、可运行代码示例和实战案例,从核心概念解构、架构设计原则、决策算法优化到多智能体协作策略,全方位揭秘如何构建具备自主规划、动态学习和复杂任务处理能力的智能体系统。无论你是AI应用开发者、系统架构师还是技术管理者,都能从本文获得将Agentic AI落地到实际业务场景的完整知识体系和实战方法论。
背景介绍
目的和范围
想象你是一家科技公司的"AI指挥官",过去指挥AI做事时,需要像教3岁孩子搭积木一样:“第一步拿起红色方块,第二步放在蓝色方块上面,第三步…”——这就是传统LLM应用的"被动响应模式"。而现在,你希望AI能像经验丰富的项目经理:接到"组织一场技术峰会"的任务后,能自己拆解目标(确定主题、邀请嘉宾、安排场地)、协调资源(联系供应商、分配团队任务)、处理突发问题(嘉宾临时请假时自动调整议程)——这就是Agentic AI的"主动决策模式"。
本文的核心目的是帮助提示工程架构师掌握从"被动提示设计"到"主动智能体系统架构"的进阶能力,具体包括:
- 理解Agentic AI与传统LLM应用的本质区别
- 掌握智能体系统的核心组件设计原则
- 学会多智能体协作的架构模式和通信协议
- 优化智能体的自主决策能力和动态提示策略
- 将Agentic AI落地到实际业务场景的实战技巧
范围限定:聚焦Agentic AI的"设计技巧"而非基础概念,假设读者已掌握提示工程基础知识(如指令设计、 few-shot学习)和Python编程能力,不涉及底层LLM模型训练技术。
预期读者
本文适合以下三类"AI建造者":
- 资深提示工程师:希望从单轮提示设计升级到复杂智能体系统架构
- AI应用架构师:需要将Agentic AI落地到具体业务场景(如智能客服、自动化办公)
- 技术团队负责人:规划AI系统演进路线,评估Agentic AI的技术可行性和投入产出比
文档结构概述
本文采用"概念→架构→算法→实战→应用"的递进式结构,共分为8个核心章节:
- 背景介绍:Agentic AI的兴起背景与学习路径
- 核心概念与联系:用生活化比喻拆解智能体系统的关键组件
- 智能体架构设计原则:从单智能体到多智能体的架构模式
- 自主决策算法与动态提示优化:数学原理与代码实现
- 项目实战:构建"智能研发项目管家"多智能体系统
- 实际应用场景:金融、医疗、制造等领域的落地案例
- 工具与资源推荐:提升开发效率的框架与最佳实践
- 未来趋势与挑战:技术瓶颈与突破方向
术语表
核心术语定义
术语 | 通俗解释 | 专业定义 |
---|---|---|
Agentic AI | “会自己找事做的AI” | 具备感知环境、自主决策、执行任务和学习优化能力的AI系统 |
智能体(Agent) | “AI团队里的一个角色” | 能独立完成特定功能的AI实体,包含感知、决策、执行模块 |
多智能体系统(MAS) | “AI团队协作网络” | 由多个智能体通过通信协议协作完成复杂任务的系统 |
自主决策机制 | “AI自己做决定的思考流程” | 智能体基于目标、环境和历史经验选择行动方案的算法逻辑 |
工具调用(Tool Use) | “AI使用工具的能力” | 智能体调用外部API/函数完成自身无法直接处理的任务(如查天气、算数据) |
动态提示(Dynamic Prompting) | “AI根据情况调整自己的指令” | 智能体根据任务进展、环境变化动态生成或修改提示的策略 |
记忆模块(Memory) | “AI的笔记本和经验库” | 存储智能体历史交互、任务状态和学习经验的组件 |
相关概念解释
- Agentic AI vs 传统LLM应用:传统LLM像"问答机器人",需要用户明确指令;Agentic AI像"自主助手",能主动拆解目标、调用工具、持续优化。
- 智能体 vs 机器人流程自动化(RPA):RPA是"按固定脚本执行的机器",Agentic AI是"能灵活应对变化的智能体"(如RPA只能按预设步骤填表格,而智能体能发现表格格式变化并调整策略)。
- 提示工程架构师 vs 普通提示工程师:普通提示工程师关注"如何写好单轮提示",提示工程架构师关注"如何设计由多个提示模块组成的智能体系统架构"。
缩略词列表
- Agentic AI:智能体人工智能(Agentic Artificial Intelligence)
- LLM:大语言模型(Large Language Model)
- MAS:多智能体系统(Multi-Agent System)
- MDP:马尔可夫决策过程(Markov Decision Process)
- RL:强化学习(Reinforcement Learning)
- API:应用程序接口(Application Programming Interface)
核心概念与联系
故事引入
“小明的智能家居升级记”
小明家有一套智能家居系统,但体验一直不太满意:
- 智能音箱只会"一问一答"(“打开客厅灯"→"已打开”),不会主动提醒"离家时忘关灯"
- 扫地机器人每天固定时间清扫,遇到小明在家工作时也会噪音扰民
- 空调和窗帘各自为战,不会根据天气和小明的作息自动调整
后来,小明家引入了一个**“智能家居总管家智能体”**,情况彻底改变:
- 感知环境:总管家连接所有设备,实时获取天气(外部API)、小明的位置(手机定位)、设备状态(灯、空调开关)
- 拆解目标:接到"提升居住舒适度"的总目标后,自动拆解为"节能"、“便捷”、"健康"三个子目标
- 多智能体协作:总管家不是单打独斗,而是协调多个专业智能体:
- 环境监测智能体:分析温度、湿度、PM2.5数据
- 作息学习智能体:记录小明的起床、工作、睡眠规律
- 设备控制智能体:具体操作灯光、空调、窗帘
- 自主决策:某天早上,环境监测智能体发现"室外温度骤降且PM2.5超标",总管家立即决策:
- 提前15分钟打开暖气(根据作息智能体的数据,小明通常7点起床)
- 关闭窗户并开启空气净化器(无需小明指令)
- 发送提醒:“今天天气寒冷且空气质量差,建议佩戴口罩”
- 动态调整:如果小明临时加班晚归,作息学习智能体会更新数据,总管家自动推迟扫地机器人的工作时间
这个故事里的"总管家+专业智能体团队",就是典型的Agentic AI系统。接下来,我们拆解它的核心组件和设计技巧。
核心概念解释(像给小学生讲故事一样)
核心概念一:智能体(Agent)——AI世界的"全能员工"
生活比喻:智能体就像餐厅里的"全能服务员",他需要:
- 看:观察顾客需求(比如注意到顾客频频看表,可能赶时间)
- 想:思考该怎么做(推荐快速上菜的菜品,优先安排厨房制作)
- 做:执行具体动作(下单、催菜、上菜)
- 记:记住顾客偏好(比如这位顾客上次点了微辣,这次主动询问是否需要同样口味)
专业解释:一个完整的智能体包含5个核心模块(简称"PERMA"):
- 感知(Perception):获取外部信息(如API调用、用户输入、传感器数据)
- 环境建模(Environment Modeling):理解信息含义(如将"用户说’热’“转化为"温度调节需求”)
- 决策(Reasoning/Decision):选择行动方案(如"是直接开空调,还是先检查窗户是否打开")
- 执行(Action):完成具体任务(如调用空调API设置26℃)
- 记忆(Memory):存储历史数据(如记录用户偏好温度为26℃)
为什么重要:传统LLM只有"思考"能力,而智能体通过PERMA模块组合,实现了"从思考到行动"的闭环。
核心概念二:多智能体协作(MAS)——AI团队的"分工与配合"
生活比喻:多智能体协作就像"学校运动会的接力赛团队":
- 短跑选手:负责起跑(快速响应)
- 中长跑选手:负责中途加速(持续执行)
- 冲刺选手:负责最后决胜(关键任务)
- 教练:协调团队策略(分配棒次、调整顺序)
专业解释:多智能体系统是多个智能体通过通信协议协作完成复杂任务的系统,包含3个关键要素:
- 角色划分:每个智能体有明确职责(如"数据分析智能体"、“决策智能体”、“执行智能体”)
- 通信机制:智能体之间交换信息的规则(如"请求-响应"模式、"广播"模式)
- 冲突解决:当智能体意见不一致时的协调策略(如"投票制"、“优先级制”)
为什么重要:单个智能体难以处理复杂任务(就像一个人无法同时完成接力赛的所有棒次),多智能体协作能通过分工提升效率和鲁棒性。
核心概念三:自主决策机制(Autonomous Decision)——AI的"独立思考能力"
生活比喻:自主决策机制就像"小学生做数学题的思考过程":
- 理解题目:"小明有5个苹果,给了小红2个,还剩几个?“→明确目标是"计算剩余苹果数”
- 选择方法:回忆学过的减法→"5-2=?"
- 执行计算:得出结果3
- 检查答案:"5-2确实是3,没错"→确认决策正确
专业解释:自主决策是智能体基于目标、环境和经验选择行动的过程,包含4个步骤:
- 目标拆解:将复杂目标分解为可执行的子目标(如"组织会议"→"确定时间→邀请参会者→准备材料")
- 行动规划:为每个子目标制定行动步骤(如"确定时间"→"查询参会者日历→找出共同空闲时段→预订会议室")
- 行动选择:从候选方案中选最优(如"上午10点和下午2点都是共同空闲,选择上午10点因为参会者精力更集中")
- 结果评估:判断行动是否达成目标(如"检查会议室是否成功预订")
为什么重要:自主决策是Agentic AI区别于传统LLM的核心特征——传统LLM需要用户明确指令,而智能体能自己决定"下一步该做什么"。
核心概念四:工具调用策略(Tool Use)——AI的"使用工具能力"
生活比喻:工具调用策略就像"厨师做菜时选择厨具":
- 切菜:用菜刀(专用工具)而非勺子(不匹配)
- 炒菜:用铁锅(适合中国菜)而非平底锅(效果差)
- 计时:用计时器(准确)而非凭感觉(易出错)
专业解释:工具调用是智能体使用外部API/函数完成自身无法直接处理的任务,包含3个关键策略:
- 工具选择:根据任务类型选择合适工具(如"查天气"用天气API,"算数据"用Python函数)
- 参数生成:正确填写工具所需参数(如调用天气API时需指定"城市=北京,日期=今天")
- 结果解析:将工具返回的原始数据转化为决策依据(如将天气API返回的"温度28℃,湿度60%“转化为"建议开启空调制冷”)
为什么重要:LLM本身的能力有限(如无法实时获取数据、不会复杂计算),工具调用让智能体突破这些限制,就像人类通过使用工具扩展能力边界。
核心概念五:动态提示优化(Dynamic Prompting)——AI的"临时抱佛脚"
生活比喻:动态提示优化就像"老师根据学生水平调整教案":
- 对基础好的学生:“直接做综合题,我只讲难点”
- 对基础弱的学生:“先复习知识点,再做简单例题”
专业解释:动态提示是智能体根据任务进展、环境变化或用户反馈动态调整提示内容的策略,包含3种优化方式:
- 任务自适应:根据任务复杂度调整提示详细度(如简单任务用"简洁指令",复杂任务用"分步引导")
- 用户自适应:根据用户特征调整提示风格(如对技术用户用专业术语,对普通用户用通俗语言)
- 反馈自适应:根据历史结果优化提示(如发现上次提示导致错误,本次自动加入"注意检查格式")
为什么重要:静态提示无法应对复杂多变的环境(就像一份固定教案无法满足所有学生),动态提示让智能体更灵活地适应不同场景。
核心概念之间的关系(用小学生能理解的比喻)
智能体与多智能体协作的关系:单个演员 vs 整个剧组
- 比喻:智能体是"剧组里的单个演员"(如男主角),多智能体协作是"整个剧组的配合"(导演、摄影师、演员共同完成电影拍摄)。
- 实例:"智能客服系统"中,“意图识别智能体”(识别用户问题类型)是演员,“多智能体协作"是它与"知识库智能体”(查答案)、“工单智能体”(转人工)的配合。
自主决策与工具调用的关系:大脑 vs 双手
- 比喻:自主决策是"大脑"(决定做什么),工具调用是"双手"(实际执行)。就像"大脑决定要写作业",“双手拿笔写字”。
- 实例:“数据分析智能体"的自主决策模块决定"需要计算用户留存率”,工具调用模块执行"调用Python的pandas库计算留存率"。
动态提示与智能体的关系:教练 vs 运动员
- 比喻:动态提示是"教练"(根据运动员状态调整训练计划),智能体是"运动员"(执行训练)。就像教练发现运动员疲劳,临时调整今天的训练强度。
- 实例:"代码生成智能体"发现用户是新手开发者,动态调整提示:“生成代码时附带详细注释,并解释每个步骤的作用”。
所有概念的整体关系:学校组织春游活动
- 总目标:组织一次安全有趣的春游(对应"复杂任务")
- 多智能体协作:校长(总协调)、班主任(执行)、后勤老师(准备物资)、司机(交通)(对应"角色划分")
- 自主决策:班主任决定"去植物园还是动物园"(基于学生年龄和天气)(对应"决策机制")
- 工具调用:后勤老师使用"采购平台"买零食,司机使用"导航软件"规划路线(对应"工具使用")
- 动态提示:校长发现天气预报有小雨,临时提醒"各班级带好雨伞"(对应"动态提示优化")
核心概念原理和架构的文本示意图(专业定义)
智能体(Agent)的PERMA架构
+---------------------+
| 智能体 |
| +---------------+ |
| | 感知模块 | | ← 获取外部信息(用户输入、API数据、传感器)
| +-------+-------+ |
| ↓ |
| +---------------+ |
| | 环境建模模块 | | ← 理解信息含义(自然语言解析、数据结构化)
| +-------+-------+ |
| ↓ |
| +---------------+ |
| | 决策规划模块 | | ← 选择行动方案(目标拆解、行动规划、冲突解决)
| +-------+-------+ |
| ↓ |
| +---------------+ |
| | 执行/工具模块 | | ← 执行行动(调用API、生成文本、控制设备)
| +-------+-------+ |
| ↓ |
| +---------------+ |
| | 记忆模块 | | ← 存储历史数据(短期记忆、长期记忆、经验库)
| +---------------+ |
+---------------------+
多智能体系统(MAS)的架构模式
- 主从式架构(如智能家居总管家)
+------------+ +------------+ +------------+
| 主智能体 | ←→ | 子智能体A | ←→ | 工具/API |
+------------+ +------------+ +------------+
←→ ←→
+------------+ +------------+ +------------+
| 子智能体B | ←→ | 子智能体C | ←→ | 外部系统 |
+------------+ +------------+ +------------+
- 平等式架构(如多专家协作系统)
+------------+ +------------+ +------------+
| 智能体A | ←→ | 智能体B | ←→ | 智能体C |
+------------+ +------------+ +------------+
←→ ←→ ←→
+--------------------------------------------------+
| 共享通信频道 |
+--------------------------------------------------+
Mermaid 流程图 (智能体工作流程)
核心算法原理 & 具体操作步骤
自主决策机制:基于MDP的强化学习决策算法
数学原理:马尔可夫决策过程(MDP)
智能体的自主决策可以用MDP描述,包含5个要素(S, A, P, R, γ):
- 状态(S):智能体当前所处的环境状态(如"用户问题未解决"、“已调用知识库工具”)
- 行动(A):智能体可执行的动作(如"继续追问用户"、“调用代码解释器”)
- 状态转移概率(P):执行行动后从状态s转移到s’的概率(如"调用知识库后,问题解决的概率为80%")
- 奖励(R):执行行动后获得的反馈(如"问题解决"得+10分,"调用无效工具"得-5分)
- 折扣因子(γ):未来奖励的权重(γ=0.9表示未来奖励的重要性是当前的90%)
智能体的目标是学习一个策略π(s→a),即"在状态s下选择行动a",使得累积奖励最大化:
Gt=Rt+1+γRt+2+γ2Rt+3+...=∑k=0∞γkRt+k+1 G_t = R_{t+1} + γR_{t+2} + γ^2R_{t+3} + ... = \sum_{k=0}^\infty γ^k R_{t+k+1} Gt=Rt+1+γRt+2+γ2Rt+3+...=k=0∑∞γkRt+k+1
Q-Learning算法:智能体如何学习"哪个行动更好"
Q-Learning是一种常用的强化学习算法,通过学习Q值(Q(s,a)) 表示"在状态s下执行行动a的预期累积奖励",更新公式为:
Q(s,a)←Q(s,a)+α[r+γmaxa′Q(s′,a′)−Q(s,a)] Q(s,a) \leftarrow Q(s,a) + α[r + γ \max_{a'} Q(s',a') - Q(s,a)] Q(s,a)←Q(s,a)+α[r+γa′maxQ(s′,a′)−Q(s,a)]
其中:
- α:学习率(0<α<1,控制更新幅度)
- r:当前奖励
- γ:折扣因子
- max_{a’} Q(s’,a’):下一个状态s’的最大Q值
Python实现:简单的客服智能体决策示例
假设我们要构建一个客服智能体,需要决策"是否调用知识库",状态S包含"用户问题复杂度"(低/高),行动A包含"直接回答"(a1)和"调用知识库"(a2)。
import numpy as np
# 初始化Q表:状态2个(0=低复杂度,1=高复杂度),行动2个(0=直接回答,1=调用知识库)
Q = np.zeros((2, 2)) # Q[状态][行动] = 预期奖励
# 定义状态转移概率P(s'|s,a):[当前状态][行动] = [(s',概率)]
P = [
[[(0, 0.8), (1, 0.2)], # 状态0(低复杂度)下行动0(直接回答):80%保持状态0,20%转移到状态1
[(0, 0.9), (1, 0.1)]], # 状态0下行动1(调用知识库):90%保持状态0,10%转移到状态1
[[(1, 0.7), (0, 0.3)], # 状态1(高复杂度)下行动0(直接回答):70%保持状态1,30%转移到状态0
[(1, 0.2), (0, 0.8)]] # 状态1下行动1(调用知识库):20%保持状态1,80%转移到状态0
]
# 定义奖励函数R(s,a):状态s下执行行动a的即时奖励
R = [
[5, 3], # 状态0:直接回答(a0)奖励5,调用知识库(a1)奖励3(因知识库调用有成本)
[-2, 8] # 状态1:直接回答(a0)奖励-2(可能答错),调用知识库(a1)奖励8
]
# Q-Learning参数
α = 0.1 # 学习率
γ = 0.9 # 折扣因子
episodes = 1000 # 训练回合数
# 训练过程
for _ in range(episodes):
s = np.random.choice([0, 1]) # 随机初始状态
while True:
# ε-贪婪策略选择行动:90%选Q值最大的行动,10%随机探索
if np.random.rand() < 0.1:
a = np.random.choice([0, 1])
else:
a = np.argmax(Q[s, :])
# 执行行动,获取下一个状态和奖励
next_states, probs = zip(*P[s][a])
s_prime = np.random.choice(next_states, p=probs)
r = R[s][a]
# 更新Q值
Q[s, a] = Q[s, a] + α * (r + γ * np.max(Q[s_prime, :]) - Q[s, a])
# 终止条件(简化:固定步数后结束)
if np.random.rand() < 0.1: # 10%概率结束当前回合
break
# 输出训练后的Q表
print("训练后的Q表:")
print("状态0(低复杂度问题):")
print(f" 直接回答(a0)Q值:{Q[0,0]:.2f}")
print(f" 调用知识库(a1)Q值:{Q[0,1]:.2f}")
print("状态1(高复杂度问题):")
print(f" 直接回答(a0)Q值:{Q[1,0]:.2f}")
print(f" 调用知识库(a1)Q值:{Q[1,1]:.2f}")
代码解读与决策逻辑
训练后,我们会得到类似这样的Q表:
状态0(低复杂度问题):
直接回答(a0)Q值:45.23
调用知识库(a1)Q值:38.15
状态1(高复杂度问题):
直接回答(a0)Q值:-12.45
调用知识库(a1)Q值:76.89
这意味着:
- 当用户问题复杂度低(状态0)时,智能体选择"直接回答"(Q值更高)
- 当用户问题复杂度高(状态1)时,智能体选择"调用知识库"(Q值更高)
这符合我们的预期:简单问题直接回答更高效,复杂问题需要调用工具辅助。
动态提示优化:基于用户反馈的提示调整算法
算法原理:反馈驱动的提示模板迭代
动态提示优化的核心是根据用户反馈调整提示模板,包含3个步骤:
- 收集反馈:用户对智能体输出的评价(如"有用/无用"、“清晰/模糊”)
- 分析原因:判断输出问题是否由提示导致(如"未指定格式"导致输出混乱)
- 调整提示:修改提示模板(如加入"请用JSON格式输出")
Python实现:动态提示优化器
from typing import Dict, List
class DynamicPromptOptimizer:
def __init__(self, base_prompt: str):
self.base_prompt = base_prompt # 基础提示模板
self.feedback_history: List[Dict] = [] # 反馈历史
self.optimization_rules = { # 优化规则:问题类型→提示调整
"格式混乱": "请使用清晰的标题和分点列表组织内容",
"信息不足": "请先确认用户的具体需求,再提供详细回答",
"语言生硬": "请使用更口语化的表达,避免专业术语"
}
def generate_prompt(self, user_query: str) -> str:
"""生成动态提示:基础模板+优化规则"""
# 分析历史反馈,提取需要应用的优化规则
applied_rules = set()
for feedback in self.feedback_history:
if feedback["rating"] < 3: # 评分低于3分(1-5分制)视为负面反馈
problem_type = feedback["problem_type"]
if problem_type in self.optimization_rules:
applied_rules.add(self.optimization_rules[problem_type])
# 构建最终提示
optimized_prompt = self.base_prompt
if applied_rules:
optimized_prompt += "\n\n额外要求:\n" + "\n".join(applied_rules)
optimized_prompt += f"\n\n用户问题:{user_query}"
return optimized_prompt
def add_feedback(self, rating: int, problem_type: str = None):
"""添加用户反馈"""
self.feedback_history.append({
"rating": rating,
"problem_type": problem_type
})
# 限制历史长度,只保留最近10条
if len(self.feedback_history) > 10:
self.feedback_history.pop(0)
# 使用示例
if __name__ == "__main__":
# 初始化优化器,基础提示为代码解释
optimizer = DynamicPromptOptimizer(base_prompt="请解释以下Python代码的功能和实现逻辑:")
# 第一次查询:无历史反馈,使用基础提示
user_query = "def add(a,b): return a+b"
prompt = optimizer.generate_prompt(user_query)
print("第一次提示:")
print(prompt)
# 假设用户反馈:评分2分,问题类型"格式混乱"
optimizer.add_feedback(rating=2, problem_type="格式混乱")
# 第二次查询:应用"格式混乱"优化规则
user_query = "def factorial(n): return 1 if n<=1 else n*factorial(n-1)"
prompt = optimizer.generate_prompt(user_query)
print("\n第二次提示:")
print(prompt)
# 假设用户反馈:评分4分(正面),无需添加问题类型
optimizer.add_feedback(rating=4)
# 第三次查询:保留"格式混乱"规则(因最近10条有负面反馈)
user_query = "def fibonacci(n): return [0,1][:n] if n<=2 else fibonacci(n-1)+[fibonacci(n-1)[-1]+fibonacci(n-2)[-1]]"
prompt = optimizer.generate_prompt(user_query)
print("\n第三次提示:")
print(prompt)
代码输出与解读
第一次提示(无优化规则):
请解释以下Python代码的功能和实现逻辑:
用户问题:def add(a,b): return a+b
第二次提示(添加"格式混乱"规则):
请解释以下Python代码的功能和实现逻辑:
额外要求:
请使用清晰的标题和分点列表组织内容
用户问题:def factorial(n): return 1 if n<=1 else n*factorial(n-1)
第三次提示(保留优化规则):
请解释以下Python代码的功能和实现逻辑:
额外要求:
请使用清晰的标题和分点列表组织内容
用户问题:def fibonacci(n): return [0,1][:n] if n<=2 else fibonacci(n-1)+[fibonacci(n-1)[-1]+fibonacci(n-2)[-1]]
优化效果:智能体通过学习用户反馈,自动在提示中加入格式要求,避免重复出现"格式混乱"问题。
多智能体协作:基于角色的任务分配算法
算法原理:合同网协议(Contract Net Protocol)
合同网协议是多智能体协作的经典算法,模拟"招投标"过程分配任务:
- 任务宣告:管理者智能体广播任务需求(如"需要生成月度销售报告")
- 投标:有能力的智能体提交投标方案(如"数据分析智能体"投标"负责数据计算",“文档生成智能体"投标"负责报告撰写”)
- 评标与授标:管理者选择最优投标者并分配任务
- 任务执行与汇报:中标智能体执行任务并汇报结果
Python实现:多智能体任务分配示例
from typing import List, Dict, Tuple
class Agent:
def __init__(self, name: str, capabilities: List[str]):
self.name = name
self.capabilities = capabilities # 智能体能力列表(如["数据分析", "文档生成"])
self.bid_history = [] # 投标历史
def bid(self, task: str) -> Tuple[bool, float]:
"""对任务投标:返回(是否投标,投标分数0-100)"""
# 简化:根据能力匹配度计算投标分数
if task in self.capabilities:
score = 90 # 完全匹配能力,高分
elif any(cap in task for cap in self.capabilities):
score = 60 # 部分匹配,中分
else:
return (False, 0) # 不匹配,不投标
self.bid_history.append({"task": task, "score": score})
return (True, score)
class ManagerAgent:
def __init__(self, agents: List[Agent]):
self.agents = agents # 管理的智能体列表
def assign_task(self, task: str) -> Dict:
"""分配任务给最优智能体"""
# 收集投标
bids = []
for agent in self.agents:
will_bid, score = agent.bid(task)
if will_bid:
bids.append({"agent": agent.name, "score": score})
if not bids:
return {"status": "failed", "message": "无智能体可执行任务"}
# 选择分数最高的智能体
best_bid = max(bids, key=lambda x: x["score"])
return {
"status": "success",
"task": task,
"assigned_to": best_bid["agent"],
"bid_score": best_bid["score"],
"bidders": [b["agent"] for b in bids]
}
# 使用示例
if __name__ == "__main__":
# 创建智能体
data_agent = Agent("数据分析智能体", ["数据分析", "统计计算", "图表生成"])
doc_agent = Agent("文档生成智能体", ["文档撰写", "格式排版", "报告生成"])
code_agent = Agent("代码生成智能体", ["代码编写", "调试", "优化"])
agents = [data_agent, doc_agent, code_agent]
# 创建管理者智能体
manager = ManagerAgent(agents)
# 分配任务
tasks = [
"生成月度销售数据统计图表",
"撰写产品需求文档",
"开发用户注册API接口"
]
for task in tasks:
result = manager.assign_task(task)
print(f"任务:{task}")
print(f"分配结果:{result}\n")
代码输出与解读
任务:生成月度销售数据统计图表
分配结果:{'status': 'success', 'task': '生成月度销售数据统计图表', 'assigned_to': '数据分析智能体', 'bid_score': 90, 'bidders': ['数据分析智能体']}
任务:撰写产品需求文档
分配结果:{'status': 'success', 'task': '撰写产品需求文档', 'assigned_to': '文档生成智能体', 'bid_score': 90, 'bidders': ['文档生成智能体']}
任务:开发用户注册API接口
分配结果:{'status': 'success', 'task': '开发用户注册API接口', 'assigned_to': '代码生成智能体', 'bid_score': 90, 'bidders': ['代码生成智能体']}
协作逻辑:管理者通过任务宣告-投标-评标的流程,将任务分配给能力最匹配的智能体,实现高效协作。
项目实战:代码实际案例和详细解释说明
项目背景:智能研发项目管家多智能体系统
项目目标:构建一个帮助研发团队管理项目的多智能体系统,能自动完成"需求分析→任务拆解→进度跟踪→风险预警"全流程。
系统架构:
- 总协调智能体:接收项目目标,分配任务给专业智能体
- 需求分析智能体:将用户需求转化为技术规格
- 任务拆解智能体:将项目分解为可执行的开发任务
- 进度跟踪智能体:监控任务完成情况,生成进度报告
- 风险预警智能体:识别延期风险,提出解决方案
开发环境搭建
技术栈选择
- 核心框架:LangChain(智能体开发)+ FastAPI(API服务)
- LLM模型:GPT-4(核心推理)+ 开源模型(如Llama 3,本地部署工具调用)
- 数据库:PostgreSQL(存储任务数据、智能体记忆)
- 前端:Streamlit(简单的交互界面)
环境配置步骤
- 安装依赖包
pip install langchain openai fastapi uvicorn streamlit psycopg2-binary python-dotenv
- 创建.env文件配置API密钥
OPENAI_API_KEY=your_openai_key
DATABASE_URL=postgresql://user:password@localhost:5432/project_agent
源代码详细实现和代码解读
1. 定义智能体基类
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from dotenv import load_dotenv
import os
load_dotenv()
llm = OpenAI(temperature=0.7) # 中等随机性,平衡创造性和准确性
class BaseAgent:
def __init__(self, name: str, role: str, system_prompt: str):
self.name = name
self.role = role # 智能体角色描述
self.system_prompt = system_prompt # 系统提示(定义行为准则)
self.memory = [] # 短期记忆
def add_memory(self, info: str):
"""添加记忆"""
self.memory.append(info)
# 保留最近5条记忆
if len(self.memory) > 5:
self.memory.pop(0)
def get_context(self) -> str:
"""获取上下文(记忆+角色信息)"""
context = f"角色:{self.role}\n"
if self.memory:
context += "历史信息:\n" + "\n".join([f"- {m}" for m in self.memory]) + "\n"
return context
def run(self, task: str) -> str:
"""执行任务(需子类实现)"""
raise NotImplementedError("子类必须实现run方法")
2. 实现专业智能体
class RequirementAnalysisAgent(BaseAgent):
"""需求分析智能体:将用户需求转化为技术规格"""
def __init__(self):
super().__init__(
name="需求分析智能体",
role="资深产品经理,擅长将模糊的用户需求转化为清晰的技术规格说明书",
system_prompt="你是需求分析专家,需要:1. 识别用户需求的核心目标;2. 拆解功能点;3. 定义验收标准;4. 指出潜在技术挑战"
)
self.prompt_template = PromptTemplate(
input_variables=["context", "task"],
template="{context}\n系统提示:{system_prompt}\n请处理以下需求:{task}\n输出格式:核心目标+功能点列表+验收标准+技术挑战"
)
self.chain = LLMChain(llm=llm, prompt=self.prompt_template)
def run(self, task: str) -> str:
context = self.get_context()
result = self.chain.run(context=context, system_prompt=self.system_prompt, task=task)
self.add_memory(f"已分析需求:{task},结果摘要:{result[:50]}...") # 保存摘要到记忆
return result
class TaskSplittingAgent(BaseAgent):
"""任务拆解智能体:将项目分解为开发任务"""
def __init__(self):
super().__init__(
name="任务拆解智能体",
role="项目经理,擅长将技术规格分解为可执行的开发任务,估算工时和依赖关系",
system_prompt="你是任务拆解专家,需要:1. 按模块拆分任务;2. 估算每个任务的工时(人天);3. 识别任务间的依赖关系;4. 分配优先级"
)
self.prompt_template = PromptTemplate(
input_variables=["context", "task"],
template="{context}\n系统提示:{system_prompt}\n请根据技术规格拆解任务:{task}\n输出格式:模块名+任务列表(任务名,工时,优先级,依赖任务)"
)
self.chain = LLMChain(llm=llm, prompt=self.prompt_template)
def run(self, task: str) -> str:
context = self.get_context()
result = self.chain.run(context=context, system_prompt=self.system_prompt, task=task)
self.add_memory(f"已拆解任务:{task},生成{result.count('任务名')}个任务")
return result
3. 实现总协调智能体
class ProjectManagerAgent(BaseAgent):
"""项目总协调智能体:管理项目流程,分配任务给专业智能体"""
def __init__(self, agents: dict):
super().__init__(
name="项目总协调智能体",
role="研发总监,负责协调各专业智能体完成项目目标",
system_prompt="你是项目总协调专家,需要:1. 理解项目目标;2. 决定调用哪个智能体;3. 整合各智能体输出;4. 推进项目到下一阶段"
)
self.agents = agents # 管理的智能体字典:{agent_name: agent_instance}
self.project_stage = "init" # 项目阶段:init→requirement→task_split→progress→risk
self.stage_actions = { # 阶段→对应智能体和任务描述
"init": ("需求分析智能体", "请分析以下项目需求:{goal}"),
"requirement": ("任务拆解智能体", "请根据以下技术规格拆解开发任务:{output}")
# 可扩展其他阶段...
}
def run(self, goal: str) -> dict:
"""执行项目管理流程"""
project_result = {"goal": goal, "stages": {}}
# 按阶段推进项目
while self.project_stage in self.stage_actions:
agent_name, task_template = self.stage_actions[self.project_stage]
agent = self.agents.get(agent_name)
if not agent:
project_result["status"] = "failed"
project_result["error"] = f"智能体{agent_name}不存在"
break
# 生成任务(替换模板中的变量)
if self.project_stage == "init":
task = task_template.format(goal=goal)
else:
# 用上一阶段的输出作为当前任务输入
prev_stage = list(project_result["stages"].keys())[-1]
task = task_template.format(output=project_result["stages"][prev_stage]["output"])
# 执行任务
print(f"项目阶段:{self.project_stage},调用智能体:{agent_name}")
output = agent.run(task)
project_result["stages"][self.project_stage] = {
"agent": agent_name,
"task": task,
"output": output
}
# 进入下一阶段(简化:按固定顺序推进)
if self.project_stage == "init":
self.project_stage = "requirement"
elif self.project_stage == "requirement":
self.project_stage = "progress" # 下一阶段:进度跟踪
# 实际项目中可根据需要扩展更多阶段...
project_result["status"] = "success" if "error" not in project_result else "failed"
return project_result
4. 系统整合与运行
# 创建专业智能体
requirement_agent = RequirementAnalysisAgent()
task_agent = TaskSplittingAgent()
# 此处可添加进度跟踪、风险预警智能体...
# 组织智能体字典
agents = {
"需求分析智能体": requirement_agent,
"任务拆解智能体": task_agent
}
# 创建总协调智能体
project_manager
更多推荐
所有评论(0)