Google I/O 2026深度解读：AI Agent时代全面到来，从"大模型时代"到"智能体时代"的历史性跨越

import google.generativeai as genai
import time
from typing import Optional, Dict, Any

# 配置API密钥
genai.configure(api_key="YOUR_API_KEY")

class FastGeminiClient:
    """Gemini 3.5 Flash高速响应客户端"""
    
    def __init__(self, model_name: str = "gemini-3.5-flash"):
        self.model = genai.GenerativeModel(model_name)
        self.request_count = 0
        self.total_tokens = 0
    
    def generate_response(
        self, 
        prompt: str, 
        temperature: float = 0.7,
        max_output_tokens: int = 2048
    ) -> Dict[str, Any]:
        """
        生成响应并返回性能指标
        
        Args:
            prompt: 输入提示词
            temperature: 创造性温度（0-1）
            max_output_tokens: 最大输出token数
        
        Returns:
            包含响应内容和性能指标的字典
        """
        start_time = time.time()
        
        response = self.model.generate_content(
            contents=prompt,
            generation_config=genai.types.GenerationConfig(
                temperature=temperature,
                max_output_tokens=max_output_tokens,
            )
        )
        
        end_time = time.time()
        latency = end_time - start_time
        
        # 更新统计
        self.request_count += 1
        self.total_tokens += response.usage_metadata.total_token_count
        
        return {
            "response": response.text,
            "latency_ms": round(latency * 1000, 2),
            "input_tokens": response.usage_metadata.prompt_token_count,
            "output_tokens": response.usage_metadata.candidates_token_count,
            "total_tokens": response.usage_metadata.total_token_count
        }
    
    def batch_generate(
        self, 
        prompts: list[str], 
        concurrency: int = 5
    ) -> list[Dict[str, Any]]:
        """
        批量生成响应（支持并发）
        
        Args:
            prompts: 提示词列表
            concurrency: 并发数
        
        Returns:
            响应列表
        """
        import concurrent.futures
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
            futures = {
                executor.submit(self.generate_response, prompt): i 
                for i, prompt in enumerate(prompts)
            }
            results = [None] * len(prompts)
            
            for future in concurrent.futures.as_completed(futures):
                idx = futures[future]
                results[idx] = future.result()
        
        return results
    
    def get_stats(self) -> Dict[str, Any]:
        """获取使用统计"""
        return {
            "total_requests": self.request_count,
            "total_tokens": self.total_tokens,
            "avg_tokens_per_request": (
                self.total_tokens / self.request_count 
                if self.request_count > 0 else 0
            )
        }


# 使用示例
if __name__ == "__main__":
    client = FastGeminiClient()
    
    # 单次请求
    result = client.generate_response(
        prompt="解释一下什么是大语言模型，以及它如何处理自然语言。"
    )
    print(f"响应延迟: {result['latency_ms']}ms")
    print(f"输出Token数: {result['output_tokens']}")
    print(f"响应内容: {result['response'][:200]}...")
    
    # 批量请求
    prompts = [
        "什么是机器学习？",
        "深度学习和机器学习有什么区别？",
        "Transformer架构是什么？",
        "注意力机制是如何工作的？",
        "大模型的涌现能力是什么？"
    ]
    
    batch_results = client.batch_generate(prompts, concurrency=3)
    print(f"\n批量处理完成，共{len(batch_results)}个请求")
    print(f"统计信息: {client.get_stats()}")

class GeminiOmniCapabilities:
    """Gemini Omni核心能力展示"""
    
    # 1. 物理世界模拟
    PHYSICS_SIMULATION = {
        "gravity": True,           # 重力模拟
        "elasticity": True,        # 弹性碰撞
        "friction": True,          # 摩擦力
        "fluid_dynamics": True,    # 流体动力学
        "rigid_body": True        # 刚体碰撞
    }
    
    # 2. 多模态输入输出
    MULTIMODAL_IO = {
        "text_to_video": True,
        "image_to_video": True,
        "video_to_text": True,
        "text_to_3d": True,
        "sensor_to_prediction": True
    }
    
    # 3. 应用场景
    USE_CASES = [
        "机器人控制",      # 理解物理环境
        "自动驾驶仿真",    # 交通规则理解
        "游戏AI",         # 游戏世界模拟
        "建筑设计可视化",  # 材料物理属性
        "工业仿真"        # 生产线优化
    ]


class WorldModelSimulator:
    """世界模型模拟器"""
    
    def __init__(self):
        self.physics_engine = "OmniPhysics"
        self.reality_threshold = 0.95  # 物理真实性阈值
    
    def simulate_drop(self, object_type: str, height: float) -> dict:
        """
        模拟物体坠落
        
        Args:
            object_type: 物体类型（glass, metal, rubber, wood）
            height: 初始高度（米）
        
        Returns:
            模拟结果
        """
        # 材质物理属性表
        material_props = {
            "glass": {"elasticity": 0.3, "fragile": True, "density": 2.5},
            "metal": {"elasticity": 0.7, "fragile": False, "density": 7.8},
            "rubber": {"elasticity": 0.9, "fragile": False, "density": 1.2},
            "wood": {"elasticity": 0.4, "fragile": False, "density": 0.6}
        }
        
        props = material_props.get(object_type, material_props["metal"])
        gravity = 9.81  # 重力加速度 m/s²
        
        # 计算落地速度
        velocity = (2 * gravity * height) ** 0.5
        
        # 判断是否破碎
        will_break = props["fragile"] and velocity > 5.0
        
        return {
            "object": object_type,
            "initial_height": height,
            "impact_velocity": round(velocity, 2),
            "will_break": will_break,
            "elasticity": props["elasticity"],
            "bounce_height": round(
                height * props["elasticity"] ** 2 if not will_break else 0, 
                2
            ),
            "physics_accuracy": self.reality_threshold
        }
    
    def generate_scene_description(
        self, 
        scene_input: str
    ) -> str:
        """根据场景描述生成物理一致的动画"""
        prompt = f"""
        Generate a physically accurate simulation based on: {scene_input}
        
        Requirements:
        1. Respect gravity and basic physics
        2. Objects should interact realistically
        3. Consider material properties (glass breaks, metal bounces, etc.)
        4. Lighting should be consistent
        """
        return prompt  # 实际调用Gemini Omni API


# 使用示例
simulator = WorldModelSimulator()

test_cases = [
    ("glass", 2.0),   # 玻璃杯从2米高落下
    ("rubber", 5.0),   # 橡胶球从5米高落下
    ("metal", 3.0),    # 金属球从3米高落下
]

for obj_type, height in test_cases:
    result = simulator.simulate_drop(obj_type, height)
    print(f"{obj_type} from {height}m: ", end="")
    if result["will_break"]:
        print(f"会破碎！碰撞速度={result['impact_velocity']}m/s")
    else:
        print(f"弹起高度={result['bounce_height']}m")

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import asyncio


class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"


@dataclass
class Task:
    """任务定义"""
    id: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    dependencies: List[str] = field(default_factory=list)
    assigned_agent: Optional[str] = None
    result: Any = None
    error: Optional[str] = None
    priority: int = 0  # 0-10, 10最高
    estimated_duration: Optional[int] = None  # 秒


class PlanningAgent:
    """
    任务规划Agent - 负责任务分解与规划
    核心能力：
    1. 理解复杂目标
    2. 分解为可执行子任务
    3. 分析依赖关系
    4. 优化执行顺序
    5. 处理任务冲突
    """
    
    def __init__(self, model):
        self.model = model
        self.task_graph: Dict[str, Task] = {}
        self.max_parallel = 5  # 最大并行任务数
    
    async def plan(
        self, 
        goal: str, 
        constraints: Optional[Dict] = None
    ) -> List[Task]:
        """
        将复杂目标分解为任务列表
        
        Args:
            goal: 用户目标描述
            constraints: 约束条件（预算、时间、技能要求等）
        
        Returns:
            排序后的任务列表
        """
        constraints = constraints or {}
        
        # 使用LLM分析目标并生成任务列表
        prompt = f"""
        将以下目标分解为具体的执行任务列表：
        
        目标：{goal}
        
        约束条件：
        {self._format_constraints(constraints)}
        
        请以JSON格式返回任务列表，每个任务包含：
        - description: 任务描述
        - dependencies: 依赖的其他任务ID列表
        - estimated_duration: 预计耗时（分钟）
        - priority: 优先级(1-10)
        
        返回格式：
        {{
            "tasks": [
                {{
                    "id": "task_1",
                    "description": "...",
                    "dependencies": [],
                    "estimated_duration": 5,
                    "priority": 8
                }}
            ]
        }}
        """
        
        response = await self.model.generate_async(prompt)
        task_data = self._parse_json_response(response)
        
        # 构建任务图
        tasks = []
        for t in task_data.get("tasks", []):
            task = Task(
                id=t["id"],
                description=t["description"],
                dependencies=t.get("dependencies", []),
                priority=t.get("priority", 5),
                estimated_duration=t.get("estimated_duration")
            )
            self.task_graph[task.id] = task
            tasks.append(task)
        
        # 拓扑排序并返回
        return self._topological_sort(tasks)
    
    def _topological_sort(self, tasks: List[Task]) -> List[Task]:
        """拓扑排序 - 按依赖关系排序任务"""
        # 计算入度
        in_degree = {task.id: 0 for task in tasks}
        for task in tasks:
            for dep_id in task.dependencies:
                if dep_id in in_degree:
                    in_degree[task.id] += 1
        
        # Kahn算法
        queue = [t for t in tasks if in_degree[t.id] == 0]
        result = []
        
        while queue:
            # 按优先级排序
            queue.sort(key=lambda x: -x.priority)
            current = queue.pop(0)
            result.append(current)
            
            # 更新依赖
            for task in tasks:
                if current.id in task.dependencies:
                    in_degree[task.id] -= 1
                    if in_degree[task.id] == 0:
                        queue.append(task)
        
        return result
    
    async def execute_plan(
        self, 
        tasks: List[Task],
        executor_func: callable
    ) -> Dict[str, Any]:
        """
        执行计划
        
        Args:
            tasks: 任务列表（已排序）
            executor_func: 执行单个任务的函数
        
        Returns:
            执行结果
        """
        results = {}
        executing = []
        
        for task in tasks:
            # 等待依赖任务完成
            await self._wait_for_dependencies(task, results)
            
            # 检查是否可以并行执行
            if len(executing) >= self.max_parallel:
                completed = await self._wait_for_any(executing)
                results[completed.id] = completed
                executing = [t for t in executing if t.id != completed.id]
            
            # 启动任务执行
            task.status = TaskStatus.IN_PROGRESS
            executing.append(task)
            
            # 异步执行
            asyncio.create_task(self._execute_task(task, executor_func, results))
        
        # 等待所有任务完成
        for task in executing:
            await self._wait_for_completion(task)
            results[task.id] = task
        
        return {
            "success": all(t.status == TaskStatus.COMPLETED for t in results.values()),
            "tasks": {k: v.__dict__ for k, v in results.items()}
        }
    
    async def _execute_task(
        self, 
        task: Task, 
        executor_func: callable,
        results: Dict
    ) -> None:
        """执行单个任务"""
        try:
            # 收集依赖结果
            context = {
                dep_id: results[dep_id].result 
                for dep_id in task.dependencies 
                if dep_id in results and hasattr(results[dep_id], 'result')
            }
            
            # 执行任务
            task.result = await executor_func(task, context)
            task.status = TaskStatus.COMPLETED
            
        except Exception as e:
            task.status = TaskStatus.FAILED
            task.error = str(e)
    
    def _format_constraints(self, constraints: Dict) -> str:
        """格式化约束条件"""
        return "\n".join([f"- {k}: {v}" for k, v in constraints.items()])
    
    def _parse_json_response(self, response: str) -> Dict:
        """解析JSON响应"""
        import json
        import re
        
        # 提取JSON部分
        match = re.search(r'\{[\s\S]*\}', response)
        if match:
            return json.loads(match.group())
        return {}

import json
import asyncio
from typing import List, Dict, Any, Callable, Optional
from dataclasses import dataclass
from enum import Enum


class ToolType(Enum):
    BROWSER = "browser"           # 网页浏览
    CALCULATOR = "calculator"     # 数学计算
    CODE_EXEC = "code_executor"   # 代码执行
    API_CALL = "api_call"         # API调用
    FILE_OPS = "file_operations" # 文件操作
    SEARCH = "search"             # 搜索


@dataclass
class Tool:
    """工具定义"""
    name: str
    description: str
    tool_type: ToolType
    parameters: Dict[str, Any]
    execute: Callable
    requires_approval: bool = False


class ToolUseAgent:
    """
    工具使用Agent - 负责工具选择、调用和结果处理
    核心能力：
    1. 理解用户意图并选择合适工具
    2. 参数准备和验证
    3. 工具执行和错误处理
    4. 结果解析和格式化
    """
    
    def __init__(self, llm_model):
        self.llm = llm_model
        self.tools: Dict[str, Tool] = {}
        self.execution_history: List[Dict] = []
    
    def register_tool(self, tool: Tool) -> None:
        """注册工具"""
        self.tools[tool.name] = tool
    
    def unregister_tool(self, name: str) -> None:
        """注销工具"""
        if name in self.tools:
            del self.tools[name]
    
    async def select_tools(
        self, 
        task: str,
        context: Optional[Dict] = None
    ) -> List[Tool]:
        """
        根据任务选择合适的工具
        
        Args:
            task: 任务描述
            context: 执行上下文
        
        Returns:
            选中的工具列表
        """
        tools_desc = "\n".join([
            f"- {t.name}: {t.description} (参数: {list(t.parameters.keys())})"
            for t in self.tools.values()
        ])
        
        prompt = f"""
        分析以下任务，选择最合适的工具：
        
        任务：{task}
        
        可用工具：
        {tools_desc}
        
        执行上下文：
        {json.dumps(context or {}, ensure_ascii=False, indent=2)}
        
        请返回需要使用的工具名称列表（按调用顺序）：
        """
        
        response = await self.llm.generate_async(prompt)
        
        # 解析工具名称
        tool_names = self._parse_tool_names(response)
        return [self.tools[name] for name in tool_names if name in self.tools]
    
    async def execute_task(
        self,
        task: str,
        context: Optional[Dict] = None,
        max_retries: int = 3
    ) -> Dict[str, Any]:
        """
        执行任务（自动选择工具并调用）
        
        Args:
            task: 任务描述
            context: 执行上下文
            max_retries: 最大重试次数
        
        Returns:
            执行结果
        """
        # 1. 选择工具
        selected_tools = await self.select_tools(task, context)
        
        if not selected_tools:
            return {"status": "error", "message": "没有找到合适的工具"}
        
        # 2. 依次执行工具
        results = []
        current_context = context or {}
        
        for tool in selected_tools:
            for attempt in range(max_retries):
                try:
                    # 准备参数
                    params = await self._prepare_params(
                        tool, task, current_context
                    )
                    
                    # 执行工具
                    if asyncio.iscoroutinefunction(tool.execute):
                        result = await tool.execute(**params)
                    else:
                        result = tool.execute(**params)
                    
                    results.append({
                        "tool": tool.name,
                        "status": "success",
                        "result": result
                    })
                    
                    # 更新上下文
                    current_context[tool.name] = result
                    break
                    
                except Exception as e:
                    if attempt == max_retries - 1:
                        results.append({
                            "tool": tool.name,
                            "status": "failed",
                            "error": str(e)
                        })
                    else:
                        await asyncio.sleep(1 * (attempt + 1))  # 指数退避
        
        # 3. 汇总结果
        return self._summarize_results(results)
    
    async def _prepare_params(
        self,
        tool: Tool,
        task: str,
        context: Dict
    ) -> Dict[str, Any]:
        """准备工具参数"""
        prompt = f"""
        为以下工具准备调用参数：
        
        工具：{tool.name}
        工具描述：{tool.description}
        工具参数：{json.dumps(tool.parameters, ensure_ascii=False, indent=2)}
        
        任务：{task}
        
        当前上下文：
        {json.dumps(context, ensure_ascii=False, indent=2)}
        
        请返回JSON格式的参数：
        """
        
        response = await self.llm.generate_async(prompt)
        return self._parse_json_params(response, tool.parameters)
    
    def _parse_tool_names(self, response: str) -> List[str]:
        """解析工具名称"""
        import re
        names = re.findall(r'[\w_]+', response)
        return [n for n in names if n in self.tools]
    
    def _parse_json_params(
        self, 
        response: str, 
        param_schema: Dict
    ) -> Dict[str, Any]:
        """解析参数"""
        import json
        import re
        
        match = re.search(r'\{[\s\S]*\}', response)
        if match:
            return json.loads(match.group())
        return {}
    
    def _summarize_results(self, results: List[Dict]) -> Dict[str, Any]:
        """汇总执行结果"""
        success = all(r["status"] == "success" for r in results)
        
        return {
            "status": "success" if success else "partial",
            "success_count": sum(1 for r in results if r["status"] == "success"),
            "failed_count": sum(1 for r in results if r["status"] == "failed"),
            "results": results,
            "history": self.execution_history
        }


# 示例：定义常用工具
def create_calculator_tool() -> Tool:
    """计算器工具"""
    def execute(expression: str) -> float:
        # 安全计算器实现
        allowed_chars = set('0123456789+-*/.() ')
        if all(c in allowed_chars for c in expression):
            return eval(expression)
        raise ValueError("表达式包含非法字符")
    
    return Tool(
        name="calculator",
        description="执行数学计算表达式",
        tool_type=ToolType.CALCULATOR,
        parameters={"expression": {"type": "string", "description": "数学表达式"}},
        execute=execute
    )


def create_search_tool() -> Tool:
    """搜索工具"""
    async def execute(query: str, num_results: int = 5) -> List[Dict]:
        # 实际实现中调用搜索API
        return [
            {"title": f"结果 {i+1}", "url": f"https://example.com/{i}", "snippet": "..."}
            for i in range(num_results)
        ]
    
    return Tool(
        name="search",
        description="执行网络搜索",
        tool_type=ToolType.SEARCH,
        parameters={
            "query": {"type": "string", "description": "搜索关键词"},
            "num_results": {"type": "int", "description": "结果数量", "default": 5}
        },
        execute=execute
    )

package agent

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"cloud.google.com/go/vertexai"
	"cloud.google.com/go/vertexai/genai"
)

// MemoryType 记忆类型
type MemoryType int

const (
	MemoryTypeWorking   MemoryType = iota // 工作记忆（短期）
	MemoryTypeEpisodic                    // 情景记忆（事件）
	MemoryTypeSemantic                    // 语义记忆（知识）
	MemoryTypeProcedural                 // 程序记忆（技能）
)

// Memory 记忆单元
type Memory struct {
	ID          string                 `json:"id"`
	Type        MemoryType             `json:"type"`
	Content     string                 `json:"content"`
	Embedding   []float32              `json:"embedding,omitempty"`
	Metadata    map[string]interface{} `json:"metadata"`
	Importance  float32                `json:"importance"` // 0-1 重要性评分
	CreatedAt   time.Time             `json:"created_at"`
	AccessedAt  time.Time             `json:"accessed_at"`
	AccessCount int                   `json:"access_count"`
	ExpiresAt   *time.Time            `json:"expires_at,omitempty"`
}

// MemorySystem 记忆系统
type MemorySystem struct {
	client      *vertexai.Client
	model       *genai.GenerativeModel
	workingMem  []Memory              // 工作记忆（最近的交互）
	vectorStore map[string][]float32  // 向量存储
}

// NewMemorySystem 创建记忆系统
func NewMemorySystem(ctx context.Context, projectID, location string) (*MemorySystem, error) {
	client, err := vertexai.NewClient(ctx, option.WithCredentialsFile("credentials.json"))
	if err != nil {
		return nil, fmt.Errorf("failed to create client: %w", err)
	}

	model := client.GenerativeModel("gemini-3.5-flash")

	return &MemorySystem{
		client:      client,
		model:       model,
		workingMem:  make([]Memory, 0, 100),
		vectorStore: make(map[string][]float32),
	}, nil
}

// Store 存储记忆
func (ms *MemorySystem) Store(ctx context.Context, mem Memory) error {
	mem.ID = generateID()
	mem.CreatedAt = time.Now()
	mem.AccessedAt = mem.CreatedAt
	mem.AccessCount = 0

	// 生成嵌入向量
	embedding, err := ms.generateEmbedding(ctx, mem.Content)
	if err != nil {
		return fmt.Errorf("failed to generate embedding: %w", err)
	}
	mem.Embedding = embedding

	// 根据类型决定存储位置
	switch mem.Type {
	case MemoryTypeWorking:
		ms.workingMem = append(ms.workingMem, mem)
		// 工作记忆超过容量时，清理低重要性记忆
		if len(ms.workingMem) > 100 {
			ms.cleanupWorkingMemory()
		}
	case MemoryTypeSemantic, MemoryTypeEpisodic, MemoryTypeProcedural:
		ms.vectorStore[mem.ID] = embedding
		// 实际项目中这里会存入向量数据库
	}

	return nil
}

// Retrieve 检索记忆
func (ms *MemorySystem) Retrieve(ctx context.Context, query string, limit int) ([]Memory, error) {
	// 生成查询嵌入
	queryEmbedding, err := ms.generateEmbedding(ctx, query)
	if err != nil {
		return nil, fmt.Errorf("failed to generate query embedding: %w", err)
	}

	// 向量相似度搜索
	results := make([]Memory, 0, limit)

	// 1. 搜索工作记忆
	for i := len(ms.workingMem) - 1; i >= 0; i-- {
		mem := ms.workingMem[i]
		similarity := cosineSimilarity(queryEmbedding, mem.Embedding)
		if similarity > 0.7 {
			results = append(results, mem)
			mem.AccessedAt = time.Now()
			mem.AccessCount++
		}
	}

	// 2. 搜索长期记忆
	for id, embedding := range ms.vectorStore {
		similarity := cosineSimilarity(queryEmbedding, embedding)
		if similarity > 0.75 {
			// 实际项目中从数据库加载完整记忆
			mem := Memory{
				ID:        id,
				Embedding: embedding,
			}
			results = append(results, mem)
		}
	}

	// 按相似度排序
	sortBySimilarity(results, queryEmbedding)

	if len(results) > limit {
		results = results[:limit]
	}

	return results, nil
}

// Consolidate 记忆整合（睡眠时执行）
func (ms *MemorySystem) Consolidate(ctx context.Context) error {
	// 1. 从工作记忆中选择重要的经历
	importantMemories := make([]Memory, 0)
	for _, mem := range ms.workingMem {
		if mem.Importance > 0.6 || mem.AccessCount > 3 {
			importantMemories = append(importantMemories, mem)
		}
	}

	// 2. 生成摘要和泛化
	for _, mem := range importantMemories {
		summary, err := ms.summarizeMemory(ctx, mem)
		if err != nil {
			continue
		}

		// 创建语义记忆
		semanticMem := Memory{
			Type:       MemoryTypeSemantic,
			Content:    summary,
			Metadata:   map[string]interface{}{"source": mem.ID},
			Importance: mem.Importance,
		}

		if err := ms.Store(ctx, semanticMem); err != nil {
			continue
		}
	}

	// 3. 清理过期工作记忆
	ms.cleanupWorkingMemory()

	return nil
}

// generateEmbedding 生成嵌入向量
func (ms *MemorySystem) generateEmbedding(ctx context.Context, text string) ([]float32, error) {
	// 使用Gemini生成文本嵌入
	resp, err := ms.model.EmbedContent(ctx, &genai.EmbedContentRequest{
		Content: &genai.Content{Parts: []genai.Part{genai.Text(text)}},
		TaskType: "RETRIEVAL_DOCUMENT",
	})
	if err != nil {
		return nil, err
	}

	return resp.Embedding.Values, nil
}

// summarizeMemory 生成记忆摘要
func (ms *MemorySystem) summarizeMemory(ctx context.Context, mem Memory) (string, error) {
	prompt := fmt.Sprintf(`
请总结以下经历，提取关键信息和普遍规律：

经历：%s
类型：%v
重要性：%.2f

要求：
1. 保留关键事实和细节
2. 提炼出可泛化的知识和规律
3. 保持可回溯性
`, mem.Content, mem.Type, mem.Importance)

	resp, err := ms.model.GenerateContent(ctx, genai.Text(prompt))
	if err != nil {
		return "", err
	}

	return resp.Candidates[0].Content.Parts[0].(genai.Text), nil
}

// cleanupWorkingMemory 清理工作记忆
func (ms *MemorySystem) cleanupWorkingMemory() {
	// 按重要性和访问时间排序
	sort.Slice(ms.workingMem, func(i, j int) bool {
		if ms.workingMem[i].Importance != ms.workingMem[j].Importance {
			return ms.workingMem[i].Importance > ms.workingMem[j].Importance
		}
		return ms.workingMem[i].AccessedAt.After(ms.workingMem[j].AccessedAt)
	})

	// 保留前100个最重要的记忆
	if len(ms.workingMem) > 100 {
		ms.workingMem = ms.workingMem[:100]
	}
}

// cosineSimilarity 计算余弦相似度
func cosineSimilarity(a, b []float32) float32 {
	if len(a) != len(b) {
		return 0
	}

	var dotProduct, normA, normB float32
	for i := range a {
		dotProduct += a[i] * b[i]
		normA += a[i] * a[i]
		normB += b[i] * b[i]
	}

	if normA == 0 || normB == 0 {
		return 0
	}

	return dotProduct / (sqrt(normA) * sqrt(normB))
}

func sqrt(x float32) float32 {
	return float32(math.Sqrt(float64(x)))
}

// sortBySimilarity 按相似度排序
func sortBySimilarity(memories []Memory, queryEmbedding []float32) {
	sort.Slice(memories, func(i, j int) bool {
		simI := cosineSimilarity(memories[i].Embedding, queryEmbedding)
		simJ := cosineSimilarity(memories[j].Embedding, queryEmbedding)
		return simI > simJ
	})
}

func generateID() string {
	return fmt.Sprintf("mem_%d_%d", time.Now().UnixNano(), rand.Intn(10000))
}

import asyncio
import json
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field

# 导入之前定义的Agent组件
from planning_agent import PlanningAgent, Task
from tool_use_agent import ToolUseAgent, Tool, ToolType, create_search_tool


@dataclass
class TravelRequest:
    """旅行请求"""
    destination: str
    start_date: str
    end_date: str
    budget: float
    travelers: int
    children_ages: List[int] = field(default_factory=list)
    interests: List[str] = field(default_factory=list)
    special_requirements: str = ""


@dataclass
class TravelPlan:
    """旅行计划"""
    destination: str
    duration_days: int
    daily_itinerary: List[Dict[str, Any]]
    total_cost: float
    hotel_booking: Optional[Dict] = None
    flights: List[Dict] = field(default_factory=list)
    activities: List[Dict] = field(default_factory=list)
    packing_list: List[str] = field(default_factory=list)
    tips: List[str] = field(default_factory=list)


class TravelPlanningAgent:
    """
    智能旅行规划Agent
    整合Planning Agent和Tool Use Agent，实现端到端的旅行规划
    """
    
    def __init__(
        self, 
        llm_model,
        search_api_key: str,
        weather_api_key: str,
        booking_api_key: str
    ):
        self.planning_agent = PlanningAgent(llm_model)
        self.tool_agent = ToolUseAgent(llm_model)
        
        # 注册工具
        self._register_tools(search_api_key, weather_api_key, booking_api_key)
    
    def _register_tools(
        self, 
        search_key: str, 
        weather_key: str, 
        booking_key: str
    ):
        """注册旅行相关工具"""
        # 搜索工具
        self.tool_agent.register_tool(create_search_tool())
        
        # 天气查询工具
        self.tool_agent.register_tool(Tool(
            name="weather_query",
            description="查询指定城市未来一周的天气预报",
            tool_type=ToolType.API_CALL,
            parameters={
                "city": {"type": "string", "description": "城市名称"},
                "date": {"type": "string", "description": "查询日期"}
            },
            execute=self._query_weather
        ))
        
        # 酒店搜索工具
        self.tool_agent.register_tool(Tool(
            name="hotel_search",
            description="搜索符合条件的酒店",
            tool_type=ToolType.API_CALL,
            parameters={
                "city": {"type": "string"},
                "checkin": {"type": "string"},
                "checkout": {"type": "string"},
                "guests": {"type": "int"},
                "budget": {"type": "float"},
                "amenities": {"type": "list", "required": False}
            },
            execute=self._search_hotels
        ))
        
        # 景点搜索工具
        self.tool_agent.register_tool(Tool(
            name="attraction_search",
            description="搜索热门景点",
            tool_type=ToolType.API_CALL,
            parameters={
                "city": {"type": "string"},
                "interests": {"type": "list"},
                "family_friendly": {"type": "bool"}
            },
            execute=self._search_attractions
        ))
        
        # 交通搜索工具
        self.tool_agent.register_tool(Tool(
            name="transport_search",
            description="搜索航班或火车",
            tool_type=ToolType.API_CALL,
            parameters={
                "origin": {"type": "string"},
                "destination": {"type": "string"},
                "date": {"type": "string"},
                "transport_type": {"type": "string", "enum": ["flight", "train"]}
            },
            execute=self._search_transport
        ))
    
    async def plan_trip(self, request: TravelRequest) -> TravelPlan:
        """
        生成完整旅行计划
        
        Args:
            request: 旅行请求
        
        Returns:
            完整的旅行计划
        """
        print(f"🎯 开始规划旅行：{request.destination}")
        print(f"   日期：{request.start_date} 至 {request.end_date}")
        print(f"   预算：{request.budget}元，人数：{request.travelers}人")
        
        # 1. 收集基础信息
        context = await self._collect_information(request)
        
        # 2. 规划每日行程
        daily_plan = await self._plan_daily_itinerary(request, context)
        
        # 3. 处理预订
        bookings = await self._handle_bookings(request, context)
        
        # 4. 生成行李清单
        packing_list = await self._generate_packing_list(request, context)
        
        # 5. 生成旅行提示
        tips = await self._generate_tips(request, context)
        
        # 6. 计算总费用
        total_cost = self._calculate_total_cost(bookings, daily_plan)
        
        return TravelPlan(
            destination=request.destination,
            duration_days=(datetime.strptime(request.end_date, "%Y-%m-%d") - 
                          datetime.strptime(request.start_date, "%Y-%m-%d")).days,
            daily_itinerary=daily_plan,
            total_cost=total_cost,
            hotel_booking=bookings.get("hotel"),
            flights=bookings.get("flights", []),
            activities=bookings.get("activities", []),
            packing_list=packing_list,
            tips=tips
        )
    
    async def _collect_information(
        self, 
        request: TravelRequest
    ) -> Dict[str, Any]:
        """收集旅行相关信息"""
        context = {"request": request, "info": {}}
        
        # 并行查询多个信息源
        tasks = [
            # 天气信息
            self.tool_agent.execute_task(
                f"查询{request.destination}从{request.start_date}到{request.end_date}的天气预报",
                {"context": context}
            ),
            # 景点信息
            self.tool_agent.execute_task(
                f"搜索{request.destination}适合家庭的热门景点，兴趣：{request.interests}",
                {"context": context}
            ),
            # 酒店信息
            self.tool_agent.execute_task(
                f"搜索{request.destination}适合{int(request.travelers)}人的酒店，预算{request.budget}元/晚",
                {"context": context}
            ),
        ]
        
        # 如果有儿童，添加儿童友好活动搜索
        if request.children_ages:
            tasks.append(
                self.tool_agent.execute_task(
                    f"搜索{request.destination}适合儿童{request.children_ages}岁的活动和景点",
                    {"context": context}
                )
            )
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 整合结果
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"⚠️ 信息收集任务{i}失败: {result}")
            else:
                context["info"].update(result.get("results", []))
        
        return context
    
    async def _plan_daily_itinerary(
        self, 
        request: TravelRequest,
        context: Dict
    ) -> List[Dict[str, Any]]:
        """规划每日行程"""
        days = (datetime.strptime(request.end_date, "%Y-%m-%d") - 
                datetime.strptime(request.start_date, "%Y-%m-%d")).days
        
        daily_plan = []
        
        for day in range(days):
            current_date = datetime.strptime(request.start_date, "%Y-%m-%d") + timedelta(days=day)
            date_str = current_date.strftime("%Y-%m-%d")
            weekday = current_date.strftime("%A")
            
            # 使用Planning Agent生成当日行程
            tasks = await self.planning_agent.plan(
                goal=f"为{request.destination}的{date_str}（{weekday}）规划半日游行程",
                constraints={
                    "budget": request.budget / days,
                    "travelers": request.travelers,
                    "interests": request.interests,
                    "children_ages": request.children_ages,
                    "available_info": context["info"]
                }
            )
            
            # 执行行程规划任务
            day_plan = {
                "date": date_str,
                "weekday": weekday,
                "morning": {"activity": "", "duration": 0, "cost": 0},
                "afternoon": {"activity": "", "duration": 0, "cost": 0},
                "evening": {"activity": "", "duration": 0, "cost": 0},
                "meals": {"breakfast": "", "lunch": "", "dinner": ""},
                "daily_cost": 0
            }
            
            # 根据任务填充行程
            for task in tasks[:3]:  # 每天最多3个主要活动
                if "上午" in task.description or "morning" in task.description.lower():
                    day_plan["morning"]["activity"] = task.description
                elif "下午" in task.description or "afternoon" in task.description.lower():
                    day_plan["afternoon"]["activity"] = task.description
                elif "晚上" in task.description or "evening" in task.description.lower():
                    day_plan["evening"]["activity"] = task.description
            
            daily_plan.append(day_plan)
        
        return daily_plan
    
    async def _handle_bookings(
        self,
        request: TravelRequest,
        context: Dict
    ) -> Dict[str, Any]:
        """处理预订"""
        bookings = {}
        
        # 酒店预订
        hotel_result = await self.tool_agent.execute_task(
            f"预订{request.destination}{request.start_date}至{request.end_date}的酒店",
            {"budget": request.budget * 0.4}  # 40%预算用于酒店
        )
        bookings["hotel"] = hotel_result
        
        # 交通预订
        flights_result = await self.tool_agent.execute_task(
            f"搜索并推荐{request.start_date}前往{request.destination}的航班",
            {"transport_type": "flight"}
        )
        bookings["flights"] = flights_result.get("results", [])
        
        # 景点门票预订
        activities_result = await self.tool_agent.execute_task(
            f"预订{request.destination}热门景点的门票",
            {"tickets_needed": len(context["info"].get("attractions", []))}
        )
        bookings["activities"] = activities_result.get("results", [])
        
        return bookings
    
    def _calculate_total_cost(
        self, 
        bookings: Dict,
        daily_plan: List[Dict]
    ) -> float:
        """计算总费用"""
        total = 0
        
        # 酒店费用
        if bookings.get("hotel"):
            total += bookings["hotel"].get("price", 0)
        
        # 交通费用
        for flight in bookings.get("flights", []):
            total += flight.get("price", 0)
        
        # 每日活动费用
        for day in daily_plan:
            total += day.get("daily_cost", 0)
        
        # 餐饮估算
        total += len(daily_plan) * request.travelers * 150  # 人均150元/天
        
        return total
    
    # 以下是工具执行函数的占位实现
    def _query_weather(self, city: str, date: str) -> Dict:
        """天气查询实现"""
        return {
            "city": city,
            "date": date,
            "weather": "晴",
            "temperature": "25-32°C",
            "precipitation": "10%",
            "suggestion": "适合户外活动"
        }
    
    def _search_hotels(self, **kwargs) -> Dict:
        """酒店搜索实现"""
        return {
            "name": "推荐酒店",
            "rating": 4.5,
            "price": kwargs.get("budget", 500),
            "amenities": ["WiFi", "停车场", "儿童设施"],
            "booking_url": "https://example.com/book"
        }
    
    def _search_attractions(self, **kwargs) -> Dict:
        """景点搜索实现"""
        return {
            "attractions": [
                {"name": "热门景点A", "rating": 4.8, "ticket": 150},
                {"name": "热门景点B", "rating": 4.6, "ticket": 100},
            ]
        }
    
    def _search_transport(self, **kwargs) -> Dict:
        """交通搜索实现"""
        return {
            "flights": [
                {
                    "airline": "某航空公司",
                    "departure": "08:00",
                    "arrival": "10:30",
                    "price": 800,
                    "available": True
                }
            ]
        }
    
    async def _generate_packing_list(
        self, 
        request: TravelRequest, 
        context: Dict
    ) -> List[str]:
        """生成行李清单"""
        base_items = [
            "身份证/护照",
            "手机及充电器",
            "现金和信用卡",
            "个人洗漱用品",
            "换洗衣物"
        ]
        
        if request.children_ages:
            base_items.extend([
                "儿童常用药品",
                "儿童零食",
                "儿童娱乐用品",
                "婴儿车（如需要）"
            ])
        
        return base_items
    
    async def _generate_tips(
        self, 
        request: TravelRequest, 
        context: Dict
    ) -> List[str]:
        """生成旅行提示"""
        tips = [
            "建议提前2小时到达机场办理登机手续",
            "注意保管好个人财物",
            "关注当地天气预报，及时调整行程"
        ]
        
        if request.children_ages:
            tips.append("携带儿童的急救药品和常用药品")
        
        return tips


# 使用示例
async def main():
    # 初始化Agent
    agent = TravelPlanningAgent(
        llm_model=None,  # 实际使用时传入LLM模型
        search_api_key="your_search_key",
        weather_api_key="your_weather_key",
        booking_api_key="your_booking_key"
    )
    
    # 创建旅行请求
    request = TravelRequest(
        destination="杭州",
        start_date="2026-06-01",
        end_date="2026-06-03",
        budget=5000,
        travelers=3,
        children_ages=[5],  # 5岁小孩
        interests=["自然风光", "历史文化"],
        special_requirements="需要儿童友好的设施"
    )
    
    # 生成旅行计划
    plan = await agent.plan_trip(request)
    
    # 输出结果
    print("\n" + "="*60)
    print("📋 旅行计划已生成！")
    print("="*60)
    print(f"目的地：{plan.destination}")
    print(f"行程天数：{plan.duration_days}天")
    print(f"总费用预算：{plan.total_cost}元")
    print("\n📅 每日行程：")
    for day in plan.daily_itinerary:
        print(f"\n{day['date']} ({day['weekday']})")
        print(f"  上午：{day['morning']['activity']}")
        print(f"  下午：{day['afternoon']['activity']}")
        print(f"  晚上：{day['evening']['activity']}")
    
    print("\n📦 行李清单：")
    for item in plan.packing_list:
        print(f"  - {item}")


if __name__ == "__main__":
    asyncio.run(main())