iomgaa
1b652502d5
- 添加详细的SubAgent使用指南(README.md) - 创建完整的Pydantic模型示例(example_models.py) - 实现基础使用示例，展示核心功能(basic_example.py) - 构建复杂文本分析应用示例(text_analysis_example.py) - 提供数字提取实验运行器作为参考示例 - 包含多Agent协作、批量处理、性能监控等高级功能 - 支持交互式演示和完整的错误处理机制
403 lines
12 KiB
Python
403 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
SubAgent基础使用示例
|
||
|
||
展示SubAgent系统的基本功能:
|
||
1. 创建简单的对话Agent
|
||
2. 使用动态prompt模板
|
||
3. 结构化JSON输出
|
||
4. 错误处理和调试
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
from typing import Optional
|
||
|
||
# 添加项目根路径到Python路径
|
||
project_root = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
|
||
sys.path.append(project_root)
|
||
|
||
from src.agent_system import SubAgent
|
||
from example_models import BasicResponse, SentimentAnalysis
|
||
|
||
|
||
def create_simple_chat_agent() -> SubAgent:
|
||
"""创建简单的聊天Agent"""
|
||
print("🤖 创建简单聊天Agent...")
|
||
|
||
try:
|
||
agent = SubAgent(
|
||
provider="aliyun",
|
||
model_name="qwen-turbo",
|
||
name="simple_chat",
|
||
description="简单的聊天助手",
|
||
instructions=[
|
||
"你是一个友好的AI助手",
|
||
"请用简洁明了的语言回答问题",
|
||
"保持积极正面的态度"
|
||
]
|
||
)
|
||
|
||
print("✅ 简单聊天Agent创建成功")
|
||
return agent
|
||
|
||
except Exception as e:
|
||
print(f"❌ Agent创建失败: {e}")
|
||
raise
|
||
|
||
|
||
def create_structured_output_agent() -> SubAgent:
|
||
"""创建支持结构化输出的Agent"""
|
||
print("\n🔧 创建结构化输出Agent...")
|
||
|
||
try:
|
||
agent = SubAgent(
|
||
provider="aliyun",
|
||
model_name="qwen-max",
|
||
name="structured_responder",
|
||
description="提供结构化响应的智能助手",
|
||
instructions=[
|
||
"你是一个专业的响应助手",
|
||
"始终提供结构化的JSON格式输出",
|
||
"确保响应准确和有用"
|
||
],
|
||
response_model=BasicResponse
|
||
)
|
||
|
||
print("✅ 结构化输出Agent创建成功")
|
||
return agent
|
||
|
||
except Exception as e:
|
||
print(f"❌ 结构化输出Agent创建失败: {e}")
|
||
raise
|
||
|
||
|
||
def create_sentiment_agent() -> SubAgent:
|
||
"""创建情感分析Agent"""
|
||
print("\n💭 创建情感分析Agent...")
|
||
|
||
instructions = [
|
||
"你是专业的文本情感分析专家",
|
||
"准确识别文本的情感倾向：positive（积极）、negative（消极）、neutral（中性）",
|
||
"提供0-1范围的置信度评分",
|
||
"给出详细的分析说明和关键词"
|
||
]
|
||
|
||
prompt_template = """
|
||
请对以下文本进行情感分析：
|
||
|
||
文本内容："{text}"
|
||
|
||
分析要求：
|
||
1. 识别情感倾向（positive/negative/neutral）
|
||
2. 评估分析置信度（0-1之间的浮点数）
|
||
3. 提供分析说明和依据
|
||
4. 提取影响情感判断的关键词
|
||
|
||
请严格按照指定的JSON格式返回结果。
|
||
"""
|
||
|
||
try:
|
||
agent = SubAgent(
|
||
provider="aliyun",
|
||
model_name="qwen-max",
|
||
name="sentiment_analyzer",
|
||
description="专业的文本情感分析系统",
|
||
instructions=instructions,
|
||
prompt_template=prompt_template,
|
||
response_model=SentimentAnalysis
|
||
)
|
||
|
||
print("✅ 情感分析Agent创建成功")
|
||
return agent
|
||
|
||
except Exception as e:
|
||
print(f"❌ 情感分析Agent创建失败: {e}")
|
||
raise
|
||
|
||
|
||
def demo_simple_chat():
|
||
"""演示简单对话功能"""
|
||
print("\n" + "="*50)
|
||
print("🗣️ 简单对话演示")
|
||
print("="*50)
|
||
|
||
try:
|
||
agent = create_simple_chat_agent()
|
||
|
||
# 测试问题列表
|
||
test_questions = [
|
||
"你好！请介绍一下你自己",
|
||
"什么是人工智能？",
|
||
"请给我一些学习Python的建议"
|
||
]
|
||
|
||
for i, question in enumerate(test_questions, 1):
|
||
print(f"\n问题 {i}: {question}")
|
||
try:
|
||
response = agent.run(prompt=question)
|
||
print(f"回答: {response}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 回答失败: {e}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 简单对话演示失败: {e}")
|
||
return False
|
||
|
||
|
||
def demo_structured_response():
|
||
"""演示结构化响应功能"""
|
||
print("\n" + "="*50)
|
||
print("📋 结构化响应演示")
|
||
print("="*50)
|
||
|
||
try:
|
||
agent = create_structured_output_agent()
|
||
|
||
# 测试请求列表
|
||
test_requests = [
|
||
"请解释什么是机器学习",
|
||
"介绍Python编程语言的特点",
|
||
"如何提高工作效率？"
|
||
]
|
||
|
||
for i, request in enumerate(test_requests, 1):
|
||
print(f"\n请求 {i}: {request}")
|
||
try:
|
||
result = agent.run(prompt=request)
|
||
print(f"✅ 响应成功:")
|
||
print(f" 消息: {result.message}")
|
||
print(f" 成功: {result.success}")
|
||
print(f" 时间: {result.timestamp}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 响应失败: {e}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 结构化响应演示失败: {e}")
|
||
return False
|
||
|
||
|
||
def demo_dynamic_prompt():
|
||
"""演示动态prompt模板功能"""
|
||
print("\n" + "="*50)
|
||
print("🎭 动态Prompt模板演示")
|
||
print("="*50)
|
||
|
||
try:
|
||
agent = create_sentiment_agent()
|
||
|
||
# 测试文本列表
|
||
test_texts = [
|
||
"这个产品质量非常好，我很满意！强烈推荐给大家。",
|
||
"服务态度差，产品质量也不行，非常失望。",
|
||
"产品功能还可以，价格也合理，算是中规中矩的选择。",
|
||
"今天天气不错，适合出去散步。",
|
||
"这部电影真是太精彩了！演员演技出色，剧情引人入胜。"
|
||
]
|
||
|
||
for i, text in enumerate(test_texts, 1):
|
||
print(f"\n文本 {i}: {text}")
|
||
try:
|
||
# 使用动态模板构建prompt
|
||
result = agent.run(template_vars={"text": text})
|
||
|
||
print(f"✅ 分析结果:")
|
||
print(f" 情感: {result.sentiment}")
|
||
print(f" 置信度: {result.confidence}")
|
||
print(f" 说明: {result.explanation}")
|
||
print(f" 关键词: {result.keywords}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 分析失败: {e}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 动态prompt演示失败: {e}")
|
||
return False
|
||
|
||
|
||
def demo_error_handling():
|
||
"""演示错误处理功能"""
|
||
print("\n" + "="*50)
|
||
print("⚠️ 错误处理演示")
|
||
print("="*50)
|
||
|
||
# 测试各种错误情况
|
||
error_tests = [
|
||
{
|
||
"name": "无效的提供商",
|
||
"params": {"provider": "invalid_provider", "model_name": "test"},
|
||
"expected_error": "ValueError"
|
||
},
|
||
{
|
||
"name": "空的prompt模板变量",
|
||
"params": {"provider": "aliyun", "model_name": "qwen-turbo"},
|
||
"template_vars": {},
|
||
"prompt_template": "分析这个文本: {missing_var}",
|
||
"expected_error": "SubAgentError"
|
||
}
|
||
]
|
||
|
||
for test in error_tests:
|
||
print(f"\n测试: {test['name']}")
|
||
try:
|
||
if 'prompt_template' in test:
|
||
agent = SubAgent(**test['params'])
|
||
agent.update_prompt_template(test['prompt_template'])
|
||
agent.run(template_vars=test.get('template_vars', {}))
|
||
else:
|
||
agent = SubAgent(**test['params'])
|
||
|
||
print(f"❌ 预期错误未发生")
|
||
|
||
except Exception as e:
|
||
print(f"✅ 捕获到预期错误: {type(e).__name__}: {e}")
|
||
|
||
return True
|
||
|
||
|
||
def interactive_demo():
|
||
"""交互式演示"""
|
||
print("\n" + "="*50)
|
||
print("💬 交互式演示")
|
||
print("="*50)
|
||
print("输入文本进行情感分析，输入'quit'退出")
|
||
|
||
try:
|
||
agent = create_sentiment_agent()
|
||
|
||
while True:
|
||
user_input = input("\n请输入要分析的文本: ").strip()
|
||
|
||
if user_input.lower() == 'quit':
|
||
print("再见!")
|
||
break
|
||
|
||
if not user_input:
|
||
continue
|
||
|
||
try:
|
||
print(f"正在分析: {user_input}")
|
||
result = agent.run(template_vars={"text": user_input})
|
||
|
||
print(f"\n📊 分析结果:")
|
||
print(f"情感倾向: {result.sentiment}")
|
||
print(f"置信度: {result.confidence:.3f}")
|
||
print(f"分析说明: {result.explanation}")
|
||
if result.keywords:
|
||
print(f"关键词: {', '.join(result.keywords)}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 分析失败: {e}")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n程序已中断")
|
||
except Exception as e:
|
||
print(f"❌ 交互式演示失败: {e}")
|
||
|
||
|
||
def show_agent_info(agent: SubAgent):
|
||
"""显示Agent信息"""
|
||
info = agent.get_model_info()
|
||
print(f"\n📋 Agent信息:")
|
||
for key, value in info.items():
|
||
print(f" {key}: {value}")
|
||
|
||
|
||
def main():
|
||
"""主函数 - 运行所有演示"""
|
||
print("🚀 SubAgent基础使用示例")
|
||
print("=" * 60)
|
||
|
||
# 运行所有演示
|
||
demos = [
|
||
("简单对话", demo_simple_chat),
|
||
("结构化响应", demo_structured_response),
|
||
("动态Prompt", demo_dynamic_prompt),
|
||
("错误处理", demo_error_handling),
|
||
]
|
||
|
||
results = {}
|
||
|
||
for name, demo_func in demos:
|
||
print(f"\n开始演示: {name}")
|
||
try:
|
||
success = demo_func()
|
||
results[name] = success
|
||
print(f"{'✅' if success else '❌'} {name}演示{'成功' if success else '失败'}")
|
||
except Exception as e:
|
||
print(f"❌ {name}演示异常: {e}")
|
||
results[name] = False
|
||
|
||
# 显示总结
|
||
print("\n" + "="*60)
|
||
print("📊 演示总结")
|
||
print("="*60)
|
||
|
||
total_demos = len(results)
|
||
successful_demos = sum(results.values())
|
||
|
||
for name, success in results.items():
|
||
status = "✅ 成功" if success else "❌ 失败"
|
||
print(f" {name}: {status}")
|
||
|
||
print(f"\n🎯 总计: {successful_demos}/{total_demos} 个演示成功")
|
||
|
||
# 询问是否运行交互式演示
|
||
print(f"\n是否运行交互式演示？(y/n): ", end="")
|
||
try:
|
||
choice = input().strip().lower()
|
||
if choice in ['y', 'yes', '是']:
|
||
interactive_demo()
|
||
except (KeyboardInterrupt, EOFError):
|
||
print("\n程序结束")
|
||
|
||
return successful_demos == total_demos
|
||
|
||
|
||
def test_basic_functionality():
|
||
"""测试基础功能"""
|
||
print("正在测试SubAgent基础功能...")
|
||
|
||
try:
|
||
# 创建基本Agent
|
||
agent = SubAgent(
|
||
provider="aliyun",
|
||
model_name="qwen-turbo",
|
||
name="test_agent"
|
||
)
|
||
|
||
print(f"✅ Agent创建成功: {agent}")
|
||
|
||
# 显示Agent信息
|
||
show_agent_info(agent)
|
||
|
||
# 测试简单对话
|
||
response = agent.run(prompt="请简单介绍一下你自己")
|
||
print(f"✅ 对话测试成功，响应长度: {len(str(response))}字符")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 基础功能测试失败: {e}")
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 可以选择运行测试或完整演示
|
||
import sys
|
||
|
||
if len(sys.argv) > 1 and sys.argv[1] == "--test":
|
||
# 仅运行基础测试
|
||
success = test_basic_functionality()
|
||
exit(0 if success else 1)
|
||
else:
|
||
# 运行完整演示
|
||
main() |