智能体服务封装

import asyncio
import logging
from typing import Dict, List, Optional, Any, AsyncIterator, Union
from third.TranssionAgentClient import agent_client_instance
from constants.conversation_constants import AgentConstants

logger = logging.getLogger(name)

class AgentService:
“”“智能体服务封装类”“”

def __init__(self): self.client = agent_client_instanceasync def chat(self, messages: List[Dict[str, str]], options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: \"\"\" 智能体普通对话 Args: messages: 消息列表，格式: [{\"role\": \"user\", \"content\": \"消息内容\"}] options: 可选参数，支持: - dellE3Size: 如果用的是dellE3，生成图片的尺寸大小，必须是 1792x1024、1024x1024 或 1024x1792 - userId: 如果是传神云文档问答需要在智库授权，并需要传用户工号 Returns: Dict包含以下字段: - code: 响应码 - data: 原始模型返回的结构体 - dataContent: 返回的消息内容（且加工后的） - message: 响应消息 - success: 是否成功 Example: result = await agent_service.chat( [{\"role\": \"user\", \"content\": \"你好\"}] ) content = result.get(\"dataContent\", \"\") \"\"\" try: if not self.client.is_configured(): raise ValueError(\"智能体客户端未正确配置，请检查 AGENT_APP_ID、AGENT_APP_SECRET 和 AGENT_ASSISTANT_CODE\") # 验证消息格式 if not messages or not isinstance(messages, list): raise ValueError(\"messages 参数必须是非空列表\") for msg in messages: if not isinstance(msg, dict) or \"role\" not in msg or \"content\" not in msg: raise ValueError(\"消息格式错误，每个消息必须包含 role 和 content 字段\") if msg[\"role\"] not in [\"user\", \"assistant\"]: raise ValueError(\"消息角色必须是 \'user\' 或 \'assistant\'\") logger.info(f\"[AgentService.chat] 调用智能体普通对话\") result = await self.client.chat(messages, options) # 统一响应格式处理 if result.get(AgentConstants.FIELD_CODE) == AgentConstants.HTTP_SUCCESS and result.get(AgentConstants.FIELD_SUCCESS): logger.info(f\"[AgentService.chat] 调用成功，响应内容长度: {len(result.get(\'dataContent\', \'\'))}\") else: logger.warning(f\"[AgentService.chat] 调用失败，错误信息: {result.get(AgentConstants.FIELD_MESSAGE, \'未知错误\')}\") return result except Exception as e: logger.error(f\"[AgentService.chat] 智能体对话失败: {str(e)}\") return { AgentConstants.FIELD_CODE: AgentConstants.HTTP_SERVER_ERROR, AgentConstants.FIELD_MESSAGE: f\"智能体调用失败: {str(e)}\", AgentConstants.FIELD_DATA: None, \"dataContent\": \"\", AgentConstants.FIELD_SUCCESS: False }async def stream_chat(self, messages: List[Dict[str, str]],options: Optional[Dict[str, Any]] = None) -> AsyncIterator[Dict[str, Any]]: \"\"\" 智能体流式对话 Args: messages: 消息列表 options: 可选参数 Yields: Dict包含以下字段: - data: 当前的状态 - dataContent: 返回的消息内容（且加工后的） - dataObject: 原始的gpt返回对象 - errorMsg: 错误异常消息 - errorCode: 错误码 Example: async for chunk in agent_service.stream_chat( [{\"role\": \"user\", \"content\": \"写一篇文章\"}] ): content = chunk.get(\"dataContent\", \"\") if content: print(content, end=\"\", flush=True) \"\"\" try: if not self.client.is_configured(): yield { \"data\": \"error\", \"dataContent\": \"\", \"dataObject\": None, \"errorMsg\": \"智能体客户端未正确配置，请检查 AGENT_APP_ID、AGENT_APP_SECRET 和 AGENT_ASSISTANT_CODE\", \"errorCode\": 500 } return # 验证消息格式 if not messages or not isinstance(messages, list): yield { \"data\": \"error\",  \"dataContent\": \"\", \"dataObject\": None, \"errorMsg\": \"messages 参数必须是非空列表\", \"errorCode\": 400 } return for msg in messages: if not isinstance(msg, dict) or \"role\" not in msg or \"content\" not in msg: yield {  \"data\": \"error\",  \"dataContent\": \"\",  \"dataObject\": None,  \"errorMsg\": \"消息格式错误，每个消息必须包含 role 和 content 字段\",  \"errorCode\": 400 } return if msg[\"role\"] not in [\"user\", \"assistant\"]: yield {  \"data\": \"error\",  \"dataContent\": \"\",  \"dataObject\": None,  \"errorMsg\": \"消息角色必须是 \'user\' 或 \'assistant\'\",  \"errorCode\": 400 } return logger.info(f\"[AgentService.stream_chat] 调用智能体流式对话\") async for chunk in self.client.stream_chat(messages, options): yield chunk except Exception as e: logger.error(f\"[AgentService.stream_chat] 智能体流式对话失败: {str(e)}\") yield { \"data\": \"error\", \"dataContent\": \"\", \"dataObject\": None, \"errorMsg\": f\"智能体流式调用失败: {str(e)}\", \"errorCode\": 500 }async def close(self): \"\"\"关闭服务连接\"\"\" await self.client.close()def is_available(self) -> bool: \"\"\"检查服务是否可用\"\"\" return self.client.is_configured()

创建全局服务实例

agent_service = AgentService()