edbddf855d
- 实现Agent管理，支持AI辅助生成系统提示词 - 支持多个AI提供商（OpenRouter、智谱、MiniMax等） - 实现聊天室和讨论引擎 - WebSocket实时消息推送 - 前端使用React + Ant Design - 后端使用FastAPI + MongoDB Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
439 lines
12 KiB
Python
439 lines
12 KiB
Python
"""
|
||
Agent服务
|
||
管理AI代理的配置
|
||
"""
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import List, Dict, Any, Optional
|
||
from loguru import logger
|
||
|
||
from models.agent import Agent
|
||
from services.ai_provider_service import AIProviderService
|
||
|
||
|
||
class AgentService:
|
||
"""
|
||
Agent服务类
|
||
负责Agent的CRUD操作
|
||
"""
|
||
|
||
@classmethod
|
||
async def create_agent(
|
||
cls,
|
||
name: str,
|
||
role: str,
|
||
system_prompt: str,
|
||
provider_id: str,
|
||
temperature: float = 0.7,
|
||
max_tokens: int = 2000,
|
||
capabilities: Optional[Dict[str, Any]] = None,
|
||
behavior: Optional[Dict[str, Any]] = None,
|
||
avatar: Optional[str] = None,
|
||
color: str = "#1890ff"
|
||
) -> Agent:
|
||
"""
|
||
创建新的Agent
|
||
|
||
Args:
|
||
name: Agent名称
|
||
role: 角色定义
|
||
system_prompt: 系统提示词
|
||
provider_id: 使用的AI接口ID
|
||
temperature: 温度参数
|
||
max_tokens: 最大token数
|
||
capabilities: 能力配置
|
||
behavior: 行为配置
|
||
avatar: 头像URL
|
||
color: 代表颜色
|
||
|
||
Returns:
|
||
创建的Agent文档
|
||
"""
|
||
# 验证AI接口存在
|
||
provider = await AIProviderService.get_provider(provider_id)
|
||
if not provider:
|
||
raise ValueError(f"AI接口不存在: {provider_id}")
|
||
|
||
# 生成唯一ID
|
||
agent_id = f"agent-{uuid.uuid4().hex[:8]}"
|
||
|
||
# 默认能力配置
|
||
default_capabilities = {
|
||
"memory_enabled": False,
|
||
"mcp_tools": [],
|
||
"skills": [],
|
||
"multimodal": False
|
||
}
|
||
if capabilities:
|
||
default_capabilities.update(capabilities)
|
||
|
||
# 默认行为配置
|
||
default_behavior = {
|
||
"speak_threshold": 0.5,
|
||
"max_speak_per_round": 2,
|
||
"speak_style": "balanced"
|
||
}
|
||
if behavior:
|
||
default_behavior.update(behavior)
|
||
|
||
# 创建文档
|
||
agent = Agent(
|
||
agent_id=agent_id,
|
||
name=name,
|
||
role=role,
|
||
system_prompt=system_prompt,
|
||
provider_id=provider_id,
|
||
temperature=temperature,
|
||
max_tokens=max_tokens,
|
||
capabilities=default_capabilities,
|
||
behavior=default_behavior,
|
||
avatar=avatar,
|
||
color=color,
|
||
enabled=True,
|
||
created_at=datetime.utcnow(),
|
||
updated_at=datetime.utcnow()
|
||
)
|
||
|
||
await agent.insert()
|
||
|
||
logger.info(f"创建Agent: {agent_id} ({name})")
|
||
return agent
|
||
|
||
@classmethod
|
||
async def get_agent(cls, agent_id: str) -> Optional[Agent]:
|
||
"""
|
||
获取指定Agent
|
||
|
||
Args:
|
||
agent_id: Agent ID
|
||
|
||
Returns:
|
||
Agent文档或None
|
||
"""
|
||
return await Agent.find_one(Agent.agent_id == agent_id)
|
||
|
||
@classmethod
|
||
async def get_all_agents(
|
||
cls,
|
||
enabled_only: bool = False
|
||
) -> List[Agent]:
|
||
"""
|
||
获取所有Agent
|
||
|
||
Args:
|
||
enabled_only: 是否只返回启用的Agent
|
||
|
||
Returns:
|
||
Agent列表
|
||
"""
|
||
if enabled_only:
|
||
return await Agent.find(Agent.enabled == True).to_list()
|
||
return await Agent.find_all().to_list()
|
||
|
||
@classmethod
|
||
async def get_agents_by_ids(
|
||
cls,
|
||
agent_ids: List[str]
|
||
) -> List[Agent]:
|
||
"""
|
||
根据ID列表获取多个Agent
|
||
|
||
Args:
|
||
agent_ids: Agent ID列表
|
||
|
||
Returns:
|
||
Agent列表
|
||
"""
|
||
return await Agent.find(
|
||
{"agent_id": {"$in": agent_ids}}
|
||
).to_list()
|
||
|
||
@classmethod
|
||
async def update_agent(
|
||
cls,
|
||
agent_id: str,
|
||
**kwargs
|
||
) -> Optional[Agent]:
|
||
"""
|
||
更新Agent配置
|
||
|
||
Args:
|
||
agent_id: Agent ID
|
||
**kwargs: 要更新的字段
|
||
|
||
Returns:
|
||
更新后的Agent或None
|
||
"""
|
||
agent = await cls.get_agent(agent_id)
|
||
if not agent:
|
||
return None
|
||
|
||
# 如果更新了provider_id，验证其存在
|
||
if "provider_id" in kwargs:
|
||
provider = await AIProviderService.get_provider(kwargs["provider_id"])
|
||
if not provider:
|
||
raise ValueError(f"AI接口不存在: {kwargs['provider_id']}")
|
||
|
||
# 更新字段
|
||
kwargs["updated_at"] = datetime.utcnow()
|
||
|
||
for key, value in kwargs.items():
|
||
if hasattr(agent, key):
|
||
setattr(agent, key, value)
|
||
|
||
await agent.save()
|
||
|
||
logger.info(f"更新Agent: {agent_id}")
|
||
return agent
|
||
|
||
@classmethod
|
||
async def delete_agent(cls, agent_id: str) -> bool:
|
||
"""
|
||
删除Agent
|
||
|
||
Args:
|
||
agent_id: Agent ID
|
||
|
||
Returns:
|
||
是否删除成功
|
||
"""
|
||
agent = await cls.get_agent(agent_id)
|
||
if not agent:
|
||
return False
|
||
|
||
await agent.delete()
|
||
|
||
logger.info(f"删除Agent: {agent_id}")
|
||
return True
|
||
|
||
@classmethod
|
||
async def test_agent(
|
||
cls,
|
||
agent_id: str,
|
||
test_message: str = "你好，请简单介绍一下你自己。"
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
测试Agent对话
|
||
|
||
Args:
|
||
agent_id: Agent ID
|
||
test_message: 测试消息
|
||
|
||
Returns:
|
||
测试结果
|
||
"""
|
||
agent = await cls.get_agent(agent_id)
|
||
if not agent:
|
||
return {
|
||
"success": False,
|
||
"message": f"Agent不存在: {agent_id}"
|
||
}
|
||
|
||
if not agent.enabled:
|
||
return {
|
||
"success": False,
|
||
"message": "Agent已禁用"
|
||
}
|
||
|
||
# 构建消息
|
||
messages = [
|
||
{"role": "system", "content": agent.system_prompt},
|
||
{"role": "user", "content": test_message}
|
||
]
|
||
|
||
# 调用AI接口
|
||
response = await AIProviderService.chat(
|
||
provider_id=agent.provider_id,
|
||
messages=messages,
|
||
temperature=agent.temperature,
|
||
max_tokens=agent.max_tokens
|
||
)
|
||
|
||
if response.success:
|
||
return {
|
||
"success": True,
|
||
"message": "测试成功",
|
||
"response": response.content,
|
||
"model": response.model,
|
||
"tokens": response.total_tokens,
|
||
"latency_ms": response.latency_ms
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": response.error
|
||
}
|
||
|
||
@classmethod
|
||
async def duplicate_agent(
|
||
cls,
|
||
agent_id: str,
|
||
new_name: Optional[str] = None
|
||
) -> Optional[Agent]:
|
||
"""
|
||
复制Agent
|
||
|
||
Args:
|
||
agent_id: 源Agent ID
|
||
new_name: 新Agent名称
|
||
|
||
Returns:
|
||
新创建的Agent或None
|
||
"""
|
||
source_agent = await cls.get_agent(agent_id)
|
||
if not source_agent:
|
||
return None
|
||
|
||
return await cls.create_agent(
|
||
name=new_name or f"{source_agent.name} (副本)",
|
||
role=source_agent.role,
|
||
system_prompt=source_agent.system_prompt,
|
||
provider_id=source_agent.provider_id,
|
||
temperature=source_agent.temperature,
|
||
max_tokens=source_agent.max_tokens,
|
||
capabilities=source_agent.capabilities,
|
||
behavior=source_agent.behavior,
|
||
avatar=source_agent.avatar,
|
||
color=source_agent.color
|
||
)
|
||
|
||
@classmethod
|
||
async def generate_system_prompt(
|
||
cls,
|
||
provider_id: str,
|
||
name: str,
|
||
role: str,
|
||
description: Optional[str] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
使用AI生成Agent系统提示词
|
||
|
||
Args:
|
||
provider_id: AI接口ID
|
||
name: Agent名称
|
||
role: 角色定位
|
||
description: 额外描述（可选）
|
||
|
||
Returns:
|
||
生成结果，包含success和生成的prompt
|
||
"""
|
||
# 验证AI接口存在
|
||
provider = await AIProviderService.get_provider(provider_id)
|
||
if not provider:
|
||
return {
|
||
"success": False,
|
||
"message": f"AI接口不存在: {provider_id}"
|
||
}
|
||
|
||
# 构建生成提示词的请求
|
||
generate_prompt = f"""请为一个AI Agent编写系统提示词（system prompt）。
|
||
|
||
Agent名称：{name}
|
||
角色定位：{role}
|
||
{f'补充说明：{description}' if description else ''}
|
||
|
||
要求：
|
||
1. 提示词应简洁专业，控制在200字以内
|
||
2. 明确该Agent的核心职责和专业领域
|
||
3. 说明在多Agent讨论中应该关注什么
|
||
4. 使用中文编写
|
||
5. 不要包含任何问候语或开场白，直接给出提示词内容
|
||
|
||
请直接输出系统提示词，不要有任何额外的解释或包装。"""
|
||
|
||
try:
|
||
messages = [{"role": "user", "content": generate_prompt}]
|
||
|
||
response = await AIProviderService.chat(
|
||
provider_id=provider_id,
|
||
messages=messages,
|
||
temperature=0.7,
|
||
max_tokens=1000
|
||
)
|
||
|
||
if response.success:
|
||
# 清理可能的包装文本
|
||
content = response.content.strip()
|
||
# 移除可能的markdown代码块标记
|
||
if content.startswith("```"):
|
||
lines = content.split("\n")
|
||
content = "\n".join(lines[1:])
|
||
if content.endswith("```"):
|
||
content = content[:-3]
|
||
content = content.strip()
|
||
|
||
return {
|
||
"success": True,
|
||
"prompt": content,
|
||
"model": response.model,
|
||
"tokens": response.total_tokens
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"message": response.error or "生成失败"
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"生成系统提示词失败: {e}")
|
||
return {
|
||
"success": False,
|
||
"message": f"生成失败: {str(e)}"
|
||
}
|
||
|
||
|
||
# Agent预设模板
|
||
AGENT_TEMPLATES = {
|
||
"product_manager": {
|
||
"name": "产品经理",
|
||
"role": "产品规划和需求分析专家",
|
||
"system_prompt": """你是一位经验丰富的产品经理，擅长:
|
||
- 分析用户需求和痛点
|
||
- 制定产品策略和路线图
|
||
- 平衡业务目标和用户体验
|
||
- 与团队协作推进产品迭代
|
||
|
||
在讨论中，你需要从产品角度出发，关注用户价值、商业可行性和优先级排序。
|
||
请用专业但易懂的语言表达观点。""",
|
||
"color": "#1890ff"
|
||
},
|
||
"developer": {
|
||
"name": "开发工程师",
|
||
"role": "技术实现和架构设计专家",
|
||
"system_prompt": """你是一位资深的软件开发工程师，擅长:
|
||
- 系统架构设计
|
||
- 代码实现和优化
|
||
- 技术方案评估
|
||
- 性能和安全考量
|
||
|
||
在讨论中，你需要从技术角度出发，关注实现可行性、技术债务和最佳实践。
|
||
请提供具体的技术建议和潜在风险评估。""",
|
||
"color": "#52c41a"
|
||
},
|
||
"designer": {
|
||
"name": "设计师",
|
||
"role": "用户体验和界面设计专家",
|
||
"system_prompt": """你是一位专业的UI/UX设计师，擅长:
|
||
- 用户体验设计
|
||
- 界面视觉设计
|
||
- 交互流程优化
|
||
- 设计系统构建
|
||
|
||
在讨论中，你需要从设计角度出发，关注用户体验、视觉美感和交互流畅性。
|
||
请提供设计建议并考虑可用性和一致性。""",
|
||
"color": "#eb2f96"
|
||
},
|
||
"moderator": {
|
||
"name": "主持人",
|
||
"role": "讨论主持和共识判断专家",
|
||
"system_prompt": """你是讨论的主持人，负责:
|
||
- 引导讨论方向
|
||
- 总结各方观点
|
||
- 判断是否达成共识
|
||
- 提炼行动要点
|
||
|
||
在讨论中，你需要保持中立，促进有效沟通，并在适当时机总结讨论成果。
|
||
当各方观点趋于一致时，请明确指出并总结共识内容。""",
|
||
"color": "#722ed1"
|
||
}
|
||
}
|