提示工程架构师进阶: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个核心章节:

  1. 背景介绍:Agentic AI的兴起背景与学习路径
  2. 核心概念与联系:用生活化比喻拆解智能体系统的关键组件
  3. 智能体架构设计原则:从单智能体到多智能体的架构模式
  4. 自主决策算法与动态提示优化:数学原理与代码实现
  5. 项目实战:构建"智能研发项目管家"多智能体系统
  6. 实际应用场景:金融、医疗、制造等领域的落地案例
  7. 工具与资源推荐:提升开发效率的框架与最佳实践
  8. 未来趋势与挑战:技术瓶颈与突破方向

术语表

核心术语定义
术语 通俗解释 专业定义
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)

核心概念与联系

故事引入

“小明的智能家居升级记”

小明家有一套智能家居系统,但体验一直不太满意:

  • 智能音箱只会"一问一答"(“打开客厅灯"→"已打开”),不会主动提醒"离家时忘关灯"
  • 扫地机器人每天固定时间清扫,遇到小明在家工作时也会噪音扰民
  • 空调和窗帘各自为战,不会根据天气和小明的作息自动调整

后来,小明家引入了一个**“智能家居总管家智能体”**,情况彻底改变:

  1. 感知环境:总管家连接所有设备,实时获取天气(外部API)、小明的位置(手机定位)、设备状态(灯、空调开关)
  2. 拆解目标:接到"提升居住舒适度"的总目标后,自动拆解为"节能"、“便捷”、"健康"三个子目标
  3. 多智能体协作:总管家不是单打独斗,而是协调多个专业智能体:
    • 环境监测智能体:分析温度、湿度、PM2.5数据
    • 作息学习智能体:记录小明的起床、工作、睡眠规律
    • 设备控制智能体:具体操作灯光、空调、窗帘
  4. 自主决策:某天早上,环境监测智能体发现"室外温度骤降且PM2.5超标",总管家立即决策:
    • 提前15分钟打开暖气(根据作息智能体的数据,小明通常7点起床)
    • 关闭窗户并开启空气净化器(无需小明指令)
    • 发送提醒:“今天天气寒冷且空气质量差,建议佩戴口罩”
  5. 动态调整:如果小明临时加班晚归,作息学习智能体会更新数据,总管家自动推迟扫地机器人的工作时间

这个故事里的"总管家+专业智能体团队",就是典型的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的"独立思考能力"

生活比喻:自主决策机制就像"小学生做数学题的思考过程":

  1. 理解题目:"小明有5个苹果,给了小红2个,还剩几个?“→明确目标是"计算剩余苹果数”
  2. 选择方法:回忆学过的减法→"5-2=?"
  3. 执行计算:得出结果3
  4. 检查答案:"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)的架构模式
  1. 主从式架构(如智能家居总管家)
+------------+      +------------+      +------------+  
|  主智能体   | ←→ |  子智能体A  | ←→ |  工具/API  |  
+------------+      +------------+      +------------+  
       ←→               ←→  
+------------+      +------------+      +------------+  
|  子智能体B  | ←→ |  子智能体C  | ←→ |  外部系统  |  
+------------+      +------------+      +------------+  
  1. 平等式架构(如多专家协作系统)
+------------+      +------------+      +------------+  
|  智能体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+γmax⁡a′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+γamaxQ(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个步骤:

  1. 收集反馈:用户对智能体输出的评价(如"有用/无用"、“清晰/模糊”)
  2. 分析原因:判断输出问题是否由提示导致(如"未指定格式"导致输出混乱)
  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)

合同网协议是多智能体协作的经典算法,模拟"招投标"过程分配任务:

  1. 任务宣告:管理者智能体广播任务需求(如"需要生成月度销售报告")
  2. 投标:有能力的智能体提交投标方案(如"数据分析智能体"投标"负责数据计算",“文档生成智能体"投标"负责报告撰写”)
  3. 评标与授标:管理者选择最优投标者并分配任务
  4. 任务执行与汇报:中标智能体执行任务并汇报结果
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(简单的交互界面)
环境配置步骤
  1. 安装依赖包
pip install langchain openai fastapi uvicorn streamlit psycopg2-binary python-dotenv  
  1. 创建.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
Logo

更多推荐