AI Agent 智能体架构深度解析：从 GPT-5.5 到企业级规模化部署

# GPT-5.5 Agent 执行示例
任务："帮我整理过去3个月的邮件，把重要邮件提取出来，生成周报"

# GPT-5.5 会自动执行以下步骤：
# 1. 调Gmail API 获取邮件
# 2. 识别重要邮件（发件人、关键词、上下文）
# 3. 按日期/主题分类整理
# 4. 生成结构化周报
# 5. 返回完整报告

# 实际耗时：10分钟完成，无需人工干预

"""
ReAct Agent 核心实现
"""
from typing import List, Dict, Any, Callable
from dataclasses import dataclass, field
from enum import Enum
import json

class AgentAction(Enum):
    """Agent可执行的动作类型"""
    SEARCH = "search"
    CALCULATE = "calculate"
    CODE_EXEC = "code_exec"
    READ_FILE = "read_file"
    CALL_API = "call_api"
    RESPOND = "respond"

@dataclass
class AgentStep:
    """Agent执行步骤"""
    thought: str
    action: AgentAction
    action_input: Any
    observation: Any = None
    success: bool = True

@dataclass
class ReActAgent:
    """ReAct推理Agent"""
    model: Any  # 大语言模型接口
    tools: Dict[str, Callable] = field(default_factory=dict)
    max_iterations: int = 10
    verbose: bool = True
    
    def reset(self):
        """重置Agent状态"""
        self.steps: List[AgentStep] = []
        self.final_answer = None
    
    def add_tool(self, name: str, func: Callable, description: str):
        """注册工具"""
        self.tools[name] = func
        self.tool_schemas[name] = {
            "name": name,
            "description": description,
            "parameters": func.__code__.co_varnames[:func.__code__.co_argcount]
        }
    
    def execute(self, task: str) -> str:
        """执行ReAct循环"""
        self.reset()
        
        for iteration in range(self.max_iterations):
            # 1. 构建Prompt，包含历史步骤
            prompt = self._build_prompt(task)
            
            # 2. 调用LLM获取下一步行动
            response = self.model.generate(prompt)
            
            # 3. 解析响应
            step = self._parse_response(response)
            if step is None:
                break
            
            # 4. 执行动作
            step.observation = self._execute_action(step.action, step.action_input)
            
            # 5. 检查是否完成
            if step.action == AgentAction.RESPOND:
                self.final_answer = step.action_input
                break
            
            self.steps.append(step)
            
            # 打印调试信息
            if self.verbose:
                self._print_step(step)
        
        return self.final_answer or "任务未完成"
    
    def _build_prompt(self, task: str) -> str:
        """构建ReAct Prompt"""
        prompt = f"""你是一个智能助手，帮助用户完成以下任务：

任务：{task}

你可以通过以下工具来完成任务：
{json.dumps(self.tool_schemas, indent=2, ensure_ascii=False)}

请按照以下格式进行推理和行动：
Thought: 描述你当前的思考
Action: 选择要执行的动作（search/calculate/code_exec/read_file/call_api/respond）
Action Input: 动作的输入参数
Observation: 执行结果

开始执行：
"""
        # 添加历史步骤
        for step in self.steps:
            prompt += f"\nThought: {step.thought}"
            prompt += f"\nAction: {step.action.value}"
            prompt += f"\nAction Input: {step.action_input}"
            prompt += f"\nObservation: {step.observation}\n"
        
        return prompt
    
    def _execute_action(self, action: AgentAction, action_input: Any) -> Any:
        """执行具体动作"""
        if action == AgentAction.SEARCH:
            return self.tools["search"](action_input)
        elif action == AgentAction.CODE_EXEC:
            return self.tools["code_exec"](action_input)
        elif action == AgentAction.RESPOND:
            return action_input
        else:
            return self.tools.get(action.value, lambda x: "未知工具")(action_input)


# 使用示例
def search_web(query: str) -> str:
    """搜索网络"""
    # 实际实现中调用搜索API
    return f"搜索结果：{query}的相关信息..."

def execute_code(code: str) -> str:
    """执行Python代码"""
    try:
        result = {}
        exec(code, result)
        return str(result.get('result', '执行完成'))
    except Exception as e:
        return f"执行错误: {e}"

# 创建Agent
agent = ReActAgent(
    model=your_llm_model,  # 替换为实际模型
    max_iterations=10,
    verbose=True
)

# 注册工具
agent.add_tool("search", search_web, "搜索网络获取信息")
agent.add_tool("code_exec", execute_code, "执行Python代码")

# 执行任务
result = agent.execute("帮我搜索2026年AI最新进展，然后写一段Python代码计算斐波那契数列")

package agent

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
)

// ActionType 定义Agent动作类型
type ActionType string

const (
	ActionSearch    ActionType = "search"
	ActionCalculate ActionType = "calculate"
	ActionCodeExec  ActionType = "code_exec"
	ActionReadFile  ActionType = "read_file"
	ActionCallAPI   ActionType = "call_api"
	ActionRespond   ActionType = "respond"
)

// AgentStep Agent执行步骤
type AgentStep struct {
	Thought      string      `json:"thought"`
	Action       ActionType  `json:"action"`
	ActionInput  interface{} `json:"action_input"`
	Observation  interface{} `json:"observation,omitempty"`
	Success      bool        `json:"success"`
}

// ToolFunc 工具函数类型
type ToolFunc func(ctx context.Context, input interface{}) (interface{}, error)

// ToolSchema 工具描述
type ToolSchema struct {
	Name        string      `json:"name"`
	Description string      `json:"description"`
	Parameters  []string    `json:"parameters"`
}

// ReActAgent ReAct推理Agent
type ReActAgent struct {
	Model         interface{}                  // LLM接口
	Tools         map[string]ToolFunc         `json:"-"`
	ToolSchemas   map[string]ToolSchema       `json:"tool_schemas"`
	MaxIterations int                          `json:"max_iterations"`
	Verbose       bool                         `json:"verbose"`
	Steps         []AgentStep                 `json:"steps"`
}

// NewReActAgent 创建新的ReAct Agent
func NewReActAgent(model interface{}, maxIterations int, verbose bool) *ReActAgent {
	return &ReActAgent{
		Model:         model,
		Tools:         make(map[string]ToolFunc),
		ToolSchemas:   make(map[string]ToolSchema),
		MaxIterations: maxIterations,
		Verbose:       verbose,
		Steps:         make([]AgentStep, 0),
	}
}

// AddTool 注册工具
func (a *ReActAgent) AddTool(name string, description string, fn ToolFunc) {
	a.Tools[name] = fn
	a.ToolSchemas[name] = ToolSchema{
		Name:        name,
		Description: description,
		Parameters:  getFuncParams(fn),
	}
}

// Reset 重置Agent状态
func (a *ReActAgent) Reset() {
	a.Steps = make([]AgentStep, 0)
}

// Execute 执行ReAct循环
func (a *ReActAgent) Execute(ctx context.Context, task string) (string, error) {
	a.Reset()

	for i := 0; i < a.MaxIterations; i++ {
		// 1. 构建Prompt
		prompt := a.buildPrompt(task)

		// 2. 调用LLM
		response, err := a.callLLM(ctx, prompt)
		if err != nil {
			return "", fmt.Errorf("LLM调用失败: %w", err)
		}

		// 3. 解析响应
		step, err := a.parseResponse(response)
		if err != nil {
			return "", fmt.Errorf("响应解析失败: %w", err)
		}

		// 4. 执行动作
		step.Observation, err = a.executeAction(ctx, step.Action, step.ActionInput)
		if err != nil {
			step.Success = false
			step.Observation = fmt.Sprintf("执行错误: %v", err)
		}

		// 5. 检查是否完成
		if step.Action == ActionRespond {
			if a.Verbose {
				fmt.Printf("Thought: %s\nAction: %s\nFinal Answer: %v\n", 
					step.Thought, step.Action, step.ActionInput)
			}
			return fmt.Sprintf("%v", step.ActionInput), nil
		}

		a.Steps = append(a.Steps, *step)

		if a.Verbose {
			a.printStep(step)
		}
	}

	return "", fmt.Errorf("达到最大迭代次数，任务未完成")
}

// buildPrompt 构建ReAct Prompt
func (a *ReActAgent) buildPrompt(task string) string {
	schemasJSON, _ := json.MarshalIndent(a.ToolSchemas, "", "  ")
	
	prompt := fmt.Sprintf(`你是一个智能助手，帮助用户完成以下任务：

任务：%s

你可以通过以下工具来完成任务：
%s

请按照以下格式进行推理和行动：
Thought: 描述你当前的思考
Action: 选择要执行的动作（search/calculate/code_exec/read_file/call_api/respond）
Action Input: 动作的输入参数
Observation: 执行结果

开始执行：
`, task, string(schemasJSON))

	// 添加历史步骤
	for _, step := range a.Steps {
		prompt += fmt.Sprintf("\nThought: %s", step.Thought)
		prompt += fmt.Sprintf("\nAction: %s", step.Action)
		prompt += fmt.Sprintf("\nAction Input: %v", step.ActionInput)
		prompt += fmt.Sprintf("\nObservation: %v\n", step.Observation)
	}

	return prompt
}

// callLLM 调用LLM（接口，需外部实现）
func (a *ReActAgent) callLLM(ctx context.Context, prompt string) (string, error) {
	// 实际实现中调用具体LLM
	// 这里使用接口抽象
	type LLMCaller interface {
		Generate(ctx context.Context, prompt string) (string, error)
	}
	
	if caller, ok := a.Model.(LLMCaller); ok {
		return caller.Generate(ctx, prompt)
	}
	
	return "", fmt.Errorf("模型未实现LLMCaller接口")
}

// parseResponse 解析LLM响应
func (a *ReActAgent) parseResponse(response string) (*AgentStep, error) {
	// 简化实现，实际需要更复杂的解析逻辑
	var step AgentStep
	
	// 解析JSON格式响应
	if err := json.Unmarshal([]byte(response), &step); err != nil {
		// 如果不是JSON，尝试文本解析
		return a.parseTextResponse(response)
	}
	
	return &step, nil
}

// parseTextResponse 解析文本格式响应
func (a *ReActAgent) parseTextResponse(response string) (*AgentStep, error) {
	step := &AgentStep{Success: true}
	
	// 简单的文本解析逻辑
	// 实际实现中需要更健壮的解析
	return step, nil
}

// executeAction 执行具体动作
func (a *ReActAgent) executeAction(ctx context.Context, action ActionType, input interface{}) (interface{}, error) {
	toolName := string(action)
	
	tool, exists := a.Tools[toolName]
	if !exists {
		return nil, fmt.Errorf("未知工具: %s", toolName)
	}
	
	return tool(ctx, input)
}

// printStep 打印步骤信息
func (a *ReActAgent) printStep(step *AgentStep) {
	fmt.Printf("Thought: %s\n", step.Thought)
	fmt.Printf("Action: %s\n", step.Action)
	fmt.Printf("Action Input: %v\n", step.ActionInput)
	fmt.Printf("Observation: %v\n\n", step.Observation)
}

// getFuncParams 获取函数参数名
func getFuncParams(fn interface{}) []string {
	v := reflect.ValueOf(fn)
	if v.Kind() != reflect.Func {
		return nil
	}
	
	t := v.Type()
	params := make([]string, t.NumIn())
	for i := 0; i < t.NumIn(); i++ {
		params[i] = t.In(i).Name()
	}
	return params
}

"""
多Agent协作系统实现
"""
import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from abc import ABC, abstractmethod
import json

class AgentType(Enum):
    """Agent类型枚举"""
    ORCHESTRATOR = "orchestrater"  # 调度Agent
    RESEARCHER = "researcher"      # 研究Agent
    CODER = "coder"                # 编码Agent
    ANALYZER = "analyzer"          # 分析Agent
    WRITER = "writer"              # 写作Agent

@dataclass
class Task:
    """任务定义"""
    id: str
    description: str
    type: AgentType
    dependencies: List[str] = field(default_factory=list)
    result: Any = None
    status: str = "pending"  # pending, running, completed, failed

class BaseAgent(ABC):
    """Agent基类"""
    
    def __init__(self, name: str, agent_type: AgentType, llm):
        self.name = name
        self.agent_type = agent_type
        self.llm = llm
        self.tools: Dict[str, callable] = {}
    
    @abstractmethod
    async def execute(self, task: Task) -> Any:
        """执行任务"""
        pass
    
    def add_tool(self, name: str, func: callable):
        """添加工具"""
        self.tools[name] = func

class OrchestratorAgent(BaseAgent):
    """调度Agent - 负责任务分解和结果聚合"""
    
    def __init__(self, name: str, llm):
        super().__init__(name, AgentType.ORCHESTRATOR, llm)
        self.sub_agents: Dict[AgentType, BaseAgent] = {}
    
    def register_agent(self, agent: BaseAgent):
        """注册子Agent"""
        self.sub_agents[agent.agent_type] = agent
    
    async def execute(self, task: Task) -> Any:
        """分解任务并协调执行"""
        
        # 1. 分析任务，确定需要的子任务
        subtasks = await self._decompose_task(task)
        
        # 2. 构建任务依赖图
        task_graph = self._build_dependency_graph(subtasks)
        
        # 3. 执行任务（考虑依赖关系）
        results = await self._execute_with_dependencies(task_graph)
        
        # 4. 聚合结果
        final_result = await self._aggregate_results(results)
        
        return final_result
    
    async def _decompose_task(self, task: Task) -> List[Task]:
        """分解任务为子任务"""
        prompt = f"""分析以下任务，将其分解为可并行执行的子任务：

任务：{task.description}

请返回JSON格式的子任务列表，包含id、描述、类型和依赖关系。
"""
        
        response = await self.llm.generate(prompt)
        # 解析响应，创建Task对象
        # 简化实现
        return []
    
    def _build_dependency_graph(self, tasks: List[Task]) -> Dict[str, List[str]]:
        """构建依赖图"""
        graph = {}
        for task in tasks:
            graph[task.id] = task.dependencies
        return graph
    
    async def _execute_with_dependencies(self, task_graph: Dict[str, List[str]]) -> Dict[str, Any]:
        """基于依赖关系执行任务"""
        results = {}
        completed = set()
        
        while len(completed) < len(task_graph):
            # 找出所有依赖已完成的待执行任务
            ready_tasks = [
                task_id for task_id, deps in task_graph.items()
                if task_id not in completed and all(d in completed for d in deps)
            ]
            
            if not ready_tasks:
                break  # 死锁检测
            
            # 并行执行就绪的任务
            task_coroutines = []
            for task_id in ready_tasks:
                task = self._get_task_by_id(task_id)
                agent_type = task.type
                agent = self.sub_agents.get(agent_type)
                if agent:
                    task_coroutines.append(agent.execute(task))
            
            # 等待所有就绪任务完成
            task_results = await asyncio.gather(*task_coroutines, return_exceptions=True)
            
            # 收集结果
            for task_id, result in zip(ready_tasks, task_results):
                results[task_id] = result
                completed.add(task_id)
        
        return results
    
    def _get_task_by_id(self, task_id: str) -> Task:
        """根据ID获取任务"""
        pass
    
    async def _aggregate_results(self, results: Dict[str, Any]) -> Any:
        """聚合子任务结果"""
        prompt = f"""聚合以下子任务结果，生成最终回答：

{json.dumps(results, indent=2, ensure_ascii=False)}

请生成综合性的回答。
"""
        return await self.llm.generate(prompt)


class ResearcherAgent(BaseAgent):
    """研究Agent - 负责信息搜索和分析"""
    
    def __init__(self, name: str, llm):
        super().__init__(name, AgentType.RESEARCHER, llm)
    
    async def execute(self, task: Task) -> Any:
        """执行研究任务"""
        
        # 1. 搜索相关信息
        search_results = await self._search(task.description)
        
        # 2. 提取关键信息
        key_info = await self._extract_info(search_results)
        
        # 3. 生成研究报告
        report = await self._generate_report(key_info)
        
        return report
    
    async def _search(self, query: str) -> List[Dict]:
        """搜索信息"""
        # 调用搜索工具
        search_func = self.tools.get("search")
        if search_func:
            return await search_func(query)
        return []
    
    async def _extract_info(self, search_results: List[Dict]) -> Dict:
        """提取关键信息"""
        prompt = f"""从以下搜索结果中提取关键信息：

{json.dumps(search_results, indent=2, ensure_ascii=False)}

请提取摘要、关键数据、重要观点。
"""
        return await self.llm.generate(prompt)
    
    async def _generate_report(self, key_info: Dict) -> str:
        """生成研究报告"""
        return f"研究报告：\n{key_info}"


class CoderAgent(BaseAgent):
    """编码Agent - 负责代码生成和调试"""
    
    def __init__(self, name: str, llm):
        super().__init__(name, AgentType.CODER, llm)
        self.code_executor = CodeExecutor()
    
    async def execute(self, task: Task) -> Any:
        """执行编码任务"""
        
        # 1. 理解需求
        requirements = await self._understand_requirements(task.description)
        
        # 2. 生成代码
        code = await self._generate_code(requirements)
        
        # 3. 执行验证
        result = await self._execute_and_verify(code)
        
        return result
    
    async def _understand_requirements(self, description: str) -> Dict:
        """理解代码需求"""
        prompt = f"""分析以下代码需求，提取技术要求：

需求：{description}

返回：
- 编程语言
- 核心功能
- 输入输出
- 约束条件
"""
        return await self.llm.generate(prompt)
    
    async def _generate_code(self, requirements: Dict) -> str:
        """生成代码"""
        prompt = f"""根据以下需求生成Python代码：

{requirements}

要求：
- 代码完整可运行
- 包含必要的错误处理
- 添加适当的注释
"""
        return await self.llm.generate(prompt)
    
    async def _execute_and_verify(self, code: str) -> Dict:
        """执行并验证代码"""
        result = await self.code_executor.execute(code)
        return result


class CodeExecutor:
    """代码执行器"""
    
    async def execute(self, code: str, language: str = "python") -> Dict:
        """执行代码"""
        try:
            if language == "python":
                # 安全执行Python代码
                result = self._safe_exec_python(code)
                return {"success": True, "output": result}
            else:
                return {"success": False, "error": f"不支持的语言: {language}"}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def _safe_exec_python(self, code: str) -> str:
        """安全执行Python代码"""
        import io
        import sys
        
        old_stdout = sys.stdout
        sys.stdout = io.StringIO()
        
        try:
            exec(code, {"__builtins__": __builtins__})
            output = sys.stdout.getvalue()
            return output if output else "代码执行完成，无输出"
        finally:
            sys.stdout = old_stdout


# 使用示例
async def main():
    """多Agent协作示例"""
    
    # 1. 初始化LLM（替换为实际LLM）
    llm = YourLLM()
    
    # 2. 创建Agent
    orchestrator = OrchestratorAgent("调度Agent", llm)
    researcher = ResearcherAgent("研究Agent", llm)
    coder = CoderAgent("编码Agent", llm)
    
    # 3. 注册子Agent
    orchestrator.register_agent(researcher)
    orchestrator.register_agent(coder)
    
    # 4. 执行复杂任务
    task = Task(
        id="task_001",
        description="搜索2026年AI最新进展，然后写一段Python代码实现机器学习模型",
        type=AgentType.ORCHESTRATOR
    )
    
    result = await orchestrator.execute(task)
    print(result)


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

"""
Function Calling 工具调用系统
"""
from typing import List, Dict, Any, Callable, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
import asyncio

@dataclass
class ToolDefinition:
    """工具定义"""
    name: str
    description: str
    parameters: Dict[str, Any]  # JSON Schema格式
    handler: Callable = field(default=None)

@dataclass
class ToolCall:
    """工具调用请求"""
    id: str
    name: str
    arguments: Dict[str, Any]

@dataclass
class ToolResult:
    """工具调用结果"""
    call_id: str
    success: bool
    result: Any = None
    error: str = None


class FunctionCallingSystem:
    """Function Calling工具调用系统"""
    
    def __init__(self, llm):
        self.llm = llm
        self.tools: Dict[str, ToolDefinition] = {}
        self.execution_history: List[Dict] = []
    
    def register_tool(self, tool: ToolDefinition):
        """注册工具"""
        self.tools[tool.name] = tool
    
    def get_tools_schema(self) -> List[Dict]:
        """获取工具的JSON Schema"""
        schemas = []
        for name, tool in self.tools.items():
            schema = {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters
                }
            }
            schemas.append(schema)
        return schemas
    
    async def process_request(self, user_message: str) -> str:
        """处理用户请求"""
        
        # 1. 调用LLM，决定是否需要调用工具
        response = await self.llm.chat(
            messages=[{"role": "user", "content": user_message}],
            tools=self.get_tools_schema()
        )
        
        # 2. 检查是否有工具调用
        if response.choices[0].message.tool_calls:
            # 执行工具调用
            tool_results = await self._execute_tool_calls(
                response.choices[0].message.tool_calls
            )
            
            # 3. 将结果返回给LLM生成最终回答
            final_response = await self.llm.chat(
                messages=[
                    {"role": "user", "content": user_message},
                    response.choices[0].message,
                    *[{"role": "tool", "tool_call_id": tr.call_id, "content": str(tr.result) if tr.success else tr.error}
                      for tr in tool_results]
                ],
                tools=self.get_tools_schema()
            )
            
            return final_response.choices[0].message.content
        
        return response.choices[0].message.content
    
    async def _execute_tool_calls(self, tool_calls) -> List[ToolResult]:
        """执行工具调用"""
        results = []
        
        # 支持并行执行
        coroutines = [self._execute_single_tool_call(tc) for tc in tool_calls]
        results = await asyncio.gather(*coroutines, return_exceptions=True)
        
        # 处理异常
        final_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                final_results.append(ToolResult(
                    call_id=tool_calls[i].id,
                    success=False,
                    error=str(result)
                ))
            else:
                final_results.append(result)
        
        return final_results
    
    async def _execute_single_tool_call(self, tool_call) -> ToolResult:
        """执行单个工具调用"""
        try:
            tool_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            
            if tool_name not in self.tools:
                return ToolResult(
                    call_id=tool_call.id,
                    success=False,
                    error=f"未知工具: {tool_name}"
                )
            
            tool = self.tools[tool_name]
            
            # 调用工具处理函数
            if asyncio.iscoroutinefunction(tool.handler):
                result = await tool.handler(**arguments)
            else:
                result = tool.handler(**arguments)
            
            return ToolResult(
                call_id=tool_call.id,
                success=True,
                result=result
            )
            
        except Exception as e:
            return ToolResult(
                call_id=tool_call.id,
                success=False,
                error=str(e)
            )


# 预定义工具示例

def search_web_tool(query: str, max_results: int = 5) -> List[Dict]:
    """搜索网络工具"""
    # 实际实现调用搜索API
    return [
        {"title": "结果1", "url": "https://...", "snippet": "..."},
        {"title": "结果2", "url": "https://...", "snippet": "..."},
    ]

def get_weather_tool(city: str, date: str = None) -> Dict:
    """获取天气工具"""
    # 实际实现调用天气API
    return {
        "city": city,
        "weather": "晴",
        "temperature": "25°C",
        "humidity": "60%"
    }

def send_email_tool(to: str, subject: str, body: str) -> Dict:
    """发送邮件工具"""
    # 实际实现调用邮件API
    return {
        "success": True,
        "message_id": "msg_123456"
    }


# 注册工具示例
def setup_tools(fc_system: FunctionCallingSystem):
    """配置工具系统"""
    
    # 搜索工具
    fc_system.register_tool(ToolDefinition(
        name="search_web",
        description="搜索互联网获取信息",
        parameters={
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                },
                "max_results": {
                    "type": "integer",
                    "description": "最大返回结果数",
                    "default": 5
                }
            },
            "required": ["query"]
        },
        handler=search_web_tool
    ))
    
    # 天气工具
    fc_system.register_tool(ToolDefinition(
        name="get_weather",
        description="获取指定城市的天气信息",
        parameters={
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称"
                },
                "date": {
                    "type": "string",
                    "description": "日期（可选），格式YYYY-MM-DD"
                }
            },
            "required": ["city"]
        },
        handler=get_weather_tool
    ))
    
    # 邮件工具
    fc_system.register_tool(ToolDefinition(
        name="send_email",
        description="发送电子邮件",
        parameters={
            "type": "object",
            "properties": {
                "to": {
                    "type": "string",
                    "description": "收件人邮箱"
                },
                "subject": {
                    "type": "string",
                    "description": "邮件主题"
                },
                "body": {
                    "type": "string",
                    "description": "邮件正文"
                }
            },
            "required": ["to", "subject", "body"]
        },
        handler=send_email_tool
    ))

"""
Agent 记忆系统实现
"""
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
import numpy as np

class MemoryType(Enum):
    """记忆类型"""
    EPISODIC = "episodic"      # 情景记忆
    SEMANTIC = "semantic"      # 语义记忆
    PROCEDURAL = "procedural"  # 程序记忆
    WORKING = "working"        # 工作记忆

@dataclass
class Memory:
    """记忆单元"""
    id: str
    content: str
    type: MemoryType
    importance: float = 0.5     # 重要性评分 0-1
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    metadata: Dict[str, Any] = field(default_factory=dict)
    embedding: np.ndarray = None

@dataclass
class ConversationTurn:
    """对话轮次"""
    user_message: str
    assistant_message: str
    timestamp: datetime
    context: Dict[str, Any] = field(default_factory=dict)


class MemoryManager:
    """记忆管理器"""
    
    def __init__(self, vector_dim: int = 1536):
        self.vector_dim = vector_dim
        self.working_memory: List[Memory] = []  # 工作记忆
        self.long_term_memories: Dict[str, Memory] = {}  # 长期记忆
        self.conversation_history: List[ConversationTurn] = []
        
        # 向量存储（简化实现，实际应使用专门的向量数据库）
        self.vectors: Dict[str, np.ndarray] = {}
        self.index: Dict[str, List[str]] = {}  # 记忆类型索引
    
    def add_working_memory(self, content: str, memory_type: MemoryType = MemoryType.WORKING,
                          importance: float = 0.5, metadata: Dict = None) -> Memory:
        """添加工作记忆"""
        memory = Memory(
            id=self._generate_id(),
            content=content,
            type=memory_type,
            importance=importance,
            metadata=metadata or {},
            embedding=self._create_embedding(content)
        )
        
        self.working_memory.append(memory)
        
        # 限制工作记忆大小
        if len(self.working_memory) > 20:
            self._prune_working_memory()
        
        return memory
    
    def store_long_term_memory(self, content: str, memory_type: MemoryType,
                               importance: float = 0.7, metadata: Dict = None) -> Memory:
        """存储长期记忆"""
        memory = Memory(
            id=self._generate_id(),
            content=content,
            type=memory_type,
            importance=importance,
            metadata=metadata or {},
            embedding=self._create_embedding(content)
        )
        
        self.long_term_memories[memory.id] = memory
        self.vectors[memory.id] = memory.embedding
        
        # 更新索引
        if memory_type.value not in self.index:
            self.index[memory_type.value] = []
        self.index[memory_type.value].append(memory.id)
        
        return memory
    
    def add_conversation_turn(self, user_message: str, assistant_message: str,
                              context: Dict = None) -> ConversationTurn:
        """添加对话轮次"""
        turn = ConversationTurn(
            user_message=user_message,
            assistant_message=assistant_message,
            timestamp=datetime.now(),
            context=context or {}
        )
        
        self.conversation_history.append(turn)
        
        # 提取关键信息存入工作记忆
        self._extract_to_working_memory(turn)
        
        return turn
    
    def retrieve_relevant_memories(self, query: str, top_k: int = 5,
                                   memory_types: List[MemoryType] = None) -> List[Tuple[Memory, float]]:
        """检索相关记忆"""
        query_embedding = self._create_embedding(query)
        
        candidates = []
        
        # 搜索工作记忆
        for memory in self.working_memory:
            if memory_types and memory.type not in memory_types:
                continue
            candidates.append(memory)
        
        # 搜索长期记忆
        for memory_id, memory in self.long_term_memories.items():
            if memory_types and memory.type not in memory_types:
                continue
            candidates.append(memory)
        
        # 计算相似度
        scored_memories = []
        for memory in candidates:
            similarity = self._cosine_similarity(query_embedding, memory.embedding)
            # 综合重要性调整
            adjusted_score = similarity * (0.5 + 0.5 * memory.importance)
            scored_memories.append((memory, adjusted_score))
        
        # 排序返回top_k
        scored_memories.sort(key=lambda x: x[1], reverse=True)
        return scored_memories[:top_k]
    
    def get_conversation_context(self, last_n: int = 5) -> str:
        """获取最近N轮对话作为上下文"""
        recent = self.conversation_history[-last_n:]
        
        context_parts = []
        for turn in recent:
            context_parts.append(f"用户: {turn.user_message}")
            context_parts.append(f"助手: {turn.assistant_message}")
        
        return "\n".join(context_parts)
    
    def build_context_prompt(self, query: str, max_memories: int = 10) -> str:
        """构建包含记忆的上下文Prompt"""
        # 检索相关记忆
        relevant = self.retrieve_relevant_memories(query, top_k=max_memories)
        
        # 获取最近对话
        recent_conversation = self.get_conversation_context(last_n=5)
        
        prompt_parts = []
        
        if recent_conversation:
            prompt_parts.append("【最近对话】")
            prompt_parts.append(recent_conversation)
        
        if relevant:
            prompt_parts.append("\n【相关记忆】")
            for memory, score in relevant:
                prompt_parts.append(f"- [{memory.type.value}] {memory.content}")
        
        return "\n".join(prompt_parts)
    
    def consolidate_memories(self):
        """记忆整合 - 将重要的工作记忆转移到长期记忆"""
        for memory in self.working_memory:
            # 检查是否需要转入长期记忆
            if memory.importance > 0.7 or memory.access_count > 3:
                # 从工作记忆移除
                self.working_memory.remove(memory)
                # 添加到长期记忆
                self.store_long_term_memory(
                    content=memory.content,
                    memory_type=memory.type,
                    importance=memory.importance * 0.8,  # 衰减
                    metadata=memory.metadata
                )
    
    def _prune_working_memory(self):
        """修剪工作记忆，保留最重要的"""
        # 按重要性和访问频率排序
        sorted_memories = sorted(
            self.working_memory,
            key=lambda m: (m.importance, m.access_count),
            reverse=True
        )
        
        # 保留前10条
        self.working_memory = sorted_memories[:10]
    
    def _extract_to_working_memory(self, turn: ConversationTurn):
        """从对话中提取关键信息"""
        # 简化实现：直接存储
        if turn.context.get("important"):
            self.add_working_memory(
                content=f"用户偏好: {turn.user_message}",
                memory_type=MemoryType.PROCEDURAL,
                importance=0.8
            )
    
    def _generate_id(self) -> str:
        """生成唯一ID"""
        import uuid
        return str(uuid.uuid4())
    
    def _create_embedding(self, text: str) -> np.ndarray:
        """创建文本嵌入（简化实现）"""
        # 实际应调用嵌入API
        # 这里使用简单的hash作为占位符
        np.random.seed(hash(text) % (2**32))
        return np.random.randn(self.vector_dim)
    
    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
        """计算余弦相似度"""
        if a is None or b is None:
            return 0.0
        dot_product = np.dot(a, b)
        norm_a = np.linalg.norm(a)
        norm_b = np.linalg.norm(b)
        if norm_a == 0 or norm_b == 0:
            return 0.0
        return float(dot_product / (norm_a * norm_b))


# 使用示例
def example_memory_usage():
    """记忆系统使用示例"""
    memory_mgr = MemoryManager(vector_dim=1536)
    
    # 1. 存储用户偏好
    memory_mgr.store_long_term_memory(
        content="用户喜欢使用Python编程，偏好简洁的代码风格",
        memory_type=MemoryType.PROCEDURAL,
        importance=0.9,
        metadata={"user": "user_001", "category": "preference"}
    )
    
    # 2. 对话交互
    memory_mgr.add_conversation_turn(
        user_message="帮我写一个快速排序算法",
        assistant_message="好的，以下是Python实现的快速排序...",
        context={"task": "coding"}
    )
    
    # 3. 检索相关记忆
    relevant = memory_mgr.retrieve_relevant_memories(
        query="用户喜欢什么编程语言？",
        top_k=3
    )
    
    # 4. 构建上下文Prompt
    context_prompt = memory_mgr.build_context_prompt(
        query="帮我优化一下排序算法",
        max_memories=5
    )
    
    print(context_prompt)

package agent

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

	"github.com/redis/go-redis/v9"
)

// AgentRequest Agent请求
type AgentRequest struct {
	ID          string                 `json:"id"`
	UserID      string                 `json:"user_id"`
	Task        string                 `json:"task"`
	Type        AgentType              `json:"type"`
	Priority    int                    `json:"priority"`
	Timeout     time.Duration          `json:"timeout"`
	Context     map[string]interface{} `json:"context"`
	CreatedAt   time.Time             `json:"created_at"`
}

// AgentResponse Agent响应
type AgentResponse struct {
	RequestID   string      `json:"request_id"`
	Success     bool        `json:"success"`
	Result      interface{} `json:"result"`
	Error       string      `json:"error,omitempty"`
	CompletedAt time.Time   `json:"completed_at"`
	Duration    time.Duration `json:"duration"`
}

// AgentWorker Agent工作器
type AgentWorker struct {
	ID           string
	Type         AgentType
	LLM          LLMClient
	Memory       *MemoryManager
	Tools        map[string]Tool
	redisClient  *redis.Client
	taskQueue    chan AgentRequest
	resultChan   chan AgentResponse
	wg           sync.WaitGroup
	ctx          context.Context
	cancel       context.CancelFunc
}

// NewAgentWorker 创建Agent工作器
func NewAgentWorker(id string, agentType AgentType, llm LLMClient) *AgentWorker {
	ctx, cancel := context.WithCancel(context.Background())
	
	return &AgentWorker{
		ID:         id,
		Type:       agentType,
		LLM:        llm,
		Memory:     NewMemoryManager(),
		Tools:      make(map[string]Tool),
		taskQueue:  make(chan AgentRequest, 100),
		resultChan: make(chan AgentResponse, 100),
		ctx:        ctx,
		cancel:     cancel,
	}
}

// RegisterTool 注册工具
func (w *AgentWorker) RegisterTool(tool Tool) {
	w.Tools[tool.Name()] = tool
}

// Start 启动工作器
func (w *AgentWorker) Start() {
	w.wg.Add(1)
	go w.run()
	log.Printf("AgentWorker %s started", w.ID)
}

// Stop 停止工作器
func (w *AgentWorker) Stop() {
	w.cancel()
	w.wg.Wait()
	log.Printf("AgentWorker %s stopped", w.ID)
}

// SubmitTask 提交任务
func (w *AgentWorker) SubmitTask(req AgentRequest) {
	w.taskQueue <- req
}

// GetResultChannel 获取结果通道
func (w *AgentWorker) GetResultChannel() <-chan AgentResponse {
	return w.resultChan
}

func (w *AgentWorker) run() {
	defer w.wg.Done()
	
	for {
		select {
		case <-w.ctx.Done():
			return
		case task := <-w.taskQueue:
			w.processTask(task)
		}
	}
}

func (w *AgentWorker) processTask(req AgentRequest) {
	startTime := time.Now()
	
	// 创建执行上下文（带超时）
	taskCtx, cancel := context.WithTimeout(w.ctx, req.Timeout)
	defer cancel()
	
	// 构建Prompt
	prompt := w.buildPrompt(req)
	
	// 执行推理
	var result interface{}
	var err error
	
	switch w.Type {
	case AgentTypeOrchestrater:
		result, err = w.executeOrchestrator(taskCtx, prompt, req)
	case AgentTypeResearcher:
		result, err = w.executeResearcher(taskCtx, prompt, req)
	case AgentTypeCoder:
		result, err = w.executeCoder(taskCtx, prompt, req)
	default:
		result, err = w.executeGeneric(taskCtx, prompt)
	}
	
	// 构建响应
	response := AgentResponse{
		RequestID:   req.ID,
		Success:     err == nil,
		Result:      result,
		CompletedAt: time.Now(),
		Duration:    time.Since(startTime),
	}
	
	if err != nil {
		response.Error = err.Error()
	}
	
	// 发送结果
	select {
	case w.resultChan <- response:
	case <-taskCtx.Done():
		log.Printf("Task %s result send timeout", req.ID)
	}
	
	// 更新记忆
	if response.Success {
		w.Memory.AddWorkingMemory(fmt.Sprintf("Task: %s, Result: %v", req.Task, result))
	}
}

func (w *AgentWorker) buildPrompt(req AgentRequest) string {
	// 获取相关记忆
	memories := w.Memory.RetrieveRelevant(req.Task, 5)
	
	prompt := fmt.Sprintf(`任务：%s

相关记忆：
`, req.Task)
	
	for _, mem := range memories {
		prompt += fmt.Sprintf("- %s\n", mem.Content)
	}
	
	if len(req.Context) > 0 {
		prompt += fmt.Sprintf("\n上下文：%+v\n", req.Context)
	}
	
	return prompt
}

func (w *AgentWorker) executeOrchestrator(ctx context.Context, prompt string, req AgentRequest) (interface{}, error) {
	// 调度逻辑
	response, err := w.LLM.Chat(ctx, prompt)
	if err != nil {
		return nil, fmt.Errorf("LLM调用失败: %w", err)
	}
	
	// 解析并执行子任务
	// ...
	
	return response, nil
}

func (w *AgentWorker) executeResearcher(ctx context.Context, prompt string, req AgentRequest) (interface{}, error) {
	// 研究逻辑
	response, err := w.LLM.Chat(ctx, prompt)
	if err != nil {
		return nil, err
	}
	
	// 提取关键信息
	// ...
	
	return response, nil
}

func (w *AgentWorker) executeCoder(ctx context.Context, prompt string, req AgentRequest) (interface{}, error) {
	// 编码逻辑
	response, err := w.LLM.Chat(ctx, prompt)
	if err != nil {
		return nil, err
	}
	
	// 代码执行验证
	// ...
	
	return response, nil
}

func (w *AgentWorker) executeGeneric(ctx context.Context, prompt string) (interface{}, error) {
	return w.LLM.Chat(ctx, prompt)
}

// AgentPool Agent工作池
type AgentPool struct {
	workers    []*AgentWorker
	taskQueue  chan AgentRequest
	resultChan chan AgentResponse
	redis      *redis.Client
	mu         sync.RWMutex
}

// NewAgentPool 创建Agent池
func NewAgentPool(redisAddr string, workerCount int, agentType AgentType, llm LLMClient) *AgentPool {
	pool := &AgentPool{
		workers:    make([]*AgentWorker, 0, workerCount),
		taskQueue:  make(chan AgentRequest, 1000),
		resultChan: make(chan AgentResponse, 1000),
		redis:      redis.NewClient(&redis.Options{Addr: redisAddr}),
	}
	
	// 创建工作器
	for i := 0; i < workerCount; i++ {
		worker := NewAgentWorker(fmt.Sprintf("%s-%d", agentType, i), agentType, llm)
		worker.Start()
		pool.workers = append(pool.workers, worker)
	}
	
	// 启动结果聚合
	go pool.aggregateResults()
	
	return pool
}

// SubmitTask 提交任务到池
func (p *AgentPool) SubmitTask(req AgentRequest) error {
	select {
	case p.taskQueue <- req:
		return nil
	default:
		return fmt.Errorf("任务队列已满")
	}
}

// GetResultChannel 获取结果通道
func (p *AgentPool) GetResultChannel() <-chan AgentResponse {
	return p.resultChan
}

func (p *AgentPool) aggregateResults() {
	for worker := range p.workers {
		go func(w *AgentWorker) {
			for result := range w.GetResultChannel() {
				select {
				case p.resultChan <- result:
				default:
					log.Printf("Result channel full, dropping result for %s", result.RequestID)
				}
			}
		}(worker)
	}
}

// Shutdown 关闭Agent池
func (p *AgentPool) Shutdown() {
	for _, worker := range p.workers {
		worker.Stop()
	}
	close(p.resultChan)
	p.redis.Close()
}