云端 AI Agent 架构实战

from google.adk import Agent, Skill, SequentialRunner

class ExtractSkill(Skill):
    """从 API 响应中提取结构化数据"""
    def execute(self, context):
        raw_text = context.get("user_input")
        entities = self.llm.extract_entities(raw_text)
        context.set("entities", entities)
        return context

class ValidateSkill(Skill):
    """字段完整性校验——防止脏数据进入下游"""
    def execute(self, context):
        entities = context.get("entities")
        required_fields = ["name", "date", "amount"]
        missing = [f for f in required_fields if f not in entities]
        if missing:
            context.set("validation_error", f"缺少字段: {missing}")
            context.set("is_valid", False)
        else:
            context.set("is_valid", True)
        return context

class PersistSkill(Skill):
    """校验通过后写入持久化存储"""
    def execute(self, context):
        if not context.get("is_valid"):
            return context
        entities = context.get("entities")
        db.insert("records", entities)
        context.set("result", "写入成功")
        return context

agent = Agent(
    skills=[ExtractSkill(), ValidateSkill(), PersistSkill()],
    runner=SequentialRunner()
)

from google.adk import Agent, Skill, ParallelRunner, AggregatorSkill

class SearchDBSkill(Skill):
    def execute(self, context):
        result = db.query(context.get("query"))
        return {"source": "db", "data": result}

class SearchAPISkill(Skill):
    def execute(self, context):
        result = api.search(context.get("query"))
        return {"source": "api", "data": result}

class SearchCacheSkill(Skill):
    def execute(self, context):
        result = cache.get(context.get("query"))
        return {"source": "cache", "data": result}

class MergeResultsSkill(AggregatorSkill):
    def aggregate(self, results):
        valid = [r for r in results if r["data"] is not None]
        priority = {"cache": 0, "db": 1, "api": 2}
        valid.sort(key=lambda x: priority.get(x["source"], 99))
        return valid[0] if valid else {"error": "所有数据源均无结果"}

agent = Agent(
    skills=[SearchDBSkill(), SearchAPISkill(), SearchCacheSkill()],
    aggregator=MergeResultsSkill(),
    runner=ParallelRunner(timeout_seconds=10)
)

from google.adk import Agent, Skill, RouterSkill

class IntentRouter(RouterSkill):
    """根据请求意图路由到对应处理服务"""
    
    ROUTES = {
        "query": "QuerySkill",
        "create": "CreateSkill",
        "analyze": "AnalyzeSkill",
    }
    
    def route(self, context):
        user_input = context.get("user_input")
        # 方案 A：LLM 分类（适合语义复杂的场景）
        intent = self.llm.classify(
            user_input,
            categories=list(self.ROUTES.keys())
        )
        # 方案 B：规则匹配（延迟低、零 Token 消耗）
        # intent = self.rule_classify(user_input)
        
        target_skill = self.ROUTES.get(intent, "FallbackSkill")
        context.set("routed_intent", intent)
        return target_skill

from google.adk import Agent, Skill, SupervisorSkill

class TaskSupervisor(SupervisorSkill):
    MAX_RETRIES = 3
    
    def supervise(self, context):
        plan = self.llm.plan(context.get("task_description"))
        context.set("plan", plan)
        
        for step in plan["steps"]:
            skill_name = step["skill"]
            retries = 0
            
            while retries < self.MAX_RETRIES:
                result = self.dispatch(skill_name, context)
                
                quality = self.llm.evaluate(
                    task=step["description"],
                    result=result,
                    criteria=step.get("success_criteria", "完成且无错误")
                )
                
                if quality["passed"]:
                    context.set(f"step_{step['id']}_result", result)
                    break
                else:
                    retries += 1
                    context.set(f"step_{step['id']}_feedback", quality["feedback"])
                    
            if retries >= self.MAX_RETRIES:
                return {
                    "status": "failed",
                    "step": step["id"],
                    "reason": "超过最大重试次数"
                }
        
        return {"status": "completed", "results": context.get_all_results()}

# Token 预算 + 时间预算双保险
if total_tokens_used > budget_limit:
    return fallback_result()
if elapsed_time > timeout:
    return partial_result()

class SecurityExpert(Skill):
    SYSTEM_PROMPT = "你是一位安全工程师，专注于发现代码中的安全漏洞。"
    
    def execute(self, context):
        code = context.get("code_to_review")
        analysis = self.llm.analyze(
            system=self.SYSTEM_PROMPT,
            prompt=f"审查以下代码的安全性:\n{code}"
        )
        return {"expert": "security", "findings": analysis}

class PerformanceExpert(Skill):
    SYSTEM_PROMPT = "你是一位性能优化专家，专注于发现性能瓶颈和优化机会。"
    
    def execute(self, context):
        code = context.get("code_to_review")
        analysis = self.llm.analyze(
            system=self.SYSTEM_PROMPT,
            prompt=f"分析以下代码的性能:\n{code}"
        )
        return {"expert": "performance", "findings": analysis}

class ReviewModerator(AggregatorSkill):
    def aggregate(self, expert_results):
        combined = "\n".join([
            f"【{r['expert']}】{r['findings']}"
            for r in expert_results
        ])
        final = self.llm.synthesize(
            prompt=f"综合以下专家意见，给出最终代码审查报告:\n{combined}"
        )
        return final

# 路由判断用轻量模型——快且便宜
router.set_model("gemini-2.5-flash")

# 执行层用中端模型
worker.set_model("claude-sonnet-4-6")

# 关键决策走最强模型
supervisor.set_model("claude-opus-4-6")

class SmartContextManager:
    """只给每个 Skill 传它需要的最小 context"""
    def get_for_skill(self, skill_name, full_context):
        required_keys = SKILL_CONTEXT_MAP[skill_name]
        return {k: full_context[k] for k in required_keys if k in full_context}

@skill_cache(ttl=3600, key_fn=lambda ctx: hash(ctx.get("query")))
def execute(self, context):
    # 缓存命中直接返回，未命中才调 LLM
    pass