hello-agents/Co-creation-projects/Apricity-InnocoreAI/agents/coach.py

"""
InnoCore AI 写作助教 (Coach Agent)
负责风格迁移、实时润色、解释复杂概念
"""

import asyncio
import json
from typing import Dict, List, Optional, Any
from datetime import datetime

from agents.base import BaseAgent
from core.database import db_manager
from core.vector_store import vector_store_manager
from core.exceptions import AgentException

class CoachAgent(BaseAgent):
    """写作助教智能体"""
    
    def __init__(self, llm=None):
        super().__init__("Coach", llm)
        
        # 添加工具
        self.add_tool("explain_concept", self._explain_concept, "解释复杂概念")
        self.add_tool("polish_text", self._polish_text, "润色文本")
        self.add_tool("mimic_style", self._mimic_style, "模仿写作风格")
        self.add_tool("get_user_style", self._get_user_style, "获取用户写作风格")
        self.add_tool("suggest_improvements", self._suggest_improvements, "建议改进")
    
    async def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """执行写作助教任务"""
        await self.validate_input(input_data)
        
        self.set_state("running")
        
        try:
            user_id = input_data["user_id"]
            task_type = input_data["task_type"]  # explain, polish, mimic, suggest
            content = input_data["content"]
            context = input_data.get("context", {})
            
            result = None
            
            if task_type == "explain":
                result = await self._handle_explain_task(user_id, content, context)
            elif task_type == "polish":
                result = await self._handle_polish_task(user_id, content, context)
            elif task_type == "mimic":
                result = await self._handle_mimic_task(user_id, content, context)
            elif task_type == "suggest":
                result = await self._handle_suggest_task(user_id, content, context)
            else:
                raise AgentException(f"不支持的任务类型: {task_type}")
            
            self.set_state("completed")
            
            return {
                "status": "success",
                "task_type": task_type,
                "user_id": user_id,
                "result": result,
                "timestamp": datetime.now().isoformat()
            }
            
        except Exception as e:
            self.set_state("error")
            raise AgentException(f"Coach Agent执行失败: {str(e)}")
    
    def get_required_fields(self) -> List[str]:
        """获取必需的输入字段"""
        return ["user_id", "task_type", "content"]
    
    async def _handle_explain_task(self, user_id: str, content: str, context: Dict) -> Dict[str, Any]:
        """处理解释任务"""
        try:
            # 获取用户的历史论文作为上下文
            user_context = await self._get_user_context(user_id)
            
            explain_prompt = f"""
            请用通俗易懂的语言解释以下内容：
            
            需要解释的内容：
            {content}
            
            上下文信息：
            {json.dumps(context, ensure_ascii=False, indent=2)}
            
            用户研究领域背景：
            {json.dumps(user_context, ensure_ascii=False, indent=2)}
            
            请提供：
            1. 简单易懂的解释
            2. 相关的例子或类比
            3. 在该领域的重要性
            4. 可能的应用场景
            
            请以JSON格式返回结果。
            """
            
            response = await self.think(explain_prompt)
            
            try:
                result = json.loads(response)
            except json.JSONDecodeError:
                result = {
                    "explanation": response,
                    "examples": ["需要补充具体例子"],
                    "importance": "在相关领域具有重要意义",
                    "applications": ["潜在应用场景"]
                }
            
            self._add_to_history(f"完成解释任务: {content[:50]}...")
            return result
            
        except Exception as e:
            self._add_to_history(f"解释任务失败: {str(e)}")
            return {
                "explanation": f"解释过程中出现错误: {str(e)}",
                "examples": [],
                "importance": "",
                "applications": []
            }
    
    async def _handle_polish_task(self, user_id: str, content: str, context: Dict) -> Dict[str, Any]:
        """处理润色任务"""
        try:
            # 获取用户的写作风格偏好
            user_style = await self._get_user_writing_style(user_id)
            
            # 获取相关的风格参考
            style_references = await self._get_style_references(user_id, content)
            
            polish_prompt = f"""
            请将以下文本润色为地道的学术英语：
            
            原文：
            {content}
            
            用户写作风格偏好：
            {json.dumps(user_style, ensure_ascii=False, indent=2)}
            
            风格参考：
            {json.dumps(style_references, ensure_ascii=False, indent=2)}
            
            上下文信息：
            {json.dumps(context, ensure_ascii=False, indent=2)}
            
            请提供：
            1. 润色后的英文文本
            2. 主要修改说明
            3. 风格改进建议
            4. 参考的论文句式来源
            
            要求：
            - 保持原意不变
            - 使用地道的学术表达
            - 符合目标期刊/会议的写作风格
            - 在注释中说明参考了哪些历史论文的句式
            
            请以JSON格式返回结果。
            """
            
            response = await self.think(polish_prompt)
            
            try:
                result = json.loads(response)
            except json.JSONDecodeError:
                result = {
                    "polished_text": response,
                    "modifications": ["语法修正", "词汇优化"],
                    "style_suggestions": ["建议使用更正式的表达"],
                    "references": ["基于学术写作规范"]
                }
            
            self._add_to_history(f"完成润色任务: {content[:50]}...")
            return result
            
        except Exception as e:
            self._add_to_history(f"润色任务失败: {str(e)}")
            return {
                "polished_text": content,
                "modifications": [f"润色过程中出现错误: {str(e)}"],
                "style_suggestions": [],
                "references": []
            }
    
    async def _handle_mimic_task(self, user_id: str, content: str, context: Dict) -> Dict[str, Any]:
        """处理模仿任务"""
        try:
            # 获取目标风格参考
            target_style = context.get("target_style", "formal_academic")
            reference_papers = context.get("reference_papers", [])
            
            # 如果没有指定参考论文，从用户库中获取
            if not reference_papers:
                reference_papers = await self._get_user_top_papers(user_id, limit=3)
            
            mimic_prompt = f"""
            请基于以下参考论文的写作风格，重写给定内容：
            
            原文：
            {content}
            
            目标风格：
            {target_style}
            
            参考论文：
            {json.dumps(reference_papers, ensure_ascii=False, indent=2)}
            
            上下文信息：
            {json.dumps(context, ensure_ascii=False, indent=2)}
            
            请提供：
            1. 重写后的文本
            2. 风格分析（说明如何体现目标风格）
            3. 具体的模仿技巧
            4. 参考的句式结构
            
            请以JSON格式返回结果。
            """
            
            response = await self.think(mimic_prompt)
            
            try:
                result = json.loads(response)
            except json.JSONDecodeError:
                result = {
                    "rewritten_text": response,
                    "style_analysis": "基于学术写作风格进行重写",
                    "mimic_techniques": ["句式结构模仿", "词汇选择"],
                    "reference_structures": ["学术表达方式"]
                }
            
            self._add_to_history(f"完成模仿任务: {content[:50]}...")
            return result
            
        except Exception as e:
            self._add_to_history(f"模仿任务失败: {str(e)}")
            return {
                "rewritten_text": content,
                "style_analysis": f"模仿过程中出现错误: {str(e)}",
                "mimic_techniques": [],
                "reference_structures": []
            }
    
    async def _handle_suggest_task(self, user_id: str, content: str, context: Dict) -> Dict[str, Any]:
        """处理建议任务"""
        try:
            # 获取用户的历史写作数据
            user_writing_history = await self._get_user_writing_history(user_id)
            
            suggest_prompt = f"""
            请对以下文本提供改进建议：
            
            文本内容：
            {content}
            
            用户写作历史：
            {json.dumps(user_writing_history, ensure_ascii=False, indent=2)}
            
            上下文信息：
            {json.dumps(context, ensure_ascii=False, indent=2)}
            
            请提供：
            1. 整体评价
            2. 具体改进建议（按重要性排序）
            3. 语法和表达问题
            4. 结构优化建议
            5. 学术表达改进
            
            请以JSON格式返回结果。
            """
            
            response = await self.think(suggest_prompt)
            
            try:
                result = json.loads(response)
            except json.JSONDecodeError:
                result = {
                    "overall_evaluation": "文本整体质量良好",
                    "improvement_suggestions": ["建议加强逻辑表达", "可以增加更多细节"],
                    "grammar_issues": ["检查时态一致性"],
                    "structure_suggestions": ["建议优化段落结构"],
                    "academic_improvements": ["使用更正式的学术词汇"]
                }
            
            self._add_to_history(f"完成建议任务: {content[:50]}...")
            return result
            
        except Exception as e:
            self._add_to_history(f"建议任务失败: {str(e)}")
            return {
                "overall_evaluation": f"分析过程中出现错误: {str(e)}",
                "improvement_suggestions": [],
                "grammar_issues": [],
                "structure_suggestions": [],
                "academic_improvements": []
            }
    
    async def _get_user_context(self, user_id: str) -> Dict[str, Any]:
        """获取用户的研究背景"""
        try:
            user = await db_manager.get_user(user_id)
            if user:
                return user.get("profile", {})
            return {}
        except Exception:
            return {}
    
    async def _get_user_writing_style(self, user_id: str) -> Dict[str, Any]:
        """获取用户写作风格偏好"""
        user_context = await self._get_user_context(user_id)
        return user_context.get("writing_style", {
            "tone": "formal",
            "complexity": "medium",
            "preferred_journals": ["Nature", "Science"],
            "language": "english"
        })
    
    async def _get_style_references(self, user_id: str, content: str) -> List[Dict[str, Any]]:
        """获取风格参考"""
        try:
            # 搜索用户库中的相关论文
            search_results = await vector_store_manager.hybrid_search(
                query=content,
                user_id=user_id,
                top_k=3,
                include_l2=True,
                include_l1=False
            )
            
            references = []
            for result in search_results:
                payload = result["payload"]
                references.append({
                    "title": payload.get("title", ""),
                    "abstract": payload.get("abstract", "")[:200],
                    "similarity": result["score"]
                })
            
            return references
            
        except Exception:
            return []
    
    async def _get_user_top_papers(self, user_id: str, limit: int = 3) -> List[Dict[str, Any]]:
        """获取用户评分最高的论文"""
        try:
            user_papers = await db_manager.get_user_papers(user_id, limit=limit)
            
            top_papers = []
            for paper in user_papers:
                top_papers.append({
                    "title": paper.get("title", ""),
                    "abstract": paper.get("abstract", "")[:300],
                    "rating": paper.get("rating", 0),
                    "authors": paper.get("authors", [])
                })
            
            return top_papers
            
        except Exception:
            return []
    
    async def _get_user_writing_history(self, user_id: str) -> List[Dict[str, Any]]:
        """获取用户写作历史"""
        try:
            # 这里应该从用户的写作历史记录中获取数据
            # 暂时返回模拟数据
            return [
                {
                    "date": "2024-01-01",
                    "content_type": "abstract",
                    "word_count": 200,
                    "feedback_score": 4.5
                }
            ]
        except Exception:
            return []
    
    # 工具方法
    async def _explain_concept(self, concept: str, context: Dict = None) -> Dict:
        """解释概念工具"""
        return await self._handle_explain_task(
            context.get("user_id", ""), 
            concept, 
            context or {}
        )
    
    async def _polish_text(self, text: str, context: Dict = None) -> Dict:
        """润色文本工具"""
        return await self._handle_polish_task(
            context.get("user_id", ""), 
            text, 
            context or {}
        )
    
    async def _mimic_style(self, text: str, target_style: str, context: Dict = None) -> Dict:
        """模仿风格工具"""
        ctx = context or {}
        ctx["target_style"] = target_style
        return await self._handle_mimic_task(
            ctx.get("user_id", ""), 
            text, 
            ctx
        )
    
    async def _get_user_style(self, user_id: str) -> Dict:
        """获取用户风格工具"""
        return await self._get_user_writing_style(user_id)
    
    async def _suggest_improvements(self, text: str, context: Dict = None) -> Dict:
        """建议改进工具"""
        return await self._handle_suggest_task(
            context.get("user_id", ""), 
            text, 
            context or {}
        )