Agentic AI 学术研究与工程应用平台 - 实施计划方案

cognizes/
├── main.py                        # FastAPI 应用入口
├── api/
│   ├── __init__.py
│   ├── routes/
│   │   ├── sources.py             # 内容管理 API
│   │   ├── tasks.py               # 任务管理 API
│   │   ├── search.py              # 搜索 API
│   │   └── health.py              # 健康检查
│   ├── services/
│   │   ├── source_service.py      # 内容服务
│   │   ├── task_service.py        # 任务服务
│   │   └── search_service.py      # 搜索服务
│   └── websocket/
│       └── task_events.py         # WebSocket 任务事件
├── core/
│   ├── config.py                  # 配置管理
│   ├── database.py                # 数据库连接
│   ├── exceptions.py              # 异常定义
│   └── models/                    # Pydantic 模型
└── agents/                        # Agent 层 (Task 1.2)

cognizes/agents/
├── adk/                           # Google ADK 实现 (Phase 1)
│   ├── __init__.py
│   ├── coordinator.py             # 中央协调 Agent (SequentialAgent)
│   ├── reader_agent.py            # 多源内容解析 Agent (LlmAgent)
│   ├── translation_agent.py       # 翻译 Agent (LlmAgent)
│   ├── heartfelt_agent.py         # 深度分析 Agent (LlmAgent)
│   ├── tools/                     # 自定义工具
│   │   ├── pdf_parser.py
│   │   ├── web_scraper.py
│   │   └── oceanbase_search.py
│   └── workflows/                 # 工作流编排
│       └── content_pipeline.py
└── claude/                        # Claude SDK 实现 (Phase 2)
    ├── skills.py                  # Agent Skills 调用封装
    └── cognizes_agent.py          # 认知增强 Agent

# 安装 Google ADK
pip install google-adk

# 配置 Google Cloud 认证
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"
# 或使用
export GOOGLE_API_KEY="your-google-api-key"

# cognizes/agents/adk/coordinator.py
from google.adk.agents import LlmAgent, SequentialAgent, ParallelAgent
from .reader_agent import create_reader_agent
from .translation_agent import create_translation_agent
from .heartfelt_agent import create_heartfelt_agent

def create_content_pipeline():
    """创建内容处理工作流"""

    # 1. Reader Agent - 解析内容
    reader = create_reader_agent()

    # 2. 并行阶段：翻译 + 分析
    translation = create_translation_agent()
    analysis = create_heartfelt_agent()

    parallel_stage = ParallelAgent(
        name="parallel_processing",
        sub_agents=[translation, analysis]
    )

    # 3. 组合为顺序工作流
    return SequentialAgent(
        name="content_pipeline",
        sub_agents=[reader, parallel_stage]
    )

# cognizes/agents/adk/reader_agent.py
from google.adk.agents import LlmAgent
from .tools.pdf_parser import parse_pdf
from .tools.web_scraper import scrape_url

def create_reader_agent() -> LlmAgent:
    """创建 Reader Agent"""
    return LlmAgent(
        model="gemini-2.0-flash",
        name="reader_agent",
        description="解析多种格式的内容源",
        instruction="""你是一个内容解析专家。

任务流程：
1. 根据输入类型选择合适的解析工具
2. 提取标题、摘要、作者、正文等关键信息
3. 输出结构化的内容数据

支持格式：PDF, Markdown, URL, arXiv""",
        tools=[parse_pdf, scrape_url]
    )

# cognizes/agents/adk/translation_agent.py
from google.adk.agents import LlmAgent

def create_translation_agent() -> LlmAgent:
    """创建翻译 Agent"""
    return LlmAgent(
        model="gemini-2.0-flash",
        name="translation_agent",
        description="学术内容翻译",
        instruction="""你是一个学术翻译专家，负责将英文学术内容翻译为中文。

翻译规范：
1. 保留学术术语原文：Chain-of-Thought、ReAct、RAG 等
2. 保持原文结构：标题层级、列表、代码块
3. 术语表格：附加关键术语英中对照表
4. 质量目标：BLEU > 0.7"""
    )

# cognizes/agents/adk/heartfelt_agent.py
from google.adk.agents import LlmAgent
from .tools.oceanbase_search import semantic_search

def create_heartfelt_agent() -> LlmAgent:
    """创建深度分析 Agent"""
    return LlmAgent(
        model="gemini-2.0-flash",
        name="heartfelt_agent",
        description="论文深度分析与洞察提取",
        instruction="""你是一个学术分析专家，负责深度分析论文内容。

分析维度：
1. 核心创新点提取
2. 方法论优缺点分析
3. 与相关工作对比
4. 实践应用场景
5. 研究局限与未来方向

输出格式：结构化 Markdown 分析报告""",
        tools=[semantic_search]
    )

-- 1. 创建数据库
CREATE DATABASE IF NOT EXISTS cognizes DEFAULT CHARSET utf8mb4;

-- 2. 内容元数据表
CREATE TABLE sources (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    source_type ENUM('paper', 'article', 'document', 'code_repo') NOT NULL,
    title VARCHAR(500) NOT NULL,
    abstract TEXT,
    authors JSON,
    url VARCHAR(1000),
    format VARCHAR(50),
    publication_date DATE,
    category VARCHAR(100),
    status ENUM('pending', 'processing', 'translated', 'analyzed') DEFAULT 'pending',
    metadata JSON,
    file_path VARCHAR(1000),
    translation_path VARCHAR(1000),
    analysis_path VARCHAR(1000),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- 3. 向量表
CREATE TABLE source_embeddings (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    source_id BIGINT NOT NULL,
    chunk_index INT DEFAULT 0,
    chunk_text TEXT,
    embedding VECTOR(1536),
    FOREIGN KEY (source_id) REFERENCES sources(id) ON DELETE CASCADE
);

-- 4. HNSW 向量索引
CREATE INDEX idx_embedding_hnsw ON source_embeddings
USING HNSW (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 128);

-- 5. 任务表
CREATE TABLE tasks (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    source_id BIGINT,
    task_type ENUM('parse', 'translate', 'analyze', 'full') NOT NULL,
    status ENUM('pending', 'running', 'completed', 'failed', 'cancelled') DEFAULT 'pending',
    progress FLOAT DEFAULT 0,
    error_message TEXT,
    started_at TIMESTAMP,
    completed_at TIMESTAMP,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (source_id) REFERENCES sources(id) ON DELETE CASCADE
);

-- 6. 全文索引（用于关键词检索）
CREATE FULLTEXT INDEX idx_sources_fulltext ON sources (title, abstract);

ui/src/
├── app/
│   ├── layout.tsx             # 全局布局
│   ├── page.tsx               # 首页（仪表板）
│   ├── sources/
│   │   ├── page.tsx           # 内容列表
│   │   ├── [id]/page.tsx      # 内容详情
│   │   └── upload/page.tsx    # 上传页面
│   ├── tasks/page.tsx         # 任务监控
│   └── search/page.tsx        # 搜索页面
├── components/
│   ├── SourceCard.tsx         # 内容卡片
│   ├── UploadZone.tsx         # 上传区域
│   ├── TaskList.tsx           # 任务列表
│   └── SearchBox.tsx          # 搜索框
├── hooks/
│   ├── useApi.ts              # API 调用
│   └── useWebSocket.ts        # WebSocket 连接
└── store/
    ├── sourceStore.ts         # 内容状态 (Zustand)
    └── taskStore.ts           # 任务状态 (Zustand)

version: "3.8"
services:
  neo4j:
    image: neo4j:5.26.0
    ports:
      - "7474:7474" # HTTP
      - "7687:7687" # Bolt
    environment:
      - NEO4J_AUTH=neo4j/password
      - NEO4J_PLUGINS=["apoc", "graph-data-science"]
    volumes:
      - neo4j_data:/data
      - neo4j_logs:/logs
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:7474"]
      interval: 30s
      timeout: 10s
      retries: 3

volumes:
  neo4j_data:
  neo4j_logs:

// 1. 创建约束 - 确保唯一性
CREATE CONSTRAINT source_id_unique FOR (s:Source) REQUIRE s.id IS UNIQUE;
CREATE CONSTRAINT paper_id_unique FOR (p:Paper) REQUIRE p.id IS UNIQUE;
CREATE CONSTRAINT article_id_unique FOR (a:Article) REQUIRE a.id IS UNIQUE;
CREATE CONSTRAINT author_name_unique FOR (a:Author) REQUIRE a.name IS UNIQUE;
CREATE CONSTRAINT concept_name_unique FOR (c:Concept) REQUIRE c.name IS UNIQUE;
CREATE CONSTRAINT method_name_unique FOR (m:Method) REQUIRE m.name IS UNIQUE;

// 2. 创建向量索引
CREATE VECTOR INDEX source_embedding FOR (s:Source) ON (s.embedding)
OPTIONS {
  indexConfig: {
    `vector.dimensions`: 1536,
    `vector.similarity_function`: 'cosine'
  }
};

// 3. 创建全文索引
CREATE FULLTEXT INDEX source_fulltext FOR (s:Source) ON EACH [s.title, s.abstract];

# cognizes/core/cognee_config.py
import cognee
from cognee.infrastructure.databases.graph import Neo4jConfig
from cognee.infrastructure.databases.vector import QdrantConfig

async def init_cognee():
    """初始化 Cognee 配置"""

    # 1. 配置 LLM
    cognee.config.set_llm_api_key(os.getenv("ANTHROPIC_API_KEY"))
    cognee.config.set_llm_provider("anthropic")
    cognee.config.set_llm_model("claude-sonnet-4-20250514")

    # 2. 配置 Embedding
    cognee.config.set_embedding_provider("openai")
    cognee.config.set_embedding_model("text-embedding-3-small")

    # 3. 配置图存储 (Neo4j)
    cognee.config.set_graph_db_config(Neo4jConfig(
        uri=os.getenv("NEO4J_URI", "bolt://localhost:7687"),
        username=os.getenv("NEO4J_USER", "neo4j"),
        password=os.getenv("NEO4J_PASSWORD", "password")
    ))

    # 4. 配置向量存储 (使用 OceanBase 或 Qdrant)
    # 注：Cognee 原生支持 Qdrant，OceanBase 需自定义适配器

    # 5. 重置数据（可选，开发时使用）
    # await cognee.prune.prune_data()
    # await cognee.prune.prune_system(metadata=True)

# cognizes/core/memory.py
import cognee
from typing import List, Dict, Any

class CogneeMemory:
    """Cognee 认知记忆层封装"""

    async def add_content(self, content: str, dataset: str = "default") -> None:
        """添加内容到记忆层"""
        await cognee.add(content, dataset_name=dataset)

    async def add_file(self, file_path: str, dataset: str = "default") -> None:
        """添加文件到记忆层"""
        await cognee.add(file_path, dataset_name=dataset)

    async def cognify(self) -> None:
        """处理数据，构建知识图谱"""
        await cognee.cognify()

    async def search(self, query: str, search_type: str = "INSIGHTS") -> List[Dict]:
        """混合检索"""
        results = await cognee.search(
            query_text=query,
            query_type=search_type  # INSIGHTS, SUMMARIES, CHUNKS, GRAPH_COMPLETION
        )
        return results

    async def get_graph_data(self) -> Dict[str, Any]:
        """获取图谱数据（用于可视化）"""
        graph = await cognee.get_knowledge_graph()
        return {
            "nodes": [{"id": n.id, "label": n.name, "type": type(n).__name__}
                      for n in graph.nodes],
            "edges": [{"source": e.source_id, "target": e.target_id, "type": e.type}
                      for e in graph.edges]
        }

# cognizes/core/cognee_oceanbase.py
from typing import List, Optional
from cognee.infrastructure.databases.vector import BaseVectorStore
import pymysql

class OceanBaseVectorStore(BaseVectorStore):
    """自定义 OceanBase 向量存储适配器"""

    def __init__(self, connection_config: dict):
        self.config = connection_config
        self._connection = None

    async def add_vectors(
        self,
        vectors: List[List[float]],
        ids: List[str],
        collection: str,
        metadata: Optional[List[dict]] = None
    ) -> None:
        """添加向量到 OceanBase"""
        sql = f"""
            INSERT INTO {collection}_embeddings (id, embedding, metadata)
            VALUES (%s, %s, %s)
        """
        # 执行批量插入
        pass

    async def search(
        self,
        query_vector: List[float],
        collection: str,
        k: int = 10
    ) -> List[dict]:
        """向量相似度搜索"""
        sql = f"""
            SELECT id, chunk_text,
                   embedding <-> %s AS distance
            FROM {collection}_embeddings
            ORDER BY embedding <-> %s
            LIMIT %s
        """
        # 执行查询并返回结果
        pass

    async def delete(self, ids: List[str], collection: str) -> None:
        """删除向量"""
        sql = f"DELETE FROM {collection}_embeddings WHERE id IN %s"
        pass

# cognizes/agents/claude/entity_extractor.py
from typing import List, Dict
import anthropic

class EntityExtractor:
    """实体和关系抽取器"""

    ENTITY_TYPES = ["Paper", "Author", "Concept", "Method", "Framework", "Dataset"]
    RELATION_TYPES = ["AUTHORED_BY", "CITES", "USES_METHOD", "INTRODUCES", "EXTENDS"]

    def __init__(self):
        self.client = anthropic.Anthropic()

    async def extract(self, content: str) -> Dict:
        """从内容中抽取实体和关系"""
        prompt = f"""从以下学术内容中抽取实体和关系。

实体类型：{', '.join(self.ENTITY_TYPES)}
关系类型：{', '.join(self.RELATION_TYPES)}

内容：
{content[:4000]}

请返回 JSON 格式：
{{
    "entities": [
        {{"type": "...", "name": "...", "properties": {{...}}}}
    ],
    "relations": [
        {{"type": "...", "source": "...", "target": "...", "properties": {{...}}}}
    ]
}}
"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}]
        )
        return self._parse_response(response.content[0].text)

# cognizes/api/services/hybrid_search.py
from typing import List, Dict

def reciprocal_rank_fusion(
    results_list: List[List[Dict]],
    k: int = 60
) -> List[Dict]:
    """RRF 融合多路检索结果

    RRF Score = Σ 1/(k + rank_i) for each result list
    """
    scores = {}

    for results in results_list:
        for rank, item in enumerate(results, 1):
            doc_id = item["id"]
            if doc_id not in scores:
                scores[doc_id] = {"item": item, "score": 0}
            scores[doc_id]["score"] += 1 / (k + rank)

    # 按分数排序
    sorted_results = sorted(
        scores.values(),
        key=lambda x: x["score"],
        reverse=True
    )

    return [r["item"] for r in sorted_results]

# cognizes/api/services/search_router.py
from enum import Enum

class QueryType(Enum):
    FACTUAL = "factual"      # 事实查询 → 向量检索
    RELATIONAL = "relational" # 关系查询 → 图谱检索
    COMPLEX = "complex"       # 复杂查询 → 混合检索

class SearchRouter:
    """智能查询路由器"""

    async def classify(self, query: str) -> QueryType:
        """分类查询类型"""
        # 使用 LLM 或规则判断查询类型
        if any(kw in query for kw in ["关系", "引用", "使用了", "基于"]):
            return QueryType.RELATIONAL
        elif any(kw in query for kw in ["什么是", "定义", "解释"]):
            return QueryType.FACTUAL
        else:
            return QueryType.COMPLEX

    async def route(self, query: str) -> List[str]:
        """返回应使用的检索方法"""
        query_type = await self.classify(query)

        if query_type == QueryType.FACTUAL:
            return ["vector"]
        elif query_type == QueryType.RELATIONAL:
            return ["graph", "vector"]
        else:
            return ["keyword", "vector", "graph"]

# cognizes/api/services/hybrid_search.py
import asyncio
from typing import List, Dict

class HybridSearchService:
    """混合检索服务"""

    async def search(self, query: str, limit: int = 10) -> List[Dict]:
        """并发执行三路检索并融合结果"""

        # 并发执行三路检索
        keyword_task = self._keyword_search(query, limit)
        vector_task = self._vector_search(query, limit)
        graph_task = self._graph_search(query, limit)

        results = await asyncio.gather(
            keyword_task,
            vector_task,
            graph_task,
            return_exceptions=True
        )

        # 过滤失败的检索
        valid_results = [
            r for r in results
            if not isinstance(r, Exception) and r is not None
        ]

        # RRF 融合 (k=60 为学术推荐值 [Cormack 2009])
        return reciprocal_rank_fusion(valid_results, k=60)

    async def _keyword_search(self, query: str, limit: int) -> List[Dict]:
        """关键词检索 - OceanBase FULLTEXT"""
        sql = """
            SELECT id, title, abstract,
                   MATCH(title, abstract) AGAINST(%s) AS score
            FROM sources
            WHERE MATCH(title, abstract) AGAINST(%s IN BOOLEAN MODE)
            ORDER BY score DESC
            LIMIT %s
        """
        # 执行查询
        pass

    async def _vector_search(self, query: str, limit: int) -> List[Dict]:
        """向量检索 - OceanBase HNSW"""
        # 1. 生成查询向量
        # 2. 执行 HNSW 检索
        pass

    async def _graph_search(self, query: str, limit: int) -> List[Dict]:
        """图谱检索 - Neo4j Cypher"""
        # 执行 Cypher 查询
        pass

# cognizes/agents/claude/reasoning_agent.py
from typing import List, Dict
from enum import Enum

class ReasoningStep(Enum):
    DECOMPOSE = "decompose"    # 分解问题
    RETRIEVE = "retrieve"      # 检索信息
    REASON = "reason"          # 推理分析
    SYNTHESIZE = "synthesize"  # 综合回答
    REFLECT = "reflect"        # 反思验证

class MultiHopReasoningAgent:
    """多跳推理 Agent"""

    async def answer(self, query: str) -> Dict:
        """执行多跳推理回答问题"""

        # Step 1: 分解复杂问题
        sub_questions = await self._decompose(query)

        # Step 2: 逐个回答子问题
        sub_answers = []
        for sq in sub_questions:
            # 检索相关内容
            context = await self._retrieve(sq)
            # 生成子答案
            answer = await self._reason(sq, context)
            sub_answers.append({"question": sq, "answer": answer})

        # Step 3: 综合最终答案
        final_answer = await self._synthesize(query, sub_answers)

        # Step 4: 自我反思与验证
        reflection = await self._reflect(query, final_answer)

        if reflection["needs_revision"]:
            final_answer = await self._revise(final_answer, reflection)

        return {
            "query": query,
            "sub_questions": sub_questions,
            "sub_answers": sub_answers,
            "answer": final_answer,
            "confidence": reflection["confidence"],
            "sources": self._collect_sources(sub_answers)
        }

# cognizes/core/evaluation.py
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset

class RAGASEvaluator:
    """RAGAS 评估器"""

    def __init__(self):
        self.metrics = [
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]

    async def evaluate(self, test_cases: List[Dict]) -> Dict:
        """评估 RAG 系统质量"""

        # 构建数据集
        dataset = Dataset.from_dict({
            "question": [tc["question"] for tc in test_cases],
            "answer": [tc["answer"] for tc in test_cases],
            "contexts": [tc["contexts"] for tc in test_cases],
            "ground_truth": [tc.get("ground_truth", "") for tc in test_cases]
        })

        # 执行评估
        results = evaluate(dataset, metrics=self.metrics)

        return {
            "faithfulness": results["faithfulness"],
            "answer_relevancy": results["answer_relevancy"],
            "context_precision": results["context_precision"],
            "context_recall": results["context_recall"],
            "overall_score": sum(results.values()) / len(results)
        }

# cognizes/core/memory_manager.py
from typing import Dict, Any, Optional

class MemoryManager:
    """Agent 记忆管理器"""

    def __init__(self, user_id: str, session_id: str):
        self.user_id = user_id
        self.session_id = session_id
        self.short_term = []  # 短期记忆（内存）

    async def remember(self, content: str, memory_type: str = "short") -> None:
        """存储记忆"""
        if memory_type == "short":
            self.short_term.append({
                "content": content,
                "timestamp": datetime.now().isoformat()
            })
        elif memory_type == "long":
            await self._store_long_term(content)
        elif memory_type == "episodic":
            await self._store_episodic(content)

    async def recall(self, query: str, memory_type: str = "all") -> List[Dict]:
        """检索相关记忆"""
        results = []

        if memory_type in ["short", "all"]:
            results.extend(self._search_short_term(query))
        if memory_type in ["long", "all"]:
            results.extend(await self._search_long_term(query))
        if memory_type in ["episodic", "all"]:
            results.extend(await self._search_episodic(query))

        return results

    async def consolidate(self) -> None:
        """记忆固化：将短期记忆转移到长期记忆"""
        for memory in self.short_term:
            await self._store_long_term(memory["content"])
        self.short_term.clear()

# cognizes/api/routes/graph.py
from fastapi import APIRouter

router = APIRouter(prefix="/api/v1/graph", tags=["graph"])

@router.get("/")
async def get_graph_data(
    center_id: Optional[str] = None,
    depth: int = 2,
    limit: int = 100
):
    """获取图谱数据用于可视化"""

    if center_id:
        # 以某节点为中心展开
        query = """
        MATCH path = (center)-[*1..{depth}]-(related)
        WHERE center.id = $center_id
        RETURN path LIMIT $limit
        """
    else:
        # 获取全局概览
        query = """
        MATCH (n) OPTIONAL MATCH (n)-[r]->(m)
        RETURN n, r, m LIMIT $limit
        """

    # 执行查询并格式化为 vis-network 格式
    return {
        "nodes": [...],
        "edges": [...]
    }

// ui/src/components/KnowledgeGraph.tsx
"use client";
import { useEffect, useRef } from "react";
import { Network } from "vis-network";

interface GraphProps {
  centerId?: string;
  depth?: number;
}

export function KnowledgeGraph({ centerId, depth = 2 }: GraphProps) {
  const containerRef = useRef<HTMLDivElement>(null);

  useEffect(() => {
    const fetchAndRender = async () => {
      const response = await fetch(
        `/api/v1/graph?center_id=${centerId}&depth=${depth}`
      );
      const data = await response.json();

      if (containerRef.current) {
        new Network(containerRef.current, data, {
          nodes: {
            shape: "dot",
            scaling: { min: 10, max: 30 },
          },
          edges: {
            arrows: "to",
            smooth: { type: "curvedCW" },
          },
          physics: {
            stabilization: { iterations: 100 },
          },
        });
      }
    };

    fetchAndRender();
  }, [centerId, depth]);

  return <div ref={containerRef} className="w-full h-[600px]" />;
}

# cognizes/agents/adk/solutions_architect.py
from google.adk.agents import LlmAgent
from .tools.oceanbase_search import semantic_search
from .tools.neo4j_query import graph_query

def create_solutions_architect_agent() -> LlmAgent:
    """创建 Solutions Architect Agent"""
    return LlmAgent(
        model="gemini-2.0-flash",
        name="solutions_architect",
        description="业务场景分析与技术方案设计",
        instruction="""你是一个资深的解决方案架构师。

分析流程：
1. 理解用户业务场景和技术约束
2. 使用搜索工具检索相关技术方案和最佳实践
3. 综合分析适用性、优缺点和实施难度
4. 输出结构化的架构设计方案

输出格式：
## 1. 需求理解
## 2. 技术选型对比
## 3. 推荐架构
## 4. 实施路径
## 5. 风险与缓解""",
        tools=[semantic_search, graph_query]
    )

# cognizes/agents/adk/coordinator.py
from .solutions_architect import create_solutions_architect_agent

# 在 create_content_pipeline 中添加可选的 Solutions Architect 分支
solutions_agent = create_solutions_architect_agent()

# cognizes/api/routes/architect.py
@router.post("/api/v1/architect")
async def generate_solution(request: ArchitectRequest):
    """生成技术方案"""
    agent = create_solutions_architect_agent()
    result = await agent.run(request.scenario)
    return {"solution": result}

# cognizes/agents/adk/forum_engine.py
from google.adk.agents import LlmAgent, LoopAgent
from typing import List, Dict

class ForumEngine:
    """Agent 论坛协作引擎"""

    def __init__(self, max_rounds: int = 3):
        self.max_rounds = max_rounds
        self.host = self._create_host()
        self.participants = self._create_participants()

    def _create_host(self) -> LlmAgent:
        """创建主持人 Agent"""
        return LlmAgent(
            model="gemini-2.0-flash",
            name="forum_host",
            instruction="""你是论坛主持人，负责：
            1. 向各参与者分发讨论议题
            2. 收集各方观点
            3. 引导讨论向共识方向发展
            4. 判断是否达成共识"""
        )

    def _create_participants(self) -> List[LlmAgent]:
        """创建参与者 Agent"""
        perspectives = [
            ("advocate", "你负责支持主流观点，强调优势"),
            ("skeptic", "你负责质疑和挑战，指出潜在问题"),
            ("synthesizer", "你负责综合各方观点，寻找平衡")
        ]

        return [
            LlmAgent(
                model="gemini-2.0-flash",
                name=name,
                instruction=prompt
            )
            for name, prompt in perspectives
        ]

    async def discuss(self, topic: str) -> Dict:
        """执行论坛讨论"""
        discussion_log = []

        for round_num in range(self.max_rounds):
            # 主持人分发议题
            host_prompt = await self.host.run(topic)

            # 各参与者发言
            responses = []
            for participant in self.participants:
                response = await participant.run(host_prompt)
                responses.append({
                    "agent": participant.name,
                    "response": response
                })

            discussion_log.append({
                "round": round_num + 1,
                "responses": responses
            })

            # 检查共识
            if await self._check_consensus(responses):
                break

        # 综合最终结论
        return await self._synthesize(discussion_log)

# cognizes/api/routes/forum.py
@router.post("/api/v1/forum/discuss")
async def forum_discussion(request: ForumRequest):
    """论坛式深度讨论"""
    engine = ForumEngine(max_rounds=request.max_rounds or 3)
    result = await engine.discuss(request.topic)
    return result

-- 用户表
CREATE TABLE users (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    email VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    username VARCHAR(100),
    avatar_url VARCHAR(500),
    role ENUM('user', 'admin') DEFAULT 'user',
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- 用户收藏
CREATE TABLE user_favorites (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id BIGINT NOT NULL,
    source_id BIGINT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(id),
    FOREIGN KEY (source_id) REFERENCES sources(id),
    UNIQUE KEY unique_favorite (user_id, source_id)
);

-- 用户阅读历史
CREATE TABLE user_reading_history (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id BIGINT NOT NULL,
    source_id BIGINT NOT NULL,
    read_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    read_progress FLOAT DEFAULT 0,
    FOREIGN KEY (user_id) REFERENCES users(id),
    FOREIGN KEY (source_id) REFERENCES sources(id)
);

# 后端单元测试
cd cognizes && pytest tests/unit -v --cov=.

# 后端集成测试
cd cognizes && pytest tests/integration -v

# 前端单元测试
cd ui && npm run test

# E2E 测试
cd ui && npx playwright test

# 全量测试
npm run test:all

{
  "question": "Agentic RAG 的核心组件有哪些？",
  "answer": "系统生成的答案",
  "contexts": ["检索到的上下文片段 1", "检索到的上下文片段 2"],
  "ground_truth": "Agentic RAG 的核心组件包括：1) 智能路由器..."
}

# cognizes/evaluation/ragas_eval.py
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

async def run_ragas_evaluation(test_data: List[Dict]) -> Dict:
    """执行 RAGAS 评估"""

    # 构建评估数据集
    dataset = Dataset.from_list(test_data)

    # 运行评估
    results = evaluate(
        dataset=dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]
    )

    return {
        "faithfulness": results["faithfulness"],
        "answer_relevancy": results["answer_relevancy"],
        "context_precision": results["context_precision"],
        "context_recall": results["context_recall"],
        "passed": all([
            results["faithfulness"] > 0.85,
            results["answer_relevancy"] > 0.90,
            results["context_precision"] > 0.80,
            results["context_recall"] > 0.85
        ])
    }

// ui/tests/e2e/upload.spec.ts
import { test, expect } from "@playwright/test";

test("E2E-001 论文上传", async ({ page }) => {
  // 1. 导航到上传页面
  await page.goto("/upload");

  // 2. 上传 PDF 文件
  const fileInput = page.locator('input[type="file"]');
  await fileInput.setInputFiles("tests/fixtures/sample.pdf");

  // 3. 提交
  await page.click('button[type="submit"]');

  // 4. 等待处理完成
  await expect(page.locator('[data-testid="status"]')).toHaveText("analyzed", {
    timeout: 60000,
  });
});

// tests/performance/search.k6.js
import http from "k6/http";
import { check, sleep } from "k6";

export const options = {
  stages: [
    { duration: "1m", target: 10 }, // 预热
    { duration: "3m", target: 50 }, // 正常负载
    { duration: "1m", target: 100 }, // 峰值负载
    { duration: "1m", target: 0 }, // 冷却
  ],
  thresholds: {
    http_req_duration: ["p(95)<500"],
    http_req_failed: ["rate<0.01"],
  },
};

export default function () {
  const payload = JSON.stringify({
    query: "Agentic RAG 的核心组件",
    limit: 10,
  });

  const params = {
    headers: { "Content-Type": "application/json" },
  };

  const res = http.post("http://localhost:8000/api/v1/search", payload, params);

  check(res, {
    "status is 200": (r) => r.status === 200,
    "response time < 500ms": (r) => r.timings.duration < 500,
  });

  sleep(1);
}

# cognizes/evaluation/bleu_eval.py
import sacrebleu
import jieba

def evaluate_translation_quality(
    hypotheses: List[str],  # 系统翻译
    references: List[List[str]]  # 参考翻译（可多个）
) -> Dict:
    """评估翻译质量"""

    # 中文分词
    hyps_tokenized = [' '.join(jieba.cut(h)) for h in hypotheses]
    refs_tokenized = [
        [' '.join(jieba.cut(r)) for r in ref_set]
        for ref_set in references
    ]

    # 计算 BLEU
    bleu = sacrebleu.corpus_bleu(
        hyps_tokenized,
        refs_tokenized,
        tokenize='zh'
    )

    return {
        "bleu_score": bleu.score / 100,  # 归一化到 0-1
        "passed": bleu.score / 100 > 0.7,
        "details": {
            "brevity_penalty": bleu.bp,
            "precisions": bleu.precisions
        }
    }

# pyproject.toml 关键依赖
[project.dependencies]
fastapi = ">=0.110.0"
pydantic = ">=2.6.0"
google-adk = ">=1.0.0"
anthropic = ">=0.40.0"
openai = ">=1.40.0"
cognee = ">=0.1.17"
neo4j = ">=5.26.0"
sqlalchemy = ">=2.0.0"
ragas = ">=0.1.0"
pymysql = ">=1.1.0"
httpx = ">=0.27.0"
vis-network = ">=9.1.0"

# .env.template
# LLM Configuration
ANTHROPIC_API_KEY=sk-ant-...
OPENAI_API_KEY=sk-...

# Embedding Configuration
EMBEDDING_PROVIDER=openai
EMBEDDING_MODEL=text-embedding-3-small

# Cognee Configuration
COGNEE_LLM_PROVIDER=anthropic
COGNEE_LLM_MODEL=claude-sonnet-4-20250514

# Database Configuration
OCEANBASE_HOST=localhost
OCEANBASE_PORT=2881
OCEANBASE_USER=root
OCEANBASE_PASSWORD=
OCEANBASE_DATABASE=cognizes

# Neo4j Configuration
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=password

# Application Configuration
API_HOST=0.0.0.0
API_PORT=8000
CORS_ORIGINS=http://localhost:3000

# 开发环境启动
# 1. 启动数据库
docker compose up -d oceanbase neo4j

# 2. 启动后端
cd cognizes && uvicorn api.main:app --reload --port 8000

# 3. 启动前端
cd ui && npm run dev

# 生产环境启动
docker compose -f docker-compose.prod.yml up -d