Files
aiagent/backend/scripts/generate_batch_agents.py

627 lines
18 KiB
Python
Raw Permalink Normal View History

2026-01-20 18:05:31 +08:00
#!/usr/bin/env python3
"""
批量生成多个Agent
生成一批不同类型的Agent展示各种工作流模式
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy.orm import Session
from app.core.database import SessionLocal
from app.models.agent import Agent
from app.models.user import User
from datetime import datetime
import uuid
def generate_text_summary_agent(db: Session, user: User):
"""生成文本摘要Agent"""
nodes = []
edges = []
start_node = {
"id": "start-1",
"type": "start",
"position": {"x": 50, "y": 300},
"data": {"label": "开始", "output_format": "json"}
}
nodes.append(start_node)
summary_node = {
"id": "llm-summary",
"type": "llm",
"position": {"x": 250, "y": 300},
"data": {
"label": "文本摘要",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.3",
"max_tokens": "2000",
"prompt": """你是一个专业的文本摘要专家。请对以下文本进行摘要。
文本内容{{query}}
请生成
1. 核心要点3-5
2. 简要摘要100-200
3. 关键词5-10
请以Markdown格式输出"""
}
}
nodes.append(summary_node)
end_node = {
"id": "end-1",
"type": "end",
"position": {"x": 450, "y": 300},
"data": {"label": "结束"}
}
nodes.append(end_node)
edges.append({"id": "e1", "source": "start-1", "target": "llm-summary", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e2", "source": "llm-summary", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"})
return {
"name": "文本摘要Agent",
"description": "智能文本摘要工具,能够提取文本核心要点、生成简要摘要和关键词。",
"workflow_config": {"nodes": nodes, "edges": edges}
}
def generate_code_review_agent(db: Session, user: User):
"""生成代码审查Agent"""
nodes = []
edges = []
start_node = {
"id": "start-1",
"type": "start",
"position": {"x": 50, "y": 300},
"data": {"label": "开始", "output_format": "json"}
}
nodes.append(start_node)
analysis_node = {
"id": "llm-analysis",
"type": "llm",
"position": {"x": 250, "y": 300},
"data": {
"label": "代码分析",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.2",
"max_tokens": "3000",
"prompt": """你是一个资深的代码审查专家。请审查以下代码。
代码{{query}}
请检查
1. 代码规范命名格式注释
2. 潜在bug和错误
3. 性能问题
4. 安全性问题
5. 最佳实践建议
请以Markdown格式输出包含问题列表和改进建议"""
}
}
nodes.append(analysis_node)
end_node = {
"id": "end-1",
"type": "end",
"position": {"x": 450, "y": 300},
"data": {"label": "结束"}
}
nodes.append(end_node)
edges.append({"id": "e1", "source": "start-1", "target": "llm-analysis", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e2", "source": "llm-analysis", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"})
return {
"name": "代码审查Agent",
"description": "专业的代码审查工具能够检查代码规范、潜在bug、性能问题和安全性。",
"workflow_config": {"nodes": nodes, "edges": edges}
}
def generate_translation_agent(db: Session, user: User):
"""生成翻译Agent"""
nodes = []
edges = []
start_node = {
"id": "start-1",
"type": "start",
"position": {"x": 50, "y": 300},
"data": {"label": "开始", "output_format": "json"}
}
nodes.append(start_node)
detect_node = {
"id": "llm-detect",
"type": "llm",
"position": {"x": 250, "y": 300},
"data": {
"label": "语言检测",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.1",
"max_tokens": "500",
"prompt": """请检测以下文本的语言类型。
文本{{query}}
请输出JSON格式
{
"language": "检测到的语言(中文/英文/日文等)",
"confidence": "置信度(高/中/低)"
}"""
}
}
nodes.append(detect_node)
translate_node = {
"id": "llm-translate",
"type": "llm",
"position": {"x": 450, "y": 300},
"data": {
"label": "翻译",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.3",
"max_tokens": "2000",
"prompt": """你是一个专业的翻译专家。请翻译以下文本。
原文{{query}}
语言信息{{output}}
请提供
1. 翻译结果
2. 翻译说明如有特殊处理
请以Markdown格式输出"""
}
}
nodes.append(translate_node)
end_node = {
"id": "end-1",
"type": "end",
"position": {"x": 650, "y": 300},
"data": {"label": "结束"}
}
nodes.append(end_node)
edges.append({"id": "e1", "source": "start-1", "target": "llm-detect", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e2", "source": "llm-detect", "target": "llm-translate", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e3", "source": "llm-translate", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"})
return {
"name": "智能翻译Agent",
"description": "多语言翻译工具,支持语言自动检测和高质量翻译。",
"workflow_config": {"nodes": nodes, "edges": edges}
}
def generate_qa_agent(db: Session, user: User):
"""生成问答助手Agent"""
nodes = []
edges = []
start_node = {
"id": "start-1",
"type": "start",
"position": {"x": 50, "y": 300},
"data": {"label": "开始", "output_format": "json"}
}
nodes.append(start_node)
understand_node = {
"id": "llm-understand",
"type": "llm",
"position": {"x": 250, "y": 300},
"data": {
"label": "问题理解",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.3",
"max_tokens": "1000",
"prompt": """请分析用户的问题,提取关键信息。
用户问题{{query}}
请输出JSON格式
{
"question_type": "问题类型(技术/生活/学习等)",
"keywords": ["关键词1", "关键词2"],
"intent": "用户意图"
}"""
}
}
nodes.append(understand_node)
answer_node = {
"id": "llm-answer",
"type": "llm",
"position": {"x": 450, "y": 300},
"data": {
"label": "生成答案",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.7",
"max_tokens": "2000",
"prompt": """你是一个知识渊博的助手。请回答用户的问题。
用户问题{{query}}
问题分析{{output}}
请提供
1. 直接答案
2. 详细解释
3. 相关建议
请以Markdown格式输出确保答案准确清晰有用"""
}
}
nodes.append(answer_node)
end_node = {
"id": "end-1",
"type": "end",
"position": {"x": 650, "y": 300},
"data": {"label": "结束"}
}
nodes.append(end_node)
edges.append({"id": "e1", "source": "start-1", "target": "llm-understand", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e2", "source": "llm-understand", "target": "llm-answer", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e3", "source": "llm-answer", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"})
return {
"name": "智能问答助手",
"description": "智能问答系统,能够理解问题意图并提供详细准确的答案。",
"workflow_config": {"nodes": nodes, "edges": edges}
}
def generate_document_agent(db: Session, user: User):
"""生成文档生成Agent"""
nodes = []
edges = []
start_node = {
"id": "start-1",
"type": "start",
"position": {"x": 50, "y": 300},
"data": {"label": "开始", "output_format": "json"}
}
nodes.append(start_node)
plan_node = {
"id": "llm-plan",
"type": "llm",
"position": {"x": 250, "y": 300},
"data": {
"label": "文档规划",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.3",
"max_tokens": "1500",
"prompt": """请根据用户需求规划文档结构。
用户需求{{query}}
请输出JSON格式的文档大纲
{
"title": "文档标题",
"sections": [
{"name": "章节1", "content": "内容描述"},
{"name": "章节2", "content": "内容描述"}
]
}"""
}
}
nodes.append(plan_node)
generate_node = {
"id": "llm-generate",
"type": "llm",
"position": {"x": 450, "y": 300},
"data": {
"label": "生成文档",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.5",
"max_tokens": "4000",
"prompt": """请根据文档规划生成完整的文档内容。
用户需求{{query}}
文档规划{{output}}
请生成结构完整内容详实的Markdown文档"""
}
}
nodes.append(generate_node)
end_node = {
"id": "end-1",
"type": "end",
"position": {"x": 650, "y": 300},
"data": {"label": "结束"}
}
nodes.append(end_node)
edges.append({"id": "e1", "source": "start-1", "target": "llm-plan", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e2", "source": "llm-plan", "target": "llm-generate", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e3", "source": "llm-generate", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"})
return {
"name": "文档生成Agent",
"description": "智能文档生成工具,能够根据需求规划文档结构并生成完整内容。",
"workflow_config": {"nodes": nodes, "edges": edges}
}
def generate_data_analysis_agent(db: Session, user: User):
"""生成数据分析Agent"""
nodes = []
edges = []
start_node = {
"id": "start-1",
"type": "start",
"position": {"x": 50, "y": 300},
"data": {"label": "开始", "output_format": "json"}
}
nodes.append(start_node)
parse_node = {
"id": "llm-parse",
"type": "llm",
"position": {"x": 250, "y": 300},
"data": {
"label": "数据解析",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.2",
"max_tokens": "2000",
"prompt": """请解析用户提供的数据。
数据内容{{query}}
请输出JSON格式
{
"data_type": "数据类型(表格/列表/文本等)",
"structure": "数据结构描述",
"key_fields": ["字段1", "字段2"]
}"""
}
}
nodes.append(parse_node)
analysis_node = {
"id": "llm-analysis",
"type": "llm",
"position": {"x": 450, "y": 300},
"data": {
"label": "数据分析",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.3",
"max_tokens": "3000",
"prompt": """请对数据进行深入分析。
原始数据{{query}}
数据解析{{output}}
请提供
1. 数据概览
2. 关键指标
3. 趋势分析
4. 洞察建议
请以Markdown格式输出包含数据表格和图表描述"""
}
}
nodes.append(analysis_node)
end_node = {
"id": "end-1",
"type": "end",
"position": {"x": 650, "y": 300},
"data": {"label": "结束"}
}
nodes.append(end_node)
edges.append({"id": "e1", "source": "start-1", "target": "llm-parse", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e2", "source": "llm-parse", "target": "llm-analysis", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e3", "source": "llm-analysis", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"})
return {
"name": "数据分析Agent",
"description": "智能数据分析工具,能够解析数据、提取关键指标并提供深度洞察。",
"workflow_config": {"nodes": nodes, "edges": edges}
}
def generate_creative_writing_agent(db: Session, user: User):
"""生成创意写作Agent"""
nodes = []
edges = []
start_node = {
"id": "start-1",
"type": "start",
"position": {"x": 50, "y": 300},
"data": {"label": "开始", "output_format": "json"}
}
nodes.append(start_node)
brainstorm_node = {
"id": "llm-brainstorm",
"type": "llm",
"position": {"x": 250, "y": 300},
"data": {
"label": "头脑风暴",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.8",
"max_tokens": "1500",
"prompt": """你是一个创意写作专家。请根据用户需求进行头脑风暴。
用户需求{{query}}
请提供
1. 创意主题3-5
2. 故事大纲
3. 角色设定
4. 写作风格建议
请以Markdown格式输出"""
}
}
nodes.append(brainstorm_node)
write_node = {
"id": "llm-write",
"type": "llm",
"position": {"x": 450, "y": 300},
"data": {
"label": "创作内容",
"provider": "deepseek",
"model": "deepseek-chat",
"temperature": "0.9",
"max_tokens": "4000",
"prompt": """请根据创意方案进行创作。
用户需求{{query}}
创意方案{{output}}
请创作一篇完整的作品文章故事诗歌等确保内容生动有趣有创意"""
}
}
nodes.append(write_node)
end_node = {
"id": "end-1",
"type": "end",
"position": {"x": 650, "y": 300},
"data": {"label": "结束"}
}
nodes.append(end_node)
edges.append({"id": "e1", "source": "start-1", "target": "llm-brainstorm", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e2", "source": "llm-brainstorm", "target": "llm-write", "sourceHandle": "right", "targetHandle": "left"})
edges.append({"id": "e3", "source": "llm-write", "target": "end-1", "sourceHandle": "right", "targetHandle": "left"})
return {
"name": "创意写作Agent",
"description": "创意写作助手,能够进行头脑风暴并创作各种类型的创意内容。",
"workflow_config": {"nodes": nodes, "edges": edges}
}
def generate_batch_agents(db: Session, username: str = "admin"):
"""批量生成Agent"""
print("=" * 60)
print("批量生成Agent")
print("=" * 60)
print()
# 查找用户
user = db.query(User).filter(User.username == username).first()
if not user:
print(f"❌ 未找到用户 '{username}',请先创建该用户")
return
print(f"✅ 找到用户: {user.username} (ID: {user.id})")
print()
# 定义要生成的Agent列表
agent_generators = [
generate_text_summary_agent,
generate_code_review_agent,
generate_translation_agent,
generate_qa_agent,
generate_document_agent,
generate_data_analysis_agent,
generate_creative_writing_agent,
]
created_count = 0
updated_count = 0
failed_count = 0
for generator in agent_generators:
try:
agent_data = generator(db, user)
agent_name = agent_data["name"]
# 检查Agent是否已存在
existing_agent = db.query(Agent).filter(
Agent.name == agent_name,
Agent.user_id == user.id
).first()
if existing_agent:
existing_agent.workflow_config = agent_data["workflow_config"]
existing_agent.description = agent_data["description"]
existing_agent.updated_at = datetime.now()
existing_agent.status = "published"
updated_count += 1
print(f"⚠️ 更新Agent: {agent_name}")
else:
agent = Agent(
id=str(uuid.uuid4()),
name=agent_name,
description=agent_data["description"],
workflow_config=agent_data["workflow_config"],
status="published",
user_id=user.id,
version=1
)
db.add(agent)
created_count += 1
print(f"✅ 创建Agent: {agent_name}")
except Exception as e:
failed_count += 1
print(f"❌ 生成Agent失败: {generator.__name__} - {str(e)}")
import traceback
traceback.print_exc()
try:
db.commit()
print()
print("=" * 60)
print("✅ 批量生成完成!")
print("=" * 60)
print(f" - 新建: {created_count}")
print(f" - 更新: {updated_count}")
print(f" - 失败: {failed_count}")
print()
print("📋 生成的Agent列表")
for generator in agent_generators:
agent_data = generator(db, user)
print(f"{agent_data['name']}")
print()
except Exception as e:
db.rollback()
print(f"❌ 提交失败: {str(e)}")
import traceback
traceback.print_exc()
def main():
"""主函数"""
db = SessionLocal()
try:
generate_batch_agents(db, username="admin")
finally:
db.close()
if __name__ == "__main__":
main()