【单智能体】AI顾问智能体-案例讲解（附完整源码）

目录

1. 案例简介

2. 案例目标

3. 技术栈与核心依赖

4. 项目配置

4.1 环境变量配置

4.2 项目配置常量

5. 项目结构

6. 核心代码实现

6.1 工具安全包装器

6.2 市场分析工具

6.3 战略建议工具

6.4 Perplexity 搜索工具

6.5 智能体定义

6.6 运行器配置

7. 运行与测试

7.1 安装步骤

7.2 启动应用

7.3 访问应用

7.4 示例咨询主题

8. 实现思路与扩展建议

8.1 核心设计思想

8.2 可能的优化方向

8.3 扩展功能建议

完整源码

requirements.txt

__init__.py

agent.py

ai_consultant_agent.py

1. 案例简介

本案例展示了如何使用 Google ADK（Agent Development Kit）构建一个强大的业务顾问智能体。该智能体能够提供全面的市场分析、战略规划和可操作的业务建议，并支持实时网络研究功能。

2. 案例目标

• 学习如何使用 Google ADK 构建智能体应用
• 掌握智能体工具（Tools）的设计和实现方法
• 了解如何集成外部 API（如 Perplexity AI）扩展智能体能力
• 实现市场分析和战略建议的自动化生成
• 构建具有交互式界面的 AI 顾问应用

3. 技术栈与核心依赖

4. 项目配置

4.1 环境变量配置

需要配置以下环境变量：

GOOGLE_API_KEY=你的-google-api-密钥
PERPLEXITY_API_KEY=你的-perplexity-api-密钥

4.2 项目配置常量

在 ai_consultant_agent.py 中定义了以下配置：

• MODEL_ID = "gemini-2.5-flash" – 使用的大语言模型
• APP_NAME = "ai_consultant_agent" – 应用名称
• USER_ID = "consultant-user" – 用户标识
• SESSION_ID = "consultant-session" – 会话标识

5. 项目结构

ai_consultant_agent-zh/
├── README.md # 项目说明文档
├── requirements.txt # 依赖包列表
├── __init__.py # 包初始化文件
├── agent.py # 智能体导出模块
└── ai_consultant_agent.py # 核心实现代码

6. 核心代码实现

6.1 工具安全包装器

项目实现了一个安全包装器，确保工具函数返回的结果可以被正确序列化：

• sanitize_bytes_for_json() – 递归将字节对象转换为字符串
• safe_tool_wrapper() – 包装工具函数，处理异常并清理输出

6.2 市场分析工具

analyze_market_data() 函数负责分析市场数据并生成洞察：

• 接收研究查询和可选的行业上下文
• 根据关键词匹配生成相应的市场洞察
• 返回结构化的分析结果

6.3 战略建议工具

generate_strategic_recommendations() 函数基于分析结果生成战略建议：

• 解析市场分析数据
• 根据洞察生成相应的战略建议
• 包含优先级、理由、时间线和行动项

6.4 Perplexity 搜索工具

perplexity_search() 函数集成 Perplexity AI 进行实时网络搜索：

• 调用 Perplexity API 进行网络搜索
• 使用 sonar 模型获取最新信息
• 返回搜索结果、引用和使用信息

6.5 智能体定义

使用 LlmAgent 类创建智能体实例：

• 配置模型、名称和描述
• 设置详细的指令（INSTRUCTIONS）
• 注册可用工具列表
• 配置输出键

6.6 运行器配置

设置会话服务和运行器：

• InMemorySessionService – 内存会话服务
• Runner – 智能体运行器

7. 运行与测试

7.1 安装步骤

7.2 启动应用

adk web

7.3 访问应用

打开浏览器访问：http://localhost:8000

7.4 示例咨询主题

• "我想为小型企业推出一个 SaaS 创业公司"
• "我应该将零售业务扩展到电子商务吗？"
• "医疗科技领域有哪些市场机会？"
• "我应该如何定位我的新金融科技产品？"
• "进入可再生能源市场有哪些风险？"

8. 实现思路与扩展建议

8.1 核心设计思想

• 工具化设计：将功能分解为独立的工具，便于扩展和维护
• 安全优先：使用包装器处理异常和序列化问题
• 实时数据：集成网络搜索获取最新市场信息
• 结构化输出：使用数据类和字典组织输出结果

8.2 可能的优化方向

• 添加更多专业领域的分析工具（如财务分析、技术评估等）
• 实现历史会话持久化存储（使用数据库替代内存存储）
• 添加更多的评估指标和反馈机制
• 支持多语言咨询
• 集成更多的数据源（如金融数据 API、行业报告等）

8.3 扩展功能建议

• 添加 PDF/Word 报告生成功能
• 实现多智能体协作（市场研究 + 财务分析 + 战略规划）
• 添加用户偏好学习功能
• 集成图表生成功能，可视化市场数据
• 添加竞品分析专项工具

完整源码

requirements.txt

google-adk>=1.5.0
google-genai>=0.3.0
python-dotenv>=1.0.0
pydantic>=2.0.0

__init__.py

from .ai_consultant_agent import root_agent, session_service, runner, APP_NAME
from . import agent

__all__ = ['root_agent', 'session_service', 'runner', 'APP_NAME', 'agent']

agent.py

from .ai_consultant_agent import root_agent

# 用于 ADK CLI 发现的导出
__all__ = ['root_agent']

ai_consultant_agent.py

import logging
from typing import Dict, Any, List, Union
from dataclasses import dataclass
import base64
import requests
import os

# Google ADK 导入
from google.adk.agents import LlmAgent
from google.adk.tools import google_search
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner

# 定义智能体配置常量
MODEL_ID = "gemini-2.5-flash"
APP_NAME = "ai_consultant_agent"
USER_ID = "consultant-user"
SESSION_ID = "consultant-session"

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def sanitize_bytes_for_json(obj: Any) -> Any:
"""
递归地将字节对象转换为字符串，以确保 JSON 可序列化。

参数:
obj: 可能包含字节的任何对象

返回:
所有字节都转换为字符串的对象
"""
if isinstance(obj, bytes):
try:
# 首先尝试解码为 UTF-8 文本
return obj.decode('utf-8')
except UnicodeDecodeError:
# 如果不是有效的 UTF-8，编码为 base64 字符串
return base64.b64encode(obj).decode('ascii')
elif isinstance(obj, dict):
return {key: sanitize_bytes_for_json(value) for key, value in obj.items()}
elif isinstance(obj, list):
return [sanitize_bytes_for_json(item) for item in obj]
elif isinstance(obj, tuple):
return tuple(sanitize_bytes_for_json(item) for item in obj)
else:
return obj

def safe_tool_wrapper(tool_func):
"""
包装器，确保工具函数永远不会返回字节对象。

参数:
tool_func: 原始工具函数

返回:
清理输出的包装函数
"""
def wrapped_tool(*args, **kwargs):
try:
result = tool_func(*args, **kwargs)
return sanitize_bytes_for_json(result)
except Exception as e:
logger.error(f"工具 {tool_func.__name__} 出错: {e}")
return {
"error": f"工具执行失败: {str(e)}",
"tool": tool_func.__name__,
"status": "error"
}

# 保留函数元数据
wrapped_tool.__name__ = tool_func.__name__
wrapped_tool.__doc__ = tool_func.__doc__
return wrapped_tool

@dataclass
class MarketInsight:
"""市场研究洞察结构"""
category: str
finding: str
confidence: float
source: str

def analyze_market_data(research_query: str, industry: str = "") -> Dict[str, Any]:
"""
分析市场数据并生成洞察

参数:
research_query: 要分析的业务查询
industry: 可选的行业上下文

返回:
市场分析洞察和建议
"""
# 模拟市场分析 – 在实际实现中，这将处理实际的搜索结果
insights = []

if "startup" in research_query.lower() or "launch" in research_query.lower():
insights.extend([
MarketInsight("市场机会", "市场增长，竞争适中", 0.8, "市场研究"),
MarketInsight("风险评估", "适用标准创业风险 – 资金、竞争", 0.7, "分析"),
MarketInsight("建议", "在全面推出前进行 MVP 测试", 0.9, "战略规划")
])

if "saas" in research_query.lower() or "software" in research_query.lower():
insights.extend([
MarketInsight("技术趋势", "云端解决方案采用率不断增长", 0.9, "技术分析"),
MarketInsight("客户行为", "企业偏好订阅模式", 0.8, "市场研究")
])

if industry:
insights.append(
MarketInsight("行业特定", f"{industry} 领域显示出增长潜力", 0.7, "行业报告")
)

return {
"query": research_query,
"industry": industry,
"insights": [
{
"category": insight.category,
"finding": insight.finding,
"confidence": insight.confidence,
"source": insight.source
}
for insight in insights
],
"summary": f"分析完成: {research_query}",
"total_insights": len(insights)
}

def generate_strategic_recommendations(analysis_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
基于分析生成战略业务建议

参数:
analysis_data: 市场分析结果

返回:
战略建议列表
"""
recommendations = []

# 基于洞察生成建议
insights = analysis_data.get("insights", [])

if any("startup" in insight["finding"].lower() for insight in insights):
recommendations.append({
"category": "市场进入策略",
"priority": "高",
"recommendation": "实施分阶段市场进入，进行 MVP 测试",
"rationale": "降低风险并在重大投资前验证市场契合度",
"timeline": "3-6 个月",
"action_items": [
"开发最小可行产品",
"识别目标客户群体",
"进行市场验证测试"
]
})

if any("saas" in insight["finding"].lower() for insight in insights):
recommendations.append({
"category": "技术策略",
"priority": "中",
"recommendation": "专注于云原生架构和订阅模式",
"rationale": "符合市场趋势和客户偏好",
"timeline": "2-4 个月",
"action_items": [
"设计可扩展的云基础设施",
"实施订阅计费系统",
"规划多租户架构"
]
})

# 始终包含风险管理
recommendations.append({
"category": "风险管理",
"priority": "高",
"recommendation": "建立全面的风险监控框架",
"rationale": "主动风险管理对业务成功至关重要",
"timeline": "1-2 个月",
"action_items": [
"识别关键业务风险",
"制定缓解策略",
"实施监控系统"
]
})

return recommendations

def perplexity_search(query: str, system_prompt: str = "精确简洁。专注于业务洞察和市场数据。") -> Dict[str, Any]:
"""使用 Perplexity AI 搜索网络以获取实时信息和洞察。"""
try:
api_key = os.getenv("PERPLEXITY_API_KEY")
if not api_key:
return {"error": "未找到 Perplexity API 密钥。请设置 PERPLEXITY_API_KEY 环境变量。", "query": query, "status": "error"}

response = requests.post("https://api.perplexity.ai/chat/completions",
json={"model": "sonar", "messages": [{"role": "system", "content": system_prompt}, {"role": "user", "content": query}]},
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, timeout=30)
response.raise_for_status()
result = response.json()

if "choices" in result and result["choices"]:
return {"query": query, "content": result["choices"][0]["message"]["content"], "citations": result.get("citations", []),
"search_results": result.get("search_results", []), "status": "success", "source": "Perplexity AI",
"model": result.get("model", "sonar"), "usage": result.get("usage", {}), "response_id": result.get("id", ""), "created": result.get("created", 0)}
return {"error": "未找到响应内容", "query": query, "status": "error", "raw_response": result}
except Exception as e:
return {"error": f"错误: {str(e)}", "query": query, "status": "error"}

# 定义带有安全包装器的顾问工具
consultant_tools = [
safe_tool_wrapper(analyze_market_data),
safe_tool_wrapper(generate_strategic_recommendations),
safe_tool_wrapper(perplexity_search)
]

INSTRUCTIONS = """你是一位专注于市场分析和战略规划的高级 AI 业务顾问。

你的专业领域包括:
– 业务战略制定和建议
– 风险评估和缓解规划
– 实施规划和时间安排
– 使用你的知识和可用工具进行市场分析
– 使用 Perplexity AI 搜索功能进行实时网络研究

在为客户提供咨询服务时:
1. 使用 Perplexity 搜索从网络上收集当前的市场数据、竞争对手信息和行业趋势
2. 使用市场分析工具处理业务查询并生成洞察
3. 使用战略建议工具创建可操作的业务建议
4. 提供清晰、具体的建议，包含实施时间表
5. 专注于能够带来可衡量业务成果的实用解决方案

**核心职责:**
– 使用 Perplexity AI 进行实时网络研究，获取当前市场数据和趋势
– 使用搜索结果和你的知识分析竞争格局和市场机会
– 基于最新信息提供明确的战略指导和行动项目
– 根据当前市场状况评估风险并提出缓解策略
– 制定具有现实时间表的实施路线图
– 结合网络研究和分析工具生成全面的业务洞察

**关键规则:**
– 在相关时始终使用 Perplexity 搜索查找当前的市场数据、趋势和竞争对手信息
– 基于合理的业务原则、当前市场洞察和实时网络数据提供建议
– 提供具体、可操作的建议，而非通用指导
– 在建议中包含时间表和成功指标
– 按业务影响和可行性对建议进行优先级排序
– 使用 Perplexity 搜索验证假设并收集带引用的支持证据
– 结合搜索结果和分析工具提供全面的咨询服务

**搜索策略:**
– 使用 Perplexity 搜索进行竞争对手分析、市场规模、行业趋势和监管变化
– 查找相关领域的最新新闻、融资轮次和市场发展
– 在提出建议前用当前网络数据验证市场假设
– 研究类似企业的最佳实践和案例研究
– 在引用搜索结果时始终包含引用和来源

始终保持专业、分析性的方法，同时注重结果导向。
使用所有可用工具（包括 Perplexity 搜索）提供基于当前网络数据和引用的全面、深入研究型咨询服务。"""

# 定义智能体实例
root_agent = LlmAgent(
model=MODEL_ID,
name=APP_NAME,
description="一位 AI 业务顾问，提供市场研究、战略分析和可操作的建议。",
instruction=INSTRUCTIONS,
tools=consultant_tools,
output_key="consultation_response"
)

# 设置运行器和会话服务
session_service = InMemorySessionService()
runner = Runner(
agent=root_agent,
app_name=APP_NAME,
session_service=session_service
)

if __name__ == "__main__":
print("🤖 基于 Google ADK 的 AI 顾问智能体")
print("=====================================")
print()
print("此智能体提供全面的业务咨询服务，包括:")
print("• 市场研究和分析")
print("• 战略建议")
print("• 实施规划")
print("• 风险评估")
print()
print("使用此智能体:")
print("1. 运行: adk web .")
print("2. 打开 Web 界面")
print("3. 选择 'AI Business Consultant' 智能体")
print("4. 开始你的咨询")
print()
print("示例查询:")
print('• "我想为小型企业推出一个 SaaS 创业公司"')
print('• "我应该将零售业务扩展到电子商务吗？"')
print('• "医疗科技领域有哪些市场机会？"')
print()
print("📊 使用 ADK Web 中的评估标签页保存和评估咨询会话！")
print()
print(f"✅ 智能体 '{APP_NAME}' 初始化成功！")
print(f" 模型: {MODEL_ID}")
print(f" 工具: {len(consultant_tools)} 个可用")
print(f" 会话服务: {type(session_service).__name__}")
print(f" 运行器: {type(runner).__name__}")