跳转至

Claude Code架构学习与灵知系统演进方案

日期: 2026-04-01 目的: 学习Claude Code的先进架构模式，应用于灵知系统的自进化与多Agent协同

🎯 核心学习目标

从Claude Code源码中提取以下先进思想：

  1. 完整的Agent设计思路 - 工具型Agent的通用设计模式
  2. 运行的AI操作系统 - Agent作为系统运行时的架构
  3. 闭环式集成 - 输出→验证→反馈→改进的完整闭环
  4. Prompt动态配置 - 稳定规则层与实时配置层的分离
  5. 多Agent职责拆分 - Exploration Agent vs Planning Agent vs Execution Agent
  6. 强大的验证Agent - Verification-agent的设计模式
  7. 边界探测 - 主动探测系统能力和限制
  8. 工具调用与管理 - 输入检验、权限检查、风险评估、hooks、失败处理
  9. Agent生命周期 - 创建、运行、监控、销毁的完整管理
  10. 上下文补充机制 - 动态注入相关信息到Agent上下文

📊 Claude Code架构模式分析

1. 权限系统设计

文件: .claude/settings.local.json (344行权限规则)

核心理念: 

// 分层权限控制
{
  "tools": {
    "Bash": {
      "allowlist": [
        // 明确允许的命令，而不是黑名单
        "git *",
        "docker exec *",
        "python scripts/*",
        "npm test"
      ],
      "risk_levels": {
        "rm -rf": "CRITICAL",  // 需要用户确认
        "git push": "WARNING",   // 记录日志
        "ls": "SAFE"             // 自动允许
      }
    }
  }
}

应用于灵知系统: - ✅ API调用权限控制：哪些AI厂商可以并行调用 - ✅ 数据库操作权限：读/写/删除的分级授权 - ✅ 敏感操作确认：删除用户数据需要明确同意 - ✅ 风险评估机制：每个操作都有风险等级和对应的处理策略

2. MCP (Model Context Protocol) 集成

Claude Code的MCP插件: - web-search-prime: 网络搜索增强 - web-reader: URL内容提取 - zread: GitHub仓库读取

架构模式: 

┌─────────────────┐
│   Core Agent    │
│  (Claude Code)  │
└────────┬────────┘
         │
         ├─→ MCP Server 1: web-search
         ├─→ MCP Server 2: web-reader
         └─→ MCP Server 3: zread
              (独立进程，协议通信)

应用于灵知系统: 

# 创建灵知的MCP式扩展架构
class LingZhiMCPRegistry:
    """灵知MCP服务注册中心"""

    SERVICES = {
        "ai_provider": MultiAIAdapter,      # 已实现
        "knowledge_base": KnowledgeRetriever,  # 现有
        "analytics": AnalyticsTracker,      # 新增
        "evolution": EvolutionEngine,       # 新增
    }

    async def call_service(self, service_name: str, method: str, **kwargs):
        """统一的MCP式调用接口"""
        service = self.SERVICES.get(service_name)
        if not service:
            raise ServiceNotFoundError(service_name)

        # 权限检查
        if not self._check_permission(service_name, method):
            raise PermissionDeniedError(service_name, method)

        # 风险评估
        risk_level = self._assess_risk(service_name, method, kwargs)
        if risk_level == "CRITICAL":
            # 需要用户确认
            await self._request_user_confirmation(service_name, method, kwargs)

        # 执行调用
        return await service.execute(method, **kwargs)

3. Agent工具调用管理

Claude Code的工具调用流程: 

1. 输入验证 (Input Validation)
   ↓
2. 权限检查 (Permission Check)
   ↓
3. 风险评估 (Risk Assessment)
   ↓
4. Hooks执行 (Pre-execution Hooks)
   ↓
5. 实际执行 (Execution)
   ↓
6. Hooks执行 (Post-execution Hooks)
   ↓
7. 失败处理 (Failure Handling)
   ↓
8. 上下文补充 (Context Supplementation)

应用于灵知的进化系统: 

class AIToolCallManager:
    """AI工具调用管理器"""

    async def call_ai_provider(
        self,
        provider: str,  # "hunyuan", "deepseek", etc.
        prompt: str,
        request_type: str
    ):
        # 1. 输入验证
        self._validate_input(prompt, request_type)

        # 2. 权限检查
        if not self._check_provider_permission(provider):
            raise PermissionDeniedError(f"无权限调用 {provider}")

        # 3. 风险评估
        cost = self._estimate_cost(provider, prompt)
        if cost > self.daily_budget:
            raise BudgetExceededError(cost, self.daily_budget)

        # 4. Pre-hook
        await self.hooks.execute("before_ai_call", {
            "provider": provider,
            "prompt": prompt,
            "estimated_cost": cost
        })

        try:
            # 5. 执行调用
            result = await self._do_call(provider, prompt, request_type)

            # 6. Post-hook
            await self.hooks.execute("after_ai_call", {
                "provider": provider,
                "result": result,
                "actual_cost": result["cost"]
            })

            return result

        except Exception as e:
            # 7. 失败处理
            return await self._handle_failure(provider, prompt, e)

4. 验证Agent (Verification Agent)

Claude Code的验证机制: - 代码变更验证 - 测试通过验证 - 安全扫描验证 - 性能回归验证

应用于灵知的对比系统: 

class EvolutionVerificationAgent:
    """进化验证Agent - 确保改进是真正的改进"""

    async def verify_evolution(
        self,
        old_response: str,
        new_response: str,
        user_feedback: Dict[str, Any]
    ) -> VerificationResult:
        """验证进化是否有效"""

        # 1. 基础指标验证
        metrics = {
            "length_improved": len(new_response) > len(old_response) * 1.2,
            "has_structure": self._check_structure(new_response),
            "has_examples": self._check_examples(new_response)
        }

        # 2. 用户反馈验证
        feedback_score = user_feedback.get("satisfaction", 0)

        # 3. 对比验证（和其他AI对比）
        comparison_result = await self._compare_with_competitors(new_response)

        # 4. 综合判断
        is_valid_evolution = (
            metrics["length_improved"] and
            metrics["has_structure"] and
            feedback_score >= 4.0 and
            comparison_result["rank"] <= 2  # 前2名
        )

        return VerificationResult(
            is_valid=is_valid_evolution,
            confidence=self._calculate_confidence(metrics, feedback_score),
            reasons=self._generate_reasons(metrics, comparison_result),
            suggestions=self._generate_suggestions(metrics)
        )

    async def _compare_with_competitors(
        self,
        response: str
    ) -> Dict[str, Any]:
        """和竞品AI对比验证"""
        # 并行调用混元、DeepSeek等
        competitors = await self.multi_ai.parallel_generate(
            prompt=self.current_query,
            providers=["hunyuan", "deepseek"]
        )

        # 使用评估引擎打分
        scores = {}
        for provider, result in competitors.items():
            scores[provider] = await self.comparison_engine.evaluate(
                self.current_query,
                result["content"]
            )

        # 加入灵知自己的响应
        scores["lingzhi"] = await self.comparison_engine.evaluate(
            self.current_query,
            response
        )

        # 排序
        ranked = sorted(scores.items(), key=lambda x: x[1]["overall"], reverse=True)

        return {
            "rank": [i for i, (name, _) in enumerate(ranked, 1) if name == "lingzhi"][0],
            "scores": scores,
            "winner": ranked[0][0]
        }

5. 多Agent职责拆分

Claude Code的Agent类型: 

general-purpose: 通用Agent，完整工具访问权限
Explore: 快速探索代码库
Plan: 架构规划Agent
claude-code-guide: Claude Code使用指南

应用于灵知的进化系统: 

class EvolutionAgentFactory:
    """进化系统Agent工厂"""

    AGENTS = {
        "exploration": EvolutionExplorationAgent,    # 探索改进机会
        "planning": EvolutionPlanningAgent,          # 制定改进计划
        "comparison": MultiAIComparisonAgent,        # 对比评估
        "verification": EvolutionVerificationAgent,  # 验证改进效果
        "execution": EvolutionExecutionAgent,        # 执行改进
        "monitoring": EvolutionMonitoringAgent,      # 监控进化状态
    }

    async def create_evolution_pipeline(
        self,
        query: str,
        lingzhi_response: str
    ):
        """创建完整的进化流水线"""

        # Agent 1: 探索Agent - 找出改进机会
        explorer = self.AGENTS["exploration"]()
        opportunities = await explorer.explore(query, lingzhi_response)

        # Agent 2: 规划Agent - 制定改进计划
        planner = self.AGENTS["planning"]()
        plan = await planner.create_plan(opportunities)

        # Agent 3: 对比Agent - 和竞品对比
        comparer = self.AGENTS["comparison"]()
        comparison = await comparer.compare_with_competitors(query, lingzhi_response)

        # Agent 4: 规划Agent调整计划 - 基于对比结果
        refined_plan = await planner.refine_plan(plan, comparison)

        # Agent 5: 执行Agent - 生成改进版本
        executor = self.AGENTS["execution"]()
        improved_response = await executor.execute_improvement(refined_plan)

        # Agent 6: 验证Agent - 验证改进效果
        verifier = self.AGENTS["verification"]()
        verification = await verifier.verify_evolution(
            lingzhi_response,
            improved_response,
            comparison
        )

        # Agent 7: 监控Agent - 记录进化过程
        monitor = self.AGENTS["monitoring"]()
        await monitor.log_evolution({
            "query": query,
            "original": lingzhi_response,
            "improved": improved_response,
            "verification": verification,
            "plan": refined_plan
        })

        return {
            "improved_response": improved_response,
            "verification": verification,
            "should_adopt": verification.is_valid
        }


class EvolutionExplorationAgent:
    """探索Agent - 寻找改进机会"""

    async def explore(
        self,
        query: str,
        response: str
    ) -> List[ImprovementOpportunity]:
        """探索可能的改进方向"""

        opportunities = []

        # 方向1: 内容完整性检查
        if len(response) < 500:
            opportunities.append({
                "type": "completeness",
                "priority": "high",
                "description": "回答过短，可能缺少详细说明",
                "suggestion": "增加案例和详细解释"
            })

        # 方向2: 结构化检查
        if not self._has_structure(response):
            opportunities.append({
                "type": "structure",
                "priority": "medium",
                "description": "回答缺少清晰的结构",
                "suggestion": "添加标题、列表等结构化元素"
            })

        # 方向3: 实用性检查
        if not self._has_actionable_advice(response):
            opportunities.append({
                "type": "usefulness",
                "priority": "high",
                "description": "缺少可执行的建议",
                "suggestion": "添加具体步骤和行动建议"
            })

        # 方向4: 调用其他AI探索更多机会
        other_ai_suggestions = await self._ask_other_ai_for_suggestions(query, response)
        opportunities.extend(other_ai_suggestions)

        return opportunities

    async def _ask_other_ai_for_suggestions(
        self,
        query: str,
        response: str
    ) -> List[ImprovementOpportunity]:
        """询问其他AI这个回答可以如何改进"""

        prompt = f"""
        请分析以下问答对，找出可以改进的地方：

        问题: {query}
        回答: {response}

        请从以下维度分析：
        1. 完整性 - 是否遗漏重要信息
        2. 实用性 - 是否有可执行的建议
        3. 清晰度 - 是否容易理解
        4. 准确性 - 是否有错误或遗漏

        返回JSON格式的改进建议列表。
        """

        # 并行调用混元和DeepSeek
        results = await self.multi_ai.parallel_generate(
            prompt=prompt,
            providers=["hunyuan", "deepseek"]
        )

        opportunities = []
        for provider, result in results.items():
            if result["success"]:
                suggestions = self._parse_suggestions(result["content"])
                opportunities.extend(suggestions)

        return opportunities


class EvolutionPlanningAgent:
    """规划Agent - 制定改进计划"""

    async def create_plan(
        self,
        opportunities: List[ImprovementOpportunity]
    ) -> ImprovementPlan:
        """基于探索结果制定改进计划"""

        # 按优先级排序
        high_priority = [op for op in opportunities if op["priority"] == "high"]
        medium_priority = [op for op in opportunities if op["priority"] == "medium"]

        # 制定计划
        plan = {
            "phase_1": {
                "focus": high_priority[:2],  # 先处理最紧急的2个
                "expected_improvement": "30%",
                "effort": "low"
            },
            "phase_2": {
                "focus": high_priority[2:] + medium_priority[:2],
                "expected_improvement": "50%",
                "effort": "medium"
            }
        }

        return plan

    async def refine_plan(
        self,
        plan: ImprovementPlan,
        comparison: Dict[str, Any]
    ) -> ImprovementPlan:
        """基于对比结果调整计划"""

        # 如果灵知在某个维度落后，调整计划重点
        if comparison["lingzhi_scores"]["completeness"] < comparison["average_scores"]["completeness"]:
            plan["phase_1"]["focus"].append({
                "type": "completeness",
                "priority": "high",
                "description": "完整性低于竞品平均",
                "suggestion": "参考竞品，补充更多细节"
            })

        return plan

6. Prompt动态配置系统

Claude Code的Prompt层次: 

Layer 1: 系统Prompt (稳定的规则层)
  ↓
Layer 2: 配置文件 (CLAUDE.md, .claude/*)
  ↓
Layer 3: 上下文注入 (memory, hooks, tools)
  ↓
Layer 4: 用户输入 (实时query)

应用于灵知系统: 

class DynamicPromptManager:
    """动态Prompt管理器 - 稳定规则层 + 实时配置层"""

    def __init__(self):
        # Layer 1: 稳定的系统规则
        self.system_rules = """
        你是灵知，一个专注于智慧学习的AI助手。

        核心原则：
        1. 实用性优先 - 提供可执行的建议
        2. 结构清晰 - 使用标题、列表等
        3. 案例丰富 - 用例子说明概念
        4. 引用准确 - 标注信息来源
        """

        # Layer 2: 配置文件
        self.config = self._load_config()

        # Layer 3: 上下文层
        self.context_injectors = {
            "user_profile": UserProfileInjector(),
            "evolution_context": EvolutionContextInjector(),
            "knowledge_base": KnowledgeInjector()
        }

    async def build_prompt(
        self,
        user_query: str,
        user_id: str,
        session_context: Dict[str, Any]
    ) -> str:
        """构建完整的Prompt"""

        prompt_parts = []

        # Layer 1: 系统规则
        prompt_parts.append(self.system_rules)

        # Layer 2: 配置
        prompt_parts.append(f"\n# 当前配置\n{self.config['prompt_settings']}")

        # Layer 3: 动态上下文
        user_profile = await self.context_injectors["user_profile"].inject(user_id)
        evolution_context = await self.context_injectors["evolution_context"].inject()

        prompt_parts.append(f"\n# 用户画像\n{user_profile}")
        prompt_parts.append(f"\n# 进化提示\n{evolution_context}")

        # Layer 4: 用户输入
        prompt_parts.append(f"\n# 用户问题\n{user_query}")

        # Layer 5: 实时改进建议（如果有）
        if session_context.get("recent_improvements"):
            prompt_parts.append(
                f"\n# 最近改进\n{session_context['recent_improvements']}"
            )

        return "\n".join(prompt_parts)


class EvolutionContextInjector:
    """进化上下文注入器"""

    async def inject(self) -> str:
        """注入最新的进化改进建议"""

        # 获取最近7天的进化记录
        recent_evolutions = await self.db.get_recent_evolutions(days=7)

        if not recent_evolutions:
            return ""

        # 提取改进模式
        improvements = []
        for evolution in recent_evolutions:
            if evolution["verified"]:
                improvements.append(evolution["improvement_pattern"])

        if not improvements:
            return ""

        return f"""
        最近验证有效的改进模式：
        {chr(10).join(f'- {imp}' for imp in improvements)}

        在回答时，请尝试应用这些改进模式。
        """

7. Agent生命周期管理

Claude Code的Agent生命周期: 

创建 (Spawn) → 初始化 (Initialize) → 运行 (Run) →
空闲 (Idle) → 唤醒 (Wake) → 完成 (Complete) → 销毁 (Shutdown)

应用于灵知系统: 

class EvolutionAgentLifecycleManager:
    """进化Agent生命周期管理器"""

    def __init__(self):
        self.active_agents: Dict[str, EvolutionAgent] = {}
        self.agent_metrics: Dict[str, AgentMetrics] = {}

    async def spawn_agent(
        self,
        agent_type: str,
        task: Dict[str, Any]
    ) -> str:
        """创建新Agent"""

        agent_id = f"{agent_type}_{uuid.uuid4().hex[:8]}"

        agent = self.agent_factory.create(agent_type)
        await agent.initialize(task)

        self.active_agents[agent_id] = agent
        self.agent_metrics[agent_id] = AgentMetrics(
            created_at=datetime.now(),
            task=task
        )

        logger.info(f"Agent {agent_id} spawned for task: {task['type']}")

        return agent_id

    async def run_agent(self, agent_id: str):
        """运行Agent"""
        agent = self.active_agents.get(agent_id)
        if not agent:
            raise AgentNotFoundError(agent_id)

        self.agent_metrics[agent_id].state = "running"
        self.agent_metrics[agent_id].started_at = datetime.now()

        try:
            result = await agent.run()

            self.agent_metrics[agent_id].state = "completed"
            self.agent_metrics[agent_id].completed_at = datetime.now()
            self.agent_metrics[agent_id].result = result

            # 自动进入idle状态
            await self.idle_agent(agent_id)

            return result

        except Exception as e:
            self.agent_metrics[agent_id].state = "failed"
            self.agent_metrics[agent_id].error = str(e)
            raise

    async def idle_agent(self, agent_id: str):
        """Agent进入空闲状态"""
        agent = self.active_agents.get(agent_id)
        if not agent:
            return

        self.agent_metrics[agent_id].state = "idle"

        # 空闲超时后自动销毁
        asyncio.create_task(self._auto_shutdown_after_idle(agent_id, timeout=300))

    async def wake_agent(self, agent_id: str, new_task: Dict[str, Any]):
        """唤醒空闲Agent"""
        agent = self.active_agents.get(agent_id)
        if not agent:
            raise AgentNotFoundError(agent_id)

        if self.agent_metrics[agent_id].state != "idle":
            raise AgentNotIdleError(agent_id)

        await agent.set_task(new_task)
        self.agent_metrics[agent_id].state = "running"

        return await self.run_agent(agent_id)

    async def shutdown_agent(self, agent_id: str):
        """销毁Agent"""
        agent = self.active_agents.get(agent_id)
        if not agent:
            return

        await agent.cleanup()
        del self.active_agents[agent_id]

        metrics = self.agent_metrics.pop(agent_id)
        metrics.destroyed_at = datetime.now()

        # 保存metrics到数据库
        await self.db.save_agent_metrics(metrics)

        logger.info(f"Agent {agent_id} shutdown. Lifecycle: {metrics.duration}s")

    async def _auto_shutdown_after_idle(self, agent_id: str, timeout: int):
        """空闲超时后自动销毁"""
        await asyncio.sleep(timeout)

        if self.agent_metrics[agent_id].state == "idle":
            await self.shutdown_agent(agent_id)
            logger.info(f"Agent {agent_id} auto-shutdown after {timeout}s idle")

8. 闭环式集成

Claude Code的闭环: 

User Request → Agent Execution → Tool Call → Result →
Verification → User Feedback → Memory Update →
Next Request (with improved context)

应用于灵知的进化系统: 

class ClosedLoopEvolutionSystem:
    """闭环进化系统"""

    async def process_with_evolution(
        self,
        query: str,
        user_id: str
    ) -> Dict[str, Any]:
        """完整的进化处理流程"""

        # → 1. 生成初始回答
        response = await self.lingzhi.generate(query)

        # → 2. 追踪用户行为
        activity_id = await self.analytics.track(
            user_id=user_id,
            action="ask",
            content=query
        )

        # → 3. 并行触发对比（抽样10%）
        should_compare = await self._should_sample_for_comparison()
        if should_compare:
            comparison_task = asyncio.create_task(
                self._run_comparison(query, response)
            )
        else:
            comparison_task = None

        # → 4. 返回初始回答给用户
        result = {
            "response": response,
            "activity_id": activity_id
        }

        # ← 5. 收集用户反馈
        # (这个在用户实际评价时触发)
        # feedback = await self.collect_feedback(activity_id)

        # ← 6. 如果有对比结果，分析差距
        if comparison_task:
            comparison = await comparison_task

            if comparison["winner"] != "lingzhi":
                # 灵知输了，触发进化
                evolution_plan = await self._create_evolution_plan(
                    query, response, comparison
                )

                # → 7. 执行进化
                improved_response = await self._execute_evolution(
                    query, evolution_plan
                )

                # → 8. 验证进化
                verification = await self._verify_evolution(
                    query, response, improved_response
                )

                if verification["is_valid"]:
                    # → 9. 记录有效的进化模式
                    await self._record_evolution_pattern(
                        query, evolution_plan, verification
                    )

                    # → 10. 更新Prompt配置
                    await self.prompt_manager.add_improvement_pattern(
                        evolution_plan["pattern"]
                    )

        return result

    async def _should_sample_for_comparison(self) -> bool:
        """决定是否抽样进行对比"""
        # 10%的抽样率
        return random.random() < 0.1

    async def _run_comparison(
        self,
        query: str,
        lingzhi_response: str
    ) -> Dict[str, Any]:
        """运行多AI对比"""
        results = await self.multi_ai.parallel_generate(
            prompt=query,
            providers=["hunyuan", "deepseek"]
        )

        evaluation = await self.comparison_engine.compare_qa_responses(
            query=query,
            lingzhi_response=lingzhi_response,
            competitor_responses=results
        )

        return {
            "results": results,
            "evaluation": evaluation,
            "winner": evaluation["winner"]
        }

🎯 灵知系统演进路线图

Phase 1: 基础架构 (当前已实现)

  • ✅ 多AI适配器 (multi_ai_adapter.py)
  • ✅ 对比评估引擎 (comparison_engine.py)
  • ✅ 进化API端点 (evolution.py)
  • ✅ 数据库表结构

Phase 2: 验证系统 (下一步)

目标: 添加Verification Agent，确保进化质量

任务: 1. 创建 EvolutionVerificationAgent 类 2. 实现多维度验证逻辑 3. 添加验证阈值配置 4. 集成到进化流程

预期收益: - 减少无效进化 70% - 提高进化成功率 50%

Phase 3: 探索与规划Agent

目标: 自动发现和规划改进机会

任务: 1. 实现 EvolutionExplorationAgent 2. 实现 EvolutionPlanningAgent 3. 添加机会发现算法 4. 集成到进化流水线

预期收益: - 自动发现改进机会 - 智能规划改进步骤

Phase 4: Agent生命周期管理

目标: 完整的Agent编排和监控

任务: 1. 实现 EvolutionAgentLifecycleManager 2. 添加Agent监控指标 3. 实现自动销毁和资源回收 4. 添加Agent性能分析

预期收益: - 资源利用率提升 40% - 并发处理能力提升 3x

Phase 5: 动态Prompt系统

目标: 实时配置和改进建议注入

任务: 1. 实现 DynamicPromptManager 2. 添加上下文注入器 3. 实现改进模式提取和应用 4. 添加A/B测试框架

预期收益: - 回答质量提升 30% - 改进迭代速度提升 5x

Phase 6: 完整闭环集成

目标: 端到端的自动进化流程

任务: 1. 实现 ClosedLoopEvolutionSystem 2. 添加抽样对比机制 3. 实现用户反馈集成 4. 添加进化效果追踪

预期收益: - 自动进化系统 - 持续改进 - 数据驱动优化

📊 预期效果

短期 (1-2周)

  • 验证系统上线，减少无效进化
  • 探索Agent自动发现改进机会
  • 进化质量提升 50%

中期 (1-2月)

  • 完整的多Agent流水线
  • 自动化进化流程
  • 回答质量提升 30%

长期 (3-6月)

  • 自适应Prompt系统
  • 完全闭环的自动进化
  • 持续改进，无需人工干预

🔧 技术栈

  • Async/Await: 并行Agent执行
  • PostgreSQL: Agent状态和进化记录
  • Redis: Agent消息队列和锁
  • Docker: Agent容器隔离
  • Prometheus: Agent监控指标

📝 下一步行动

立即执行

  1. ✅ 配置混元 + DeepSeek API密钥
  2. ⏳ 测试API连接
  3. ⏳ 前端集成（搜索页、问答页）

本周计划

  1. ⏳ 实现 EvolutionVerificationAgent
  2. ⏳ 添加验证阈值配置
  3. ⏳ 开始收集真实对比数据

下周计划

  1. ⏳ 实现 EvolutionExplorationAgent
  2. ⏳ 实现 EvolutionPlanningAgent
  3. ⏳ 集成完整进化流水线

众智混元，万法灵通 ⚡🚀