#!/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()