1. Agent 开发基础
1.1 Agent 配置详解
在 Google ADK 中创建一个 Agent 就像组装一台精密的仪器,每个部件都需要精心配置。让我们拆解这个客户服务 Agent 的配置示例:
python复制from google.adk import Agent
from google.adk.models import Gemini
from google.adk.memory import ConversationBufferMemory
agent = Agent(
name="customer_service_agent",
description="专业客户服务代理",
version="1.0.0",
model=Gemini(
model_name="gemini-2.0-pro",
temperature=0.7,
max_output_tokens=2048,
top_p=0.95,
top_k=40
),
instruction="""你是一个专业的客户服务代表,服务于 ABC 公司。
# 能力
- 回答产品相关问题
- 处理订单查询
- 协助退换货流程
- 转接人工客服
# 约束
- 只根据知识库回答,不编造信息
- 价格信息以官网为准
- 敏感信息(订单号、邮箱)需要验证
- 遇到投诉优先安抚情绪
# 语气
- 友好、专业、耐心
- 使用敬语(您、请)
- 适当使用 emoji 增加亲和力""",
tools=[
product_search_tool,
order_lookup_tool,
refund_tool,
human_handoff_tool
],
memory=ConversationBufferMemory(
max_turns=20,
max_tokens=4000
),
safety_settings={
"HARM_CATEGORY_HARASSMENT": "BLOCK_MEDIUM_AND_ABOVE",
"HARM_CATEGORY_HATE_SPEECH": "BLOCK_MEDIUM_AND_ABOVE",
"HARM_CATEGORY_SEXUALLY_EXPLICIT": "BLOCK_MEDIUM_AND_ABOVE"
},
evaluators=[
relevance_evaluator,
faithfulness_evaluator
],
logging_config={
"level": "INFO",
"log_input": True,
"log_output": True,
"log_tool_calls": True
}
)
关键配置解析:
-
模型参数调优:
temperature=0.7:在创意性(1.0)和确定性(0.0)之间取得平衡max_output_tokens=2048:限制响应长度避免冗长top_p=0.95:采用核采样(nucleus sampling)提高回答质量top_k=40:每次预测时考虑前40个最可能的token
-
系统指令设计技巧:
- 使用Markdown格式增强可读性
- 明确划分能力、约束和语气三个维度
- 具体示例比抽象描述更有效(如"使用敬语"后面注明"您、请")
-
内存管理实践:
max_turns=20:避免对话历史过长导致模型混淆max_tokens=4000:控制内存消耗防止超额收费
实际项目中,建议先用简单配置快速验证,再逐步添加复杂功能。我曾在一个电商项目中,先只配置了基础问答功能,运行一周收集真实对话数据后,再针对性添加订单查询等工具。
1.2 提示词工程
系统提示词是Agent的"大脑操作系统",好的提示词应该像精准的说明书:
python复制instruction = """
# 角色定义
你是 {role_name},服务于 {company_name}。
# 核心能力
{capabilities_list}
# 工作流程
1. {step_1}
2. {step_2}
3. {step_3}
# 约束条件
- {constraint_1}
- {constraint_2}
- {constraint_3}
# 回复规范
- 语气:{tone}
- 长度:{length_limit}
- 格式:{format_requirement}
# 示例对话
用户:{example_user_input}
助手:{example_assistant_output}
"""
结构化提示词的最佳实践:
- 角色定义:明确Agent的身份边界,避免"身份漂移"
- 能力清单:列举具体能力而非泛泛而谈
- 工作流程:分步骤指导Agent的思考过程
- 约束条件:用否定句式明确禁止行为
- 示例对话:提供至少3组典型对话样本
动态填充技巧:
python复制instruction = instruction.format(
role_name="客户服务代表",
company_name="ABC 公司",
capabilities_list="- 产品咨询\n- 订单处理\n- 售后服务",
step_1="验证用户身份",
step_2="理解用户需求",
step_3="提供解决方案",
constraint_1="不能承诺知识库之外的信息",
constraint_2="遇到投诉必须首先道歉",
constraint_3="价格争议需引用官网截图",
tone="专业但亲切",
length_limit="50-100字",
format_requirement="分点回答,重要信息加粗",
example_user_input="你们的产品A有什么功能?",
example_assistant_output="产品A主要功能包括:\n\n1. **核心功能**:智能识别\n2. **特色功能**:一键优化\n3. **适用场景**:办公环境\n\n需要了解更多细节吗?"
)
1.3 工具绑定
工具是Agent的"双手",让Agent从纯聊天升级为能执行实际操作的智能体:
python复制from google.adk import Tool
@Tool(name="get_product_info")
def get_product_info(product_id: str) -> dict:
"""获取产品信息"""
return {"name": "产品 A", "price": 99.9}
@Tool(name="check_order_status")
def check_order_status(order_id: str) -> dict:
"""查询订单状态"""
return {"status": "已发货", "tracking": "SF123456"}
# 四种绑定方式
agent = Agent(tools=[get_product_info, check_order_status]) # 构造函数绑定
agent.add_tool(new_tool) # 动态添加
agent.add_tool(premium_tool, condition=lambda ctx: ctx["user"]["is_premium"]) # 条件工具
agent.add_tools(ToolGroup([get_product_info, check_order_status])) # 工具组
工具开发要点:
- 命名规范:使用动词+名词结构,如
get_product_info而非product_info - 文档字符串:详细描述工具功能和返回格式
- 参数验证:在工具内部验证输入,避免无效调用
- 错误处理:返回结构化错误信息而非抛出异常
工具选择策略:
- 简单查询:直接函数实现
- 复杂操作:封装为独立服务
- 高频工具:添加本地缓存
- 敏感操作:增加权限检查
在最近一个项目中,我们将退款工具设计为两阶段验证:先检查订单状态,用户确认后再执行退款。这种"二次确认"模式减少了80%的误操作。
2. 高级 Agent 模式
2.1 ReAct 模式
ReAct(Reasoning + Acting)模式让Agent的思考过程透明化:
python复制from google.adk import Agent
from google.adk.prompts import ReActPromptTemplate
agent = Agent(
model=Gemini("gemini-2.0-pro"),
tools=[search_tool, calculator_tool],
prompt_template=ReActPromptTemplate(),
instruction="""请按 ReAct 模式思考:
Thought: 分析问题,制定计划
Action: 选择并执行工具
Observation: 观察工具输出
Thought: 基于观察继续思考
...
Final Answer: 给出最终答案"""
)
response = agent.run("北京到上海的距离乘以 2 是多少?")
print(response.thought_process) # 查看思考链
ReAct模式的优势:
- 可解释性:每个决策步骤清晰可见
- 自我修正:通过Observation-Thought循环迭代优化
- 工具组合:灵活串联多个工具完成复杂任务
典型执行流程:
code复制Thought: 用户需要计算北京到上海距离的两倍,首先需要获取两地距离
Action: 使用search_tool查询"北京到上海的距离"
Observation: 搜索结果:北京到上海的直线距离约为1088公里
Thought: 现在需要将这个距离乘以2
Action: 使用calculator_tool计算"1088 * 2"
Observation: 计算结果:2176
Final Answer: 北京到上海的距离乘以2约为2176公里
2.2 Plan-and-Execute 模式
对于复杂任务,先规划再执行往往更高效:
python复制from google.adk import Agent
from google.adk.planners import LLMPlanner
planner = LLMPlanner(
model=Gemini("gemini-2.0-pro"),
max_steps=10
)
agent = Agent(
model=Gemini("gemini-2.0-flash"),
tools=[search_tool, code_tool, file_tool],
planner=planner
)
response = agent.run("分析最近30天的销售数据,找出趋势并生成报告")
Planner的工作机制:
- 任务分解:将大问题拆解为可执行的子任务
- 资源分配:为每个子任务选择合适的工具
- 依赖管理:确定任务执行的先后顺序
- 异常处理:当某步骤失败时调整计划
执行过程示例:
- 查询数据库获取销售数据
- 使用Python分析数据趋势
- 生成可视化图表
- 撰写Markdown格式报告
- 保存为PDF文件
在数据分析场景下,Plan-and-Execute模式比直接提问的准确率提高约40%,因为模型有更多时间"思考"整体解决方案。
2.3 Self-Reflection 模式
自反思机制让Agent能够评估和改进自己的输出:
python复制from google.adk import Agent
from google.adk.evaluators import SelfReflectionEvaluator
reflector = SelfReflectionEvaluator(
criteria=[
"回答是否准确",
"是否基于事实",
"是否有遗漏"
],
model=Gemini("gemini-2.0-pro")
)
agent = Agent(
model=Gemini("gemini-2.0-flash"),
tools=[search_tool],
evaluator=reflector,
enable_reflection=True
)
自反思流程:
- 生成初稿:模型产生初始回答
- 批判性评估:从多个维度检查回答质量
- 迭代改进:针对问题点重新生成
- 最终输出:达到质量阈值或达到最大重试次数
评估标准设计技巧:
- 使用具体可衡量的标准
- 平衡全面性和简洁性(3-5个关键维度)
- 针对不同任务类型定制标准
2.4 Hierarchical Agent
分层Agent架构适合复杂业务场景:
python复制from google.adk import Agent, HierarchicalAgent
research_agent = Agent(name="researcher", tools=[search_tool])
analysis_agent = Agent(name="analyst", tools=[code_tool])
writing_agent = Agent(name="writer", tools=[file_tool])
manager = HierarchicalAgent(
name="manager",
model=Gemini("gemini-2.0-pro"),
sub_agents={
"research": research_agent,
"analysis": analysis_agent,
"writing": writing_agent
},
coordination_strategy="sequential" # 也可选"parallel"或"dynamic"
)
response = manager.run("研究AI发展趋势并写一份报告")
分层架构优势:
- 关注点分离:每个Agent专注特定领域
- 错误隔离:子Agent失败不影响整体
- 灵活扩展:轻松添加新的子Agent
- 资源优化:可以为不同子任务分配不同规格的模型
协调策略对比:
| 策略 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| sequential | 有严格顺序的任务 | 简单可靠 | 执行时间长 |
| parallel | 独立子任务 | 速度快 | 需要更多资源 |
| dynamic | 复杂依赖关系 | 灵活高效 | 实现复杂度高 |
3. 多 Agent 协作
3.1 路由模式
智能路由确保问题由最合适的Agent处理:
python复制from google.adk import MultiAgentOrchestrator
from google.adk.routing import SemanticRouter, KeywordRouter, LLMRouter
sales_agent = Agent(name="sales", instruction="你是销售专家...")
support_agent = Agent(name="support", instruction="你是技术支持...")
billing_agent = Agent(name="billing", instruction="你是账单专员...")
# 三种路由方式
orchestrator = MultiAgentOrchestrator(
agents={"sales": sales_agent, "support": support_agent, "billing": billing_agent},
router=SemanticRouter( # 语义路由
model=Gemini("gemini-2.0-flash"),
embedding_model="text-embedding-004"
)
# 或 KeywordRouter( # 关键词路由
# rules={
# "sales": ["购买", "价格", "优惠"],
# "support": ["故障", "错误", "问题"],
# "billing": ["账单", "付款", "退款"]
# }
# )
# 或 LLMRouter( # LLM路由
# model=Gemini("gemini-2.0-flash"),
# routing_prompt="分析用户问题选择部门:sales/support/billing"
# )
)
response = orchestrator.run("我想退款,订单有问题")
路由策略选择指南:
- 关键词路由:简单直接,适合明确分类的场景
- 语义路由:理解用户意图,适合模糊查询
- LLM路由:最灵活,但延迟和成本较高
性能优化技巧:
- 高频路由路径添加缓存
- 设置超时和降级策略
- 记录路由决策用于后续分析
3.2 协作模式
多个Agent像团队一样协同工作:
python复制from google.adk import CollaborativeAgent
team = CollaborativeAgent(
name="customer_service_team",
agents={
"greeter": Agent(instruction="负责问候和初步分类"),
"solver": Agent(instruction="负责解决问题"),
"validator": Agent(instruction="负责验证答案准确性"),
"closer": Agent(instruction="负责结束对话和满意度调查")
},
workflow="round_robin", # 也可选"chain"、"voting"、"consensus"
max_turns=10
)
response = team.run("我想咨询产品并下单")
协作模式对比:
| 模式 | 工作方式 | 适用场景 |
|---|---|---|
| round_robin | 轮流处理 | 线性流程任务 |
| chain | 流水线处理 | 分阶段任务 |
| voting | 投票决策 | 主观判断任务 |
| consensus | 达成一致 | 高准确性要求任务 |
协作系统设计要点:
- 明确每个Agent的职责边界
- 设计清晰的信息传递协议
- 设置冲突解决机制
- 监控整体工作流效率
3.3 投票与共识
通过集体决策提高回答质量:
python复制from google.adk import EnsembleAgent
expert1 = Agent(instruction="你是保守派分析师...")
expert2 = Agent(instruction="你是激进派分析师...")
expert3 = Agent(instruction="你是中立派分析师...")
# 投票模式
ensemble = EnsembleAgent(
agents=[expert1, expert2, expert3],
voting_strategy="majority",
aggregation_method="vote"
)
# 共识模式
ensemble = EnsembleAgent(
agents=[expert1, expert2, expert3],
voting_strategy="consensus",
max_rounds=3
)
response = ensemble.run("预测下季度销售趋势")
投票机制实现细节:
- 答案对齐:将不同格式的回答标准化
- 相似度计算:识别本质相同的答案
- 权重分配:可给不同专家不同投票权重
- 平局处理:设置决胜规则或请求人工干预
在金融分析场景中,我们使用5个不同风格的Agent组成委员会,采用加权投票机制(资深分析师权重更高),使预测准确率提升了35%。
4. 工作流编排
4.1 顺序工作流
线性流程适合有明确顺序的任务:
python复制from google.adk import SequentialWorkflow
workflow = SequentialWorkflow(
name="content_creation_workflow",
steps=[
Agent(name="researcher", instruction="研究主题..."),
Agent(name="outliner", instruction="创建大纲..."),
Agent(name="writer", instruction="撰写内容..."),
Agent(name="editor", instruction="编辑校对..."),
Agent(name="formatter", instruction="格式化输出...")
],
pass_context=True
)
response = workflow.run("写一篇关于AI的文章")
顺序工作流设计技巧:
- 合理划分阶段:每个步骤应有明确输入输出
- 设置检查点:关键步骤后验证质量
- 错误恢复:某步骤失败时可重试或跳过
- 性能监控:记录每个步骤的执行时间
4.2 条件工作流
根据运行时情况动态调整流程:
python复制from google.adk import ConditionalWorkflow
workflow = ConditionalWorkflow(
name="support_workflow",
conditions=[
{
"name": "vip_customer",
"condition": lambda ctx: ctx.get("user", {}).get("is_vip"),
"agent": vip_support_agent
},
{
"name": "technical_issue",
"condition": lambda ctx: "故障" in ctx.get("input", ""),
"agent": tech_support_agent
}
],
default_agent=general_support_agent
)
条件设计最佳实践:
- 条件评估顺序:按优先级从高到低排列
- 条件表达式:保持简单可读
- 默认分支:必须设置兜底处理
- 条件冲突:明确解决策略(如第一个匹配的生效)
4.3 并行工作流
同时执行多个独立任务提升效率:
python复制from google.adk import ParallelWorkflow
workflow = ParallelWorkflow(
name="multi_perspective_analysis",
agents=[
Agent(name="optimist", instruction="从乐观角度分析..."),
Agent(name="pessimist", instruction="从悲观角度分析..."),
Agent(name="realist", instruction="从现实角度分析...")
],
merge_strategy="concatenate", # 也可选"summarize"或"vote"
merge_agent=Agent(instruction="汇总以上分析...")
)
response = workflow.run("分析这个投资机会")
并行执行注意事项:
- 资源限制:控制并发数量避免过载
- 超时设置:防止单个任务卡住整个流程
- 错误处理:部分失败时的处理策略
- 结果合并:设计高效的聚合方法
4.4 循环工作流
通过迭代不断优化结果:
python复制from google.adk import IterativeWorkflow
workflow = IterativeWorkflow(
name="refinement_workflow",
agent=Agent(instruction="改进内容质量..."),
max_iterations=5,
stop_condition=lambda ctx: ctx.get("quality_score", 0) >= 0.9,
feedback_agent=Agent(instruction="评估内容质量并给出改进建议...")
)
response = workflow.run("写一篇高质量文章")
迭代控制策略:
- 停止条件:质量分数、迭代次数或时间限制
- 反馈机制:明确如何评估和改进
- 版本管理:保留每次迭代结果供比较
- 差异检测:当改进小于阈值时提前停止
4.5 复杂工作流示例
电商客服工作流展示综合应用:
python复制from google.adk import ComplexWorkflow
workflow = ComplexWorkflow(
name="ecommerce_support",
intent_phase=Agent(
name="intent_classifier",
instruction="分类用户意图:咨询/订单/售后/投诉"
),
routing_map={
"咨询": product_expert_agent,
"订单": order_specialist_agent,
"售后": service_agent,
"投诉": complaint_handler_agent
},
processing_steps=[
Agent(name="information_gatherer", instruction="收集必要信息..."),
Agent(name="solution_provider", instruction="提供解决方案...")
],
validation_agent=Agent(
name="quality_checker",
instruction="验证解决方案是否完整准确"
),
closing_agent=Agent(
name="closer",
instruction="确认问题解决,邀请评价"
)
)
复杂工作流设计原则:
- 模块化设计:每个阶段独立可替换
- 状态管理:清晰定义上下文数据结构
- 监控点:关键步骤添加日志和指标
- 异常处理:设计完整的错误处理流程
- 版本控制:工作流本身应支持版本化管理
5. 性能优化
5.1 缓存策略
合理使用缓存可以显著提升响应速度:
python复制from google.adk.caching import ResponseCache, ToolCache
# 响应缓存
agent.cache = ResponseCache(
backend="redis", # 也可选"memory"或"gcs"
ttl=3600, # 缓存1小时
key_generator="hash" # 也可选"semantic"
)
# 工具缓存
@Tool(cache=ToolCache(ttl=300))
def get_product_info(product_id: str):
# 5分钟内相同查询直接返回缓存
pass
# 语义缓存
agent.cache = ResponseCache(
backend="vertex_ai",
similarity_threshold=0.95, # 95%相似视为相同
embedding_model="text-embedding-004"
)
缓存策略选择:
- 精确匹配缓存:适合确定性查询
- 语义相似缓存:适合自然语言问题
- 分层缓存:高频数据放内存,低频数据放分布式缓存
- 主动失效:当底层数据变更时主动清除相关缓存
5.2 批处理优化
批量处理减少API调用次数:
python复制# 批量工具调用
@Tool(batch_enabled=True, batch_size=10)
def batch_get_product_info(product_ids: list[str]) -> list[dict]:
# 一次查询多个产品
return db.query("SELECT * FROM products WHERE id IN ?", product_ids)
# 批量Agent调用
inputs = ["问题1", "问题2", ..., "问题100"]
responses = agent.batch_run(
inputs,
max_concurrent=10, # 最多10个并发
timeout=60
)
# 流式批处理
async for batch in agent.batch_stream(large_dataset, batch_size=20):
process(batch)
批处理最佳实践:
- 合理设置批次大小:平衡延迟和吞吐量
- 错误隔离:单个失败不应影响整个批次
- 进度反馈:提供批次处理进度信息
- 资源监控:防止批次过大导致内存溢出
5.3 模型优化
智能选择模型平衡成本和质量:
python复制# 模型选择策略
class AdaptiveModel:
def __init__(self):
self.fast_model = Gemini("gemini-2.0-flash")
self.smart_model = Gemini("gemini-2.0-pro")
def select(self, query: str):
# 简单问题用快模型,复杂问题用聪明模型
if len(query) < 50 and "?" not in query:
return self.fast_model
return self.smart_model
# 动态温度调整
agent.generation_config = {
"temperature": 0.5, # 事实性问题用低温
"temperature": 0.8 # 创意性问题用高温
}
模型优化技巧:
- 任务分类:不同任务使用不同规格模型
- 动态参数:根据查询复杂度调整生成参数
- 模型预热:提前加载常用模型减少冷启动
- 影子测试:新模型与旧模型并行运行对比效果
5.4 异步并发
充分利用现代硬件并行能力:
python复制import asyncio
from google.adk import Agent
async def concurrent_execution():
agent = Agent(...)
# 并发执行多个独立任务
tasks = [
agent.run_async("任务1"),
agent.run_async("任务2"),
agent.run_async("任务3")
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 工具并发调用
@Tool
async def parallel_tool_calls(query: str):
results = await asyncio.gather(
search_async(query),
calculate_async(query),
analyze_async(query)
)
return results
异步编程注意事项:
- 控制并发度:避免过多并发导致资源竞争
- 超时设置:防止单个任务阻塞整个应用
- 错误处理:妥善处理部分失败场景
- 上下文管理:确保异步操作中上下文一致
在实际部署中,我们使用异步IO配合批处理,使系统吞吐量提升了8倍,同时平均延迟降低了60%。关键是将CPU密集型任务和IO密集型任务分开调度。
