Claude Science 深度技术解析：AI科研工作台——当"聊天框"进化为"全栈科研操作系统"

package science

import (
	"context"
	"fmt"
	"sync"
	"time"
)

// Task 表示一个科研任务
type Task struct {
	ID          string
	Description string
	Skills      []string
	Deps        []string // 依赖的Task ID
	Status      TaskStatus
	Result      *Result
	Error       error
}

type TaskStatus int

const (
	Pending TaskStatus = iota
	Running
	Completed
	Failed
)

// Result 任务执行结果
type Result struct {
	Data      interface{}
	Confidence float64
	Evidence  []string // 引证来源
	Timestamp time.Time
}

// Coordinator 协调代理
type Coordinator struct {
	skillRegistry map[string]Skill
	executor      *Executor
	auditor       *Auditor
	taskQueue     chan *Task
	results       map[string]*Result
	mu            sync.RWMutex
}

func NewCoordinator(skills []Skill, executor *Executor, auditor *Auditor) *Coordinator {
	reg := make(map[string]Skill)
	for _, s := range skills {
		reg[s.Name()] = s
	}
	return &Coordinator{
		skillRegistry: reg,
		executor:      executor,
		auditor:       auditor,
		taskQueue:     make(chan *Task, 100),
		results:       make(map[string]*Result),
	}
}

// Analyze 处理科研任务的主入口
func (c *Coordinator) Analyze(ctx context.Context, query string) (*Report, error) {
	// 1. 任务分解
	tasks, err := c.decompose(query)
	if err != nil {
		return nil, fmt.Errorf("task decomposition failed: %w", err)
	}

	// 2. 拓扑排序，确定执行顺序
	sorted, err := topologicalSort(tasks)
	if err != nil {
		return nil, fmt.Errorf("dependency resolution failed: %w", err)
	}

	// 3. 按依赖顺序执行
	for _, t := range sorted {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		default:
		}

		// 检查依赖是否完成
		allDepsReady := true
		for _, depID := range t.Deps {
			c.mu.RLock()
			_, ok := c.results[depID]
			c.mu.RUnlock()
			if !ok {
				allDepsReady = false
				break
			}
		}
		if !allDepsReady {
			return nil, fmt.Errorf("task %s: dependencies not ready", t.ID)
		}

		// 执行任务
		result, err := c.executor.Execute(ctx, t)
		if err != nil {
			t.Status = Failed
			t.Error = err
			continue
		}

		// 审核验证（异步，不阻塞主流程）
		go func(taskID string, res *Result) {
			auditResult := c.auditor.Audit(taskID, res)
			if !auditResult.Passed {
				c.mu.Lock()
				res.Confidence *= 0.5 // 未通过审核降低置信度
				res.Evidence = append(res.Evidence,
					fmt.Sprintf("审核警告: %s", auditResult.Details))
				c.mu.Unlock()
			}
		}(t.ID, result)

		c.mu.Lock()
		c.results[t.ID] = result
		t.Status = Completed
		t.Result = result
		c.mu.Unlock()
	}

	// 4. 聚合结果生成报告
	return c.synthesizeReport(tasks), nil
}

// decompose 自然语言任务分解
func (c *Coordinator) decompose(query string) ([]*Task, error) {
	// 使用Claude模型将自然语言分解为子任务
	// 返回结构化的任务列表
	// 示例实现：
	tasks := []*Task{
		{ID: "parse-seq", Description: "解析输入的蛋白质序列",
			Skills: []string{"protein-parser"}, Status: Pending},
		{ID: "struct-pred", Description: "预测蛋白质二级结构",
			Skills: []string{"structure-prediction"}, Deps: []string{"parse-seq"}, Status: Pending},
		{ID: "align-comp", Description: "与参考序列进行比对",
			Skills: []string{"sequence-alignment"}, Deps: []string{"struct-pred"}, Status: Pending},
	}
	return tasks, nil
}

// topologicalSort 拓扑排序处理任务依赖
func topologicalSort(tasks []*Task) ([]*Task, error) {
	graph := make(map[string][]string)
	for _, t := range tasks {
		graph[t.ID] = t.Deps
	}
	// Kahn算法实现
	inDegree := make(map[string]int)
	for id, deps := range graph {
		if _, ok := inDegree[id]; !ok {
			inDegree[id] = 0
		}
		for _, dep := range deps {
			inDegree[dep]++
		}
	}

	var queue []string
	for id, deg := range inDegree {
		if deg == 0 {
			queue = append(queue, id)
		}
	}

	taskMap := make(map[string]*Task)
	for _, t := range tasks {
		taskMap[t.ID] = t
	}

	var sorted []*Task
	for len(queue) > 0 {
		id := queue[0]
		queue = queue[1:]
		sorted = append(sorted, taskMap[id])
		for _, dep := range graph[id] {
			inDegree[dep]--
			if inDegree[dep] == 0 {
				queue = append(queue, dep)
			}
		}
	}

	if len(sorted) != len(tasks) {
		return nil, fmt.Errorf("cyclic dependency detected")
	}
	return sorted, nil
}

// Skill 是所有科研技能的通用接口
type Skill interface {
	Name() string
	Description() string
	Category() string
	Execute(ctx context.Context, input interface{}) (interface{}, error)
	Validate(input interface{}) error
	Version() string
}

// SequenceAnalysisSkill DNA/RNA序列分析技能
type SequenceAnalysisSkill struct {
	name     string
	version  string
	dbClient *BioDatabaseClient
}

func NewSequenceAnalysisSkill(dbClient *BioDatabaseClient) *SequenceAnalysisSkill {
	return &SequenceAnalysisSkill{
		name:     "sequence-analyzer",
		version:  "2.1.0",
		dbClient: dbClient,
	}
}

func (s *SequenceAnalysisSkill) Name() string  { return s.name }
func (s *SequenceAnalysisSkill) Category() string { return "genomics" }
func (s *SequenceAnalysisSkill) Version() string   { return s.version }

// SequenceInput 序列分析输入
type SequenceInput struct {
	Sequence   string `json:"sequence"`
	Format     string `json:"format"`     // fasta, genbank, raw
	Analysis   string `json:"analysis"`   // gc-content, codon-usage, orf-finding
	Parameters map[string]float64 `json:"parameters,omitempty"`
}

// SequenceOutput 序列分析输出
type SequenceOutput struct {
	GCContent       float64            `json:"gc_content"`
	CodonUsage      map[string]float64 `json:"codon_usage"`
	OpenReadingFrames []ORFResult      `json:"open_reading_frames"`
	QualityScore    float64            `json:"quality_score"`
}

type ORFResult struct {
	Start int    `json:"start"`
	End   int    `json:"end"`
	Length int   `json:"length"`
	Strand string `json:"strand"`
}

func (s *SequenceAnalysisSkill) Execute(ctx context.Context, input interface{}) (interface{}, error) {
	seqInput, ok := input.(*SequenceInput)
	if !ok {
		return nil, fmt.Errorf("invalid input type: expected *SequenceInput")
	}

	switch seqInput.Analysis {
	case "gc-content":
		return s.calculateGCContent(seqInput.Sequence), nil
	case "codon-usage":
		return s.analyzeCodonUsage(seqInput.Sequence), nil
	case "orf-finding":
		return s.findOpenReadingFrames(seqInput.Sequence), nil
	default:
		return nil, fmt.Errorf("unsupported analysis type: %s", seqInput.Analysis)
	}
}

func (s *SequenceAnalysisSkill) calculateGCContent(seq string) *SequenceOutput {
	gcCount := 0
	total := len(seq)
	for _, base := range strings.ToUpper(seq) {
		if base == 'G' || base == 'C' {
			gcCount++
		}
	}
	return &SequenceOutput{
		GCContent: float64(gcCount) / float64(total) * 100,
	}
}

func (s *SequenceAnalysisSkill) analyzeCodonUsage(seq string) *SequenceOutput {
	codonMap := make(map[string]float64)
	codonTable := map[string]string{
		"TTT": "Phe", "TTC": "Phe", "TTA": "Leu", "TTG": "Leu",
		"CTT": "Leu", "CTC": "Leu", "CTA": "Leu", "CTG": "Leu",
		"ATT": "Ile", "ATC": "Ile", "ATA": "Ile", "ATG": "Met",
		"GTT": "Val", "GTC": "Val", "GTA": "Val", "GTG": "Val",
		"TCT": "Ser", "TCC": "Ser", "TCA": "Ser", "TCG": "Ser",
		"CCT": "Pro", "CCC": "Pro", "CCA": "Pro", "CCG": "Pro",
		"ACT": "Thr", "ACC": "Thr", "ACA": "Thr", "ACG": "Thr",
		"GCT": "Ala", "GCC": "Ala", "GCA": "Ala", "GCG": "Ala",
		"TAT": "Tyr", "TAC": "Tyr", "TAA": "Stop", "TAG": "Stop",
		"CAT": "His", "CAC": "His", "CAA": "Gln", "CAG": "Gln",
		"AAT": "Asn", "AAC": "Asn", "AAA": "Lys", "AAG": "Lys",
		"GAT": "Asp", "GAC": "Asp", "GAA": "Glu", "GAG": "Glu",
		"TGT": "Cys", "TGC": "Cys", "TGA": "Stop", "TGG": "Trp",
		"CGT": "Arg", "CGC": "Arg", "CGA": "Arg", "CGG": "Arg",
		"AGT": "Ser", "AGC": "Ser", "AGA": "Arg", "AGG": "Arg",
		"GGT": "Gly", "GGC": "Gly", "GGA": "Gly", "GGG": "Gly",
	}
	totalCodons := len(seq) / 3
	for i := 0; i < len(seq)-2; i += 3 {
		codon := strings.ToUpper(seq[i : i+3])
		codonMap[codon]++
	}
	for codon := range codonMap {
		codonMap[codon] = codonMap[codon] / float64(totalCodons) * 100
	}
	return &SequenceOutput{CodonUsage: codonMap}
}

"""
Claude Science 代理生成器
动态生成科研子代理来处理特定任务
"""

import ast
import inspect
import textwrap
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum


class AgentCapability(Enum):
    """代理能力类型"""
    SEQUENCE_ANALYSIS = "sequence_analysis"
    STRUCTURE_PREDICTION = "structure_prediction"
    DATABASE_QUERY = "database_query"
    STATISTICAL_TEST = "statistical_test"
    VISUALIZATION = "visualization"
    CUSTOM = "custom"


@dataclass
class AgentSpec:
    """代理规格定义"""
    name: str
    description: str
    capability: AgentCapability
    inputs: List[Dict[str, Any]]
    outputs: Dict[str, Any]
    dependencies: List[str] = field(default_factory=list)
    constraints: Dict[str, Any] = field(default_factory=dict)


class AgentGenerator:
    """
    代理生成器：根据任务描述自动生成可执行科研代理
    
    核心流程：
    1. 解析任务描述 → 提取能力需求
    2. 匹配合适的代理模板 → 生成AgentSpec
    3. 动态生成Python代码 → AST级别构建
    4. 沙箱验证 → 编译执行
    5. 注册到运行时
    """
    
    def __init__(self, skill_registry: Dict[str, Any]):
        self.registry = skill_registry
        self.generated_agents: Dict[str, Any] = {}
        self.templates = self._load_templates()
        
    def _load_templates(self) -> Dict[str, str]:
        """加载代理模板库"""
        return {
            "analyzer": textwrap.dedent("""
            class {name}Agent:
                \"\"\"{description}\"\"\"
                
                def __init__(self, config: dict = None):
                    self.config = config or {{}}
                    self.capability = "{capability}"
                    
                def analyze(self, {inputs}) -> {outputs}:
                    \"\"\"执行科研分析任务\"\"\"
                    # 自动生成的代理代码
                    result = self._execute_analysis({input_names})
                    return result
                    
                def _execute_analysis(self, {input_names}):
                    \"\"\"核心分析逻辑 - 由LLM动态生成\"\"\"
                    pass
                    
                def validate(self, {inputs}) -> bool:
                    \"\"\"输入验证\"\"\"
                    return all(v is not None for v in [{input_names}])
            """),
            "pipeline": textwrap.dedent("""
            class {name}Pipeline:
                \"\"\"{description}\"\"\"
                
                def __init__(self, steps: list = None):
                    self.steps = steps or []
                    self.results = []
                    
                def add_step(self, step_fn, name: str = None):
                    self.steps.append({{
                        "fn": step_fn,
                        "name": name or f"step_{{len(self.steps)}}"
                    }})
                    
                def run(self, initial_input):
                    \"\"\"执行流水线\"\"\"
                    current = initial_input
                    for step in self.steps:
                        current = step["fn"](current)
                        self.results.append({{
                            "step": step["name"],
                            "output": current
                        }})
                    return current
            """)
        }
    
    def generate(self, task_spec: AgentSpec) -> Any:
        """
        根据AgentSpec生成代理
        
        Args:
            task_spec: 代理规格定义
            
        Returns:
            生成的代理实例
        """
        # 1. 选择合适的模板
        template_key = self._select_template(task_spec)
        template = self.templates[template_key]
        
        # 2. 填充模板
        input_names = [p["name"] for p in task_spec.inputs]
        input_params = ", ".join(
            f"{p['name']}: {p.get('type', 'Any')}" 
            for p in task_spec.inputs
        )
        
        code = template.format(
            name=task_spec.name.title(),
            description=task_spec.description,
            capability=task_spec.capability.value,
            inputs=input_params,
            outputs=task_spec.outputs.get("type", "dict"),
            input_names=", ".join(input_names)
        )
        
        # 3. AST语法树验证
        try:
            tree = ast.parse(code)
        except SyntaxError as e:
            raise ValueError(f"生成的代理代码语法错误: {e}")
        
        # 4. 动态执行并获取类
        namespace = {}
        exec(code, namespace)
        agent_class = namespace[f"{task_spec.name.title()}Agent"]
        
        # 5. 沙箱验证
        agent_instance = agent_class()
        self._sandbox_validate(agent_instance, task_spec)
        
        # 6. 注册到代理池
        self.generated_agents[task_spec.name] = agent_instance
        
        return agent_instance
    
    def _select_template(self, spec: AgentSpec) -> str:
        """根据能力类型选择模板"""
        template_map = {
            AgentCapability.SEQUENCE_ANALYSIS: "analyzer",
            AgentCapability.STRUCTURE_PREDICTION: "analyzer",
            AgentCapability.DATABASE_QUERY: "analyzer",
            AgentCapability.STATISTICAL_TEST: "pipeline",
            AgentCapability.CUSTOM: "analyzer",
        }
        return template_map.get(spec.capability, "analyzer")
    
    def _sandbox_validate(self, agent, spec: AgentSpec) -> None:
        """沙箱验证代理功能"""
        # 验证必要方法存在
        required_methods = ["analyze", "validate"]
        for method in required_methods:
            if not hasattr(agent, method):
                raise AttributeError(
                    f"生成的代理缺少必要方法: {method}"
                )
        
        # 验证输入参数签名
        sig = inspect.signature(agent.analyze)
        expected_params = {p["name"] for p in spec.inputs}
        actual_params = set(sig.parameters.keys()) - {"self"}
        
        if expected_params != actual_params:
            raise ValueError(
                f"参数签名不匹配: "
                f"期望 {expected_params}, 实际 {actual_params}"
            )


# 使用示例
if __name__ == "__main__":
    # 创建一个自定义基因表达分析代理
    spec = AgentSpec(
        name="gene-expression-analyzer",
        description="分析RNA-seq基因表达数据，计算差异表达基因",
        capability=AgentCapability.SEQUENCE_ANALYSIS,
        inputs=[
            {"name": "expression_matrix", "type": "pd.DataFrame"},
            {"name": "condition_a", "type": "List[str]"},
            {"name": "condition_b", "type": "List[str]"},
        ],
        outputs={"type": "Dict[str, Any]"},
    )
    
    generator = AgentGenerator({})
    agent = generator.generate(spec)
    
    print(f"代理已生成: {agent.__class__.__name__}")
    print(f"能力类型: {agent.capability}")

"""
Claude Science 审核代理 (Auditor Agent)
独立验证科研结果的准确性和完整性
"""

import hashlib
import re
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass, field


@dataclass
class AuditResult:
    """审核结果"""
    passed: bool
    score: float  # 0.0 - 1.0
    details: List[str] = field(default_factory=list)
    warnings: List[str] = field(default_factory=list)
    errors: List[str] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)


class CitationValidator:
    """引文验证器"""
    
    def __init__(self):
        # PubMed API端点
        self.pubmed_base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/"
        self.crossref_base = "https://api.crossref.org/works/"
        self.cache: Dict[str, bool] = {}
    
    def validate_doi(self, doi: str) -> Tuple[bool, str]:
        """验证DOI是否存在且有效"""
        doi_pattern = r'^10\.\d{4,}/[-._;()/:A-Za-z0-9]+$'
        if not re.match(doi_pattern, doi):
            return False, f"DOI格式无效: {doi}"
        # 缓存命中检查
        cache_key = f"doi:{doi}"
        if cache_key in self.cache:
            return self.cache[cache_key], "来自缓存"
        # 实际场景这里会调CrossRef API
        self.cache[cache_key] = True
        return True, "DOI已验证"
    
    def validate_pmid(self, pmid: str) -> Tuple[bool, str]:
        """验证PubMed ID"""
        if not pmid.isdigit():
            return False, f"PMID格式无效: {pmid}"
        return True, "PMID已验证"
    
    def cross_check_citation(self, text: str, citations: List[str]) -> AuditResult:
        """交叉验证引文"""
        result = AuditResult(passed=True, score=1.0)
        
        for citation in citations:
            # 提取DOI
            doi_match = re.search(r'10\.\d{4,}/[^\s,;]+', citation)
            if doi_match:
                valid, msg = self.validate_doi(doi_match.group())
                if not valid:
                    result.passed = False
                    result.score -= 0.2
                    result.errors.append(f"无效DOI: {doi_match.group()}: {msg}")
                else:
                    result.details.append(f"DOI验证通过: {doi_match.group()}")
            
            # 提取PMID
            pmid_match = re.search(r'PMID[:\s]*(\d+)', citation)
            if pmid_match:
                valid, msg = self.validate_pmid(pmid_match.group(1))
                if not valid:
                    result.warnings.append(f"PMID验证异常: {msg}")
        
        return result


class MathValidator:
    """数学计算验证器"""
    
    def verify_p_value(self, reported_p: float, 
                       test_statistic: float,
                       degrees_freedom: int,
                       test_type: str = "two-tailed") -> Tuple[bool, str]:
        """
        验证p-value计算的合理性
        
        使用近似公式验证p-value是否在合理范围内
        """
        # 基于t分布的近似验证
        import math
        
        if test_type == "two-tailed":
            # 近似：p ≈ 2 * (1 - Φ(|t|))
            # 使用标准正态近似（大样本）
            z = abs(test_statistic)
            # 近似误差函数
            def erf_approx(x):
                a1 = 0.254829592
                a2 = -0.284496736
                a3 = 1.421413741
                a4 = -1.453152027
                a5 = 1.061405429
                p = 0.3275911
                sign = 1 if x >= 0 else -1
                x = abs(x)
                t = 1.0 / (1.0 + p * x)
                y = 1.0 - (((((a5 * t + a4) * t) + a3) * t + a2) * t + a1) * t * math.exp(-x * x)
                return sign * y
            
            approx_p = 2 * (1 - erf_approx(z / math.sqrt(2)))
            
            # 检查报告值与近似值的差异
            ratio = reported_p / approx_p if approx_p > 0 else float('inf')
            
            if 0.1 < ratio < 10:
                return True, f"p-value在合理范围内 (报告值={reported_p:.4f}, 近似值={approx_p:.4f})"
            else:
                return False, f"p-value异常: 报告值={reported_p:.4f}, 近似值={approx_p:.4f}, 比值={ratio:.2f}"
        
        return True, "p-value验证通过"
    
    def check_statistical_power(self, effect_size: float,
                                 sample_size: int,
                                 alpha: float = 0.05) -> Tuple[bool, str]:
        """
        检查统计功效
        功效 < 0.8 通常被认为不足
        """
        # 简化功效计算
        import math
        d = effect_size
        n = sample_size
        # 近似：功效 ≈ Φ(|d|√n - z_{α/2})
        z_alpha = 1.96  # 双侧α=0.05
        power_approx = 0.5 * (1 + math.erf(
            (abs(d) * math.sqrt(n) - z_alpha) / math.sqrt(2)
        ))
        
        if power_approx >= 0.8:
            return True, f"统计功效充足: {power_approx:.3f}"
        else:
            return False, f"统计功效不足: {power_approx:.3f} (建议≥0.8)"


class Auditor:
    """审核代理主引擎"""
    
    def __init__(self):
        self.citation_validator = CitationValidator()
        self.math_validator = MathValidator()
        self.audit_log: List[AuditResult] = []
    
    def audit(self, task_id: str, result: Any) -> AuditResult:
        """
        对科研任务的输出进行全面审核
        
        Args:
            task_id: 任务ID
            result: 待审核的科研结果
            
        Returns:
            审核结果
        """
        combined_result = AuditResult(passed=True, score=1.0)
        
        # 1. 引文验证
        if hasattr(result, 'citations'):
            citation_result = self.citation_validator.cross_check_citation(
                str(result.data), result.citations
            )
            combined_result.errors.extend(citation_result.errors)
            combined_result.warnings.extend(citation_result.warnings)
            combined_result.details.extend(citation_result.details)
            if not citation_result.passed:
                combined_result.passed = False
                combined_result.score *= citation_result.score
        
        # 2. 计算验证
        if hasattr(result, 'statistical_tests'):
            for test in result.statistical_tests:
                valid, msg = self.math_validator.verify_p_value(
                    test.get('p_value', 0),
                    test.get('statistic', 0),
                    test.get('df', 30)
                )
                if not valid:
                    combined_result.warnings.append(msg)
                    combined_result.score -= 0.1
        
        # 3. 最终判定
        if combined_result.score < 0.5:
            combined_result.passed = False
        
        self.audit_log.append(combined_result)
        return combined_result
    
    def get_audit_summary(self) -> Dict[str, Any]:
        """获取审核摘要"""
        total = len(self.audit_log)
        passed = sum(1 for r in self.audit_log if r.passed)
        return {
            "total_audits": total,
            "passed": passed,
            "failed": total - passed,
            "pass_rate": passed / total if total > 0 else 0,
            "avg_score": sum(r.score for r in self.audit_log) / total if total > 0 else 0,
        }