Dify智能体模块：AI Agent架构与多策略推理引擎

graph TB
    subgraph "Agent策略层"
        FC[Function Calling 函数调用]
        COT[Chain of Thought 思维链]
        ReACT[ReACT 推理行动]
    end
    
    subgraph "Agent运行器层"
        BaseRunner[BaseAgentRunner 基础运行器]
        FCRunner[FunctionCallAgentRunner]
        CotRunner[CotAgentRunner 抽象基类]
        CotChatRunner[CotChatAgentRunner]
        CotCompletionRunner[CotCompletionAgentRunner]
    end
    
    subgraph "工具系统"
        ToolManager[ToolManager 工具管理器]
        BuiltinTools[内置工具]
        APITools[API工具]
        DatasetTools[数据集工具]
        WorkflowTools[工作流工具]
    end
    
    subgraph "提示系统"
        AgentPrompt[Agent提示模板]
        ReACTTemplate[ReACT模板]
        FCTemplate[函数调用模板]
        HistoryManager[历史管理器]
    end
    
    subgraph "执行引擎"
        Executor[推理执行器]
        IterationControl[迭代控制]
        StateManager[状态管理器]
        CallbackHandler[回调处理器]
    end
    
    FC --> FCRunner
    COT --> CotRunner
    ReACT --> CotChatRunner
    
    BaseRunner --> FCRunner
    BaseRunner --> CotRunner
    CotRunner --> CotChatRunner
    CotRunner --> CotCompletionRunner
    
    FCRunner --> ToolManager
    CotChatRunner --> BuiltinTools
    
    AgentPrompt --> ReACTTemplate
    ReACTTemplate --> FCTemplate
    
    FCRunner --> Executor
    Executor --> IterationControl
    IterationControl --> StateManager
    
    style FC fill:#e3f2fd
    style BaseRunner fill:#e8f5e8
    style ToolManager fill:#fff3e0
    style Executor fill:#fce4ec

sequenceDiagram
    participant User as 用户
    participant AgentRunner as Agent运行器
    participant ModelRuntime as 模型运行时
    participant ToolEngine as 工具引擎
    participant CallbackHandler as 回调处理器
    
    Note over User,CallbackHandler: Agent推理执行流程
    
    User->>AgentRunner: 发送查询请求
    AgentRunner->>AgentRunner: 初始化Agent配置
    AgentRunner->>AgentRunner: 加载工具列表
    
    loop 推理迭代循环
        AgentRunner->>ModelRuntime: 发送提示消息
        ModelRuntime-->>AgentRunner: 返回推理结果
        
        alt 需要调用工具
            AgentRunner->>AgentRunner: 解析工具调用
            AgentRunner->>ToolEngine: 执行工具调用
            ToolEngine-->>AgentRunner: 返回工具结果
            AgentRunner->>CallbackHandler: 发布工具执行事件
        else 生成最终答案
            AgentRunner->>CallbackHandler: 发布答案事件
            Note over AgentRunner: 推理完成，退出循环
        end
        
        AgentRunner->>AgentRunner: 更新推理历史
        AgentRunner->>AgentRunner: 检查迭代次数限制
    end
    
    AgentRunner-->>User: 返回最终答案

sequenceDiagram
    participant User as 用户
    participant AgentRunner as Agent运行器
    participant ModelRuntime as 模型运行时
    participant ToolManager as 工具管理器
    participant Tool as 工具实例
    participant Callback as 回调处理器/队列
    
    Note over User,Callback: 函数调用策略下的工具调用细化流程

    User->>AgentRunner: 发送查询
    AgentRunner->>AgentRunner: _init_prompt_tools()
    AgentRunner->>ModelRuntime: invoke_llm(prompt, tools)
    ModelRuntime-->>AgentRunner: 返回message与tool_calls

    alt 存在tool_calls
        AgentRunner->>Callback: on_tool_start(tool_name, input)
        loop 遍历每个tool_call
            AgentRunner->>ToolManager: 获取工具运行时
            ToolManager-->>AgentRunner: 返回Tool实例
            AgentRunner->>Tool: invoke(user_id, tool_parameters)
            Tool-->>AgentRunner: ToolInvokeMessage结果
            AgentRunner->>Callback: on_tool_end(tool_name, result)
        end
        AgentRunner->>AgentRunner: 保存思维/统计并组织下一轮提示
        AgentRunner->>ModelRuntime: 追加工具观察后再次invoke_llm
    else 无tool_calls
        AgentRunner->>Callback: 发布最终答案事件
        AgentRunner-->>User: 返回最终答案（流式/阻塞）
    end

# 生产环境中的策略选择优化
class AgentStrategySelector:
    """
    智能Agent策略选择器
    根据模型能力、任务复杂度和性能要求自动选择最优策略
    """
    
    def select_optimal_strategy(
        self, 
        model_instance: ModelInstance,
        task_complexity: str,
        performance_requirements: dict
    ) -> AgentEntity.Strategy:
        """
        智能选择Agent策略
        
        Args:
            model_instance: 模型实例
            task_complexity: 任务复杂度（simple, medium, complex）
            performance_requirements: 性能要求
            
        Returns:
            AgentEntity.Strategy: 最优策略
        """
        # 1. 检查模型原生能力
        model_features = self._get_model_features(model_instance)
        
        # 2. 评估任务复杂度
        complexity_score = self._calculate_complexity_score(task_complexity)
        
        # 3. 考虑性能要求
        latency_requirement = performance_requirements.get("max_latency", 30)
        accuracy_requirement = performance_requirements.get("min_accuracy", 0.8)
        
        # 4. 智能决策
        if (ModelFeature.FUNCTION_CALLING in model_features and 
            latency_requirement < 10 and accuracy_requirement > 0.9):
            return AgentEntity.Strategy.FUNCTION_CALLING
        else:
            return AgentEntity.Strategy.CHAIN_OF_THOUGHT

class AgentEntity(BaseModel):
    """
    Agent实体定义
    包含Agent的核心配置和策略信息
    """
    
    class Strategy(StrEnum):
        """
        Agent策略枚举
        定义了支持的推理策略类型
        """
        
        # 思维链策略 - 适用于不支持函数调用的模型
        CHAIN_OF_THOUGHT = "chain-of-thought"
        
        # 函数调用策略 - 适用于支持函数调用的模型
        FUNCTION_CALLING = "function-calling"

    # 模型提供者（如openai、anthropic等）
    provider: str
    
    # 模型名称（如gpt-4、claude-3等）
    model: str
    
    # 推理策略
    strategy: Strategy
    
    # 自定义提示模板
    prompt: Optional[AgentPromptEntity] = None
    
    # 可用工具列表
    tools: Optional[List[AgentToolEntity]] = None
    
    # 最大推理迭代次数
    max_iteration: int = 10

# Agent策略选择逻辑
def select_agent_strategy(model_instance: ModelInstance) -> AgentEntity.Strategy:
    """
    根据模型能力自动选择最适合的Agent策略
    
    Args:
        model_instance: 模型实例
        
    Returns:
        AgentEntity.Strategy: 选定的策略
    """
    # 获取模型架构信息
    llm_model = cast(LargeLanguageModel, model_instance.model_type_instance)
    model_schema = llm_model.get_model_schema(
        model_instance.model, 
        model_instance.credentials
    )
    
    if not model_schema or not model_schema.features:
        return AgentEntity.Strategy.CHAIN_OF_THOUGHT
    
    # 检查是否支持工具调用功能
    tool_call_features = {
        ModelFeature.MULTI_TOOL_CALL,  # 支持多工具并行调用
        ModelFeature.TOOL_CALL,        # 支持单工具调用
    }
    
    if tool_call_features.intersection(model_schema.features):
        # 模型原生支持函数调用，使用函数调用策略
        return AgentEntity.Strategy.FUNCTION_CALLING
    else:
        # 模型不支持函数调用，使用思维链策略
        return AgentEntity.Strategy.CHAIN_OF_THOUGHT

# 策略特性对比
agent_strategy_features = {
    "function-calling": {
        "model_requirement": "支持函数调用的模型（GPT-4、Claude-3等）",
        "execution_efficiency": "高效",  # 直接调用，无需解析
        "accuracy": "高",               # 结构化调用，准确性高
        "flexibility": "中等",          # 依赖模型函数调用能力
        "debugging": "容易",            # 清晰的调用日志
        "supported_models": ["gpt-4", "gpt-3.5-turbo", "claude-3"],
        "best_for": "生产环境、高精度要求"
    },
    "chain-of-thought": {
        "model_requirement": "任何文本生成模型",
        "execution_efficiency": "中等", # 需要解析文本输出
        "accuracy": "中等",             # 依赖文本解析准确性
        "flexibility": "高",            # 适用于所有模型
        "debugging": "复杂",            # 需要解析推理过程
        "supported_models": ["通用模型", "本地部署模型", "较老的模型"],
        "best_for": "通用性、模型兼容性要求"
    }
}

class FunctionCallAgentRunner(BaseAgentRunner):
    """
    函数调用Agent运行器
    利用模型的原生函数调用能力执行工具调用
    """

    def run(self, message: Message, query: str, **kwargs: Any) -> Generator[LLMResultChunk, None, None]:
        """
        执行函数调用Agent推理流程
        
        Args:
            message: 消息对象
            query: 用户查询
            **kwargs: 其他参数
            
        Yields:
            LLMResultChunk: 流式推理结果
        """
        self.query = query
        app_generate_entity = self.application_generate_entity
        app_config = self.app_config

        # 1. 初始化工具系统
        tool_instances, prompt_messages_tools = self._init_prompt_tools()

        # 2. 设置推理参数
        iteration_step = 1
        max_iteration_steps = min(app_config.agent.max_iteration, 99) + 1
        
        # 推理状态控制
        function_call_state = True  # 是否需要继续工具调用
        llm_usage: Dict[str, Optional[LLMUsage]] = {"usage": None}
        final_answer = ""

        # 3. 主推理循环
        while function_call_state and iteration_step <= max_iteration_steps:
            function_call_state = False
            
            # 在最后一次迭代时移除工具，强制生成最终答案
            if iteration_step == max_iteration_steps:
                prompt_messages_tools = []

            # 创建思维记录
            agent_thought_id = self.create_agent_thought(
                message_id=message.id,
                message="",
                tool_name="",
                tool_input="",
                messages_ids=[]
            )

            # 4. 准备提示消息
            prompt_messages = self._organize_prompt_messages()
            self.recalc_llm_max_tokens(self.model_config, prompt_messages)

            # 5. 调用模型推理
            chunks = self.model_instance.invoke_llm(
                prompt_messages=prompt_messages,
                model_parameters=app_generate_entity.model_conf.parameters,
                tools=prompt_messages_tools,  # 传递工具定义
                stop=app_generate_entity.model_conf.stop,
                stream=self.stream_tool_call,
                user=self.user_id,
                callbacks=[],
            )

            # 6. 处理推理结果
            tool_calls: List[Tuple[str, str, Dict[str, Any]]] = []
            response = ""
            current_llm_usage = None

            if isinstance(chunks, Generator):
                # 流式处理
                for chunk in chunks:
                    # 处理工具调用
                    if chunk.delta.message.tool_calls:
                        function_call_state = True
                        self._handle_tool_calls_stream(chunk, tool_calls)
                    
                    # 处理文本响应
                    if chunk.delta.message.content:
                        response += chunk.delta.message.content
                        yield chunk

                    # 记录使用统计
                    if chunk.delta.usage:
                        current_llm_usage = chunk.delta.usage
            else:
                # 非流式处理
                if chunks.message.tool_calls:
                    function_call_state = True
                    tool_calls = self._extract_tool_calls(chunks.message.tool_calls)
                
                response = chunks.message.content or ""
                current_llm_usage = chunks.usage

            # 7. 执行工具调用
            if function_call_state and tool_calls:
                yield from self._execute_tool_calls(
                    tool_calls=tool_calls,
                    tool_instances=tool_instances,
                    agent_thought_id=agent_thought_id
                )
            
            # 8. 更新推理记录
            self.save_agent_thought(
                agent_thought_id=agent_thought_id,
                tool_name=";".join([call[0] for call in tool_calls]) if tool_calls else "",
                tool_input={call[0]: call[2] for call in tool_calls} if tool_calls else {},
                thought=response,
                observation="", # 工具执行结果会在_execute_tool_calls中更新
                answer=response if not function_call_state else "",
                messages_ids=[],
                llm_usage=current_llm_usage,
            )

            # 9. 更新使用统计
            if current_llm_usage:
                self._increase_usage(llm_usage, current_llm_usage)

            iteration_step += 1

        # 10. 返回最终结果
        if final_answer:
            yield self._create_final_answer_chunk(final_answer, llm_usage["usage"])

    def _handle_tool_calls_stream(self, chunk: LLMResultChunk, tool_calls: List):
        """
        处理流式工具调用
        
        Args:
            chunk: LLM结果块
            tool_calls: 工具调用列表
        """
        for tool_call in chunk.delta.message.tool_calls:
            if tool_call.function:
                # 解析函数调用参数
                function_name = tool_call.function.name
                function_arguments = tool_call.function.arguments
                
                try:
                    # 将JSON字符串参数解析为字典
                    arguments_dict = json.loads(function_arguments) if function_arguments else {}
                    tool_calls.append((function_name, tool_call.id, arguments_dict))
                except json.JSONDecodeError:
                    logger.warning(f"无法解析工具调用参数: {function_arguments}")
                    continue

    def _execute_tool_calls(
        self, 
        tool_calls: List[Tuple[str, str, Dict[str, Any]]],
        tool_instances: Dict[str, Tool],
        agent_thought_id: str
    ) -> Generator[LLMResultChunk, None, None]:
        """
        执行工具调用列表
        
        Args:
            tool_calls: 工具调用列表 [(工具名, 调用ID, 参数)]
            tool_instances: 工具实例字典
            agent_thought_id: 思维记录ID
            
        Yields:
            LLMResultChunk: 工具执行结果流
        """
        tool_responses = {}
        
        for tool_name, call_id, arguments in tool_calls:
            if tool_name not in tool_instances:
                error_msg = f"工具 '{tool_name}' 不存在"
                tool_responses[tool_name] = error_msg
                logger.error(error_msg)
                continue

            try:
                # 获取工具实例
                tool_instance = tool_instances[tool_name]
                
                # 发布工具调用开始事件
                yield self._create_tool_start_chunk(tool_name, arguments)
                
                # 执行工具调用
                tool_result = tool_instance.invoke(
                    user_id=self.user_id,
                    tool_parameters=arguments
                )
                
                # 处理工具结果
                tool_response = self._format_tool_response(tool_result)
                tool_responses[tool_name] = tool_response
                
                # 发布工具执行结果事件
                yield self._create_tool_result_chunk(tool_name, tool_response)
                
            except Exception as e:
                error_msg = f"工具 '{tool_name}' 执行失败: {str(e)}"
                tool_responses[tool_name] = error_msg
                logger.exception(error_msg)
                
                # 发布工具执行错误事件
                yield self._create_tool_error_chunk(tool_name, error_msg)

        # 更新Agent思维记录中的工具执行结果
        self._update_agent_thought_with_tool_results(agent_thought_id, tool_responses)

    def _organize_prompt_messages(self) -> List[PromptMessage]:
        """
        组织提示消息
        构建包含系统提示、历史对话和当前查询的完整提示
        
        Returns:
            List[PromptMessage]: 组织好的提示消息列表
        """
        messages: List[PromptMessage] = []
        
        # 1. 添加系统提示
        system_prompt = self._build_system_prompt()
        if system_prompt:
            messages.append(SystemPromptMessage(content=system_prompt))
        
        # 2. 添加历史对话
        messages.extend(self.history_prompt_messages)
        
        # 3. 添加当前用户查询
        user_message = self._build_user_message()
        messages.append(user_message)
        
        return messages

    def _build_system_prompt(self) -> str:
        """
        构建系统提示
        包含Agent的角色定义、能力说明和行为指南
        
        Returns:
            str: 系统提示内容
        """
        prompt_parts = []
        
        # 基础角色定义
        prompt_parts.append(
            "你是一个有用的AI助手，能够使用各种工具来帮助用户解决问题。"
        )
        
        # 工具使用指导
        if self.app_config.agent and self.app_config.agent.tools:
            prompt_parts.append(
                "你可以使用以下工具来获取信息或执行操作："
            )
            
            for tool in self.app_config.agent.tools:
                tool_description = f"- {tool.tool_name}: {tool.description or '执行相关操作'}"
                prompt_parts.append(tool_description)
        
        # 行为准则
        prompt_parts.extend([
            "请遵循以下准则：",
            "1. 仔细分析用户的问题，选择最适合的工具",
            "2. 如果需要多个步骤，请逐步执行",
            "3. 基于工具结果提供准确的答案",
            "4. 如果无法解决问题，请诚实说明原因"
        ])
        
        return "\n\n".join(prompt_parts)

    def _build_user_message(self) -> UserPromptMessage:
        """
        构建用户消息
        包含用户查询和可能的文件附件
        
        Returns:
            UserPromptMessage: 用户消息对象
        """
        if not self.files:
            # 纯文本查询
            return UserPromptMessage(content=self.query)
        
        # 包含文件的多模态查询
        content_parts: List[PromptMessageContentUnionTypes] = []
        
        # 添加文件内容
        for file in self.files:
            file_content = file_manager.to_prompt_message_content(file)
            content_parts.append(file_content)
        
        # 添加文本查询
        content_parts.append(TextPromptMessageContent(data=self.query))
        
        return UserPromptMessage(content=content_parts)

class CotAgentRunner(BaseAgentRunner, ABC):
    """
    思维链Agent运行器抽象基类
    通过ReACT模式引导模型进行推理和行动
    """
    
    # ReACT模式的关键词
    _react_keywords = {
        "thought": "Thought:",      # 思考标记
        "action": "Action:",        # 行动标记  
        "observation": "Observation:",  # 观察标记
        "final_answer": "Final Answer:"  # 最终答案标记
    }
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._is_first_iteration = True
        self._ignore_observation_providers = ["wenxin"]  # 忽略观察的提供者
        self._historic_prompt_messages: List[PromptMessage] = []
        self._agent_scratchpad: List[AgentScratchpadUnit] = []
        self._instruction: str = ""
        self._query: str = ""
        self._prompt_messages_tools: Sequence[PromptMessageTool] = []

    def run(
        self,
        message: Message,
        query: str,
        inputs: Mapping[str, str],
    ) -> Generator:
        """
        执行思维链Agent推理流程
        
        Args:
            message: 消息对象
            query: 用户查询
            inputs: 输入变量
            
        Yields:
            推理过程中的各种事件
        """
        app_generate_entity = self.application_generate_entity
        self._repack_app_generate_entity(app_generate_entity)
        self._init_react_state(query)

        # 确保模型停止词包含"Observation"
        if "Observation" not in app_generate_entity.model_conf.stop:
            if app_generate_entity.model_conf.provider not in self._ignore_observation_providers:
                app_generate_entity.model_conf.stop.append("Observation")

        app_config = self.app_config
        
        # 初始化指令模板
        instruction = app_config.prompt_template.simple_prompt_template or ""
        self._instruction = self._fill_in_inputs_from_external_data_tools(instruction, inputs or {})

        # 设置推理参数
        iteration_step = 1
        max_iteration_steps = min(app_config.agent.max_iteration, 99) + 1

        # 初始化工具系统
        tool_instances, prompt_messages_tools = self._init_prompt_tools()
        self._prompt_messages_tools = prompt_messages_tools

        function_call_state = True
        llm_usage: Dict[str, Optional[LLMUsage]] = {"usage": None}
        final_answer = ""

        # 主推理循环
        while function_call_state and iteration_step <= max_iteration_steps:
            function_call_state = False

            # 创建思维记录
            agent_thought_id = self.create_agent_thought(
                message_id=message.id,
                message="",
                tool_name="",
                tool_input="",
                messages_ids=[]
            )

            # 构建ReACT提示
            prompt_messages = self._organize_react_prompt_messages()
            self.recalc_llm_max_tokens(self.model_config, prompt_messages)

            # 调用模型推理
            chunks = self.model_instance.invoke_llm(
                prompt_messages=prompt_messages,
                model_parameters=app_generate_entity.model_conf.parameters,
                stop=app_generate_entity.model_conf.stop,
                stream=self.stream_tool_call,
                user=self.user_id,
                callbacks=[],
            )

            # 处理推理结果
            response_text = ""
            current_llm_usage = None

            if isinstance(chunks, Generator):
                # 流式处理推理过程
                for chunk in chunks:
                    if chunk.delta.message.content:
                        response_text += chunk.delta.message.content
                        yield chunk
                    
                    if chunk.delta.usage:
                        current_llm_usage = chunk.delta.usage
            else:
                response_text = chunks.message.content or ""
                current_llm_usage = chunks.usage

            # 解析ReACT响应
            react_result = self._parse_react_response(response_text)
            
            if react_result.get("action") and react_result.get("action_input"):
                # 需要执行工具调用
                function_call_state = True
                yield from self._execute_react_tool_call(
                    action=react_result["action"],
                    action_input=react_result["action_input"],
                    tool_instances=tool_instances,
                    agent_thought_id=agent_thought_id
                )
            elif react_result.get("final_answer"):
                # 获得最终答案
                final_answer = react_result["final_answer"]
                function_call_state = False

            # 更新推理历史
            self._update_agent_scratchpad(react_result)
            
            # 保存思维记录
            self.save_agent_thought(
                agent_thought_id=agent_thought_id,
                tool_name=react_result.get("action", ""),
                tool_input=react_result.get("action_input", {}),
                thought=react_result.get("thought", ""),
                observation=react_result.get("observation", ""),
                answer=react_result.get("final_answer", ""),
                messages_ids=[],
                llm_usage=current_llm_usage,
            )

            iteration_step += 1

        # 返回最终答案
        if final_answer:
            yield self._create_final_answer_chunk(final_answer, llm_usage["usage"])

    def _init_react_state(self, query: str):
        """
        初始化ReACT推理状态
        
        Args:
            query: 用户查询
        """
        self._query = query
        self._agent_scratchpad = []
        self._is_first_iteration = True

    def _organize_react_prompt_messages(self) -> List[PromptMessage]:
        """
        组织ReACT推理提示
        构建符合ReACT模式的提示模板
        
        Returns:
            List[PromptMessage]: ReACT提示消息列表
        """
        messages: List[PromptMessage] = []

        # 1. 构建系统提示（包含ReACT模板）
        system_prompt = self._build_react_system_prompt()
        messages.append(SystemPromptMessage(content=system_prompt))

        # 2. 添加历史对话
        messages.extend(self.history_prompt_messages)

        # 3. 添加当前推理会话
        if self._agent_scratchpad:
            scratchpad_content = self._format_agent_scratchpad()
            messages.append(UserPromptMessage(content=scratchpad_content))
        else:
            # 第一次推理
            user_prompt = f"用户问题: {self._query}\n\n请开始推理："
            messages.append(UserPromptMessage(content=user_prompt))

        return messages

    def _build_react_system_prompt(self) -> str:
        """
        构建ReACT系统提示模板
        
        Returns:
            str: ReACT系统提示
        """
        # 基础指令
        base_instruction = self._instruction or "你是一个有用的AI助手。"
        
        # 工具描述
        tools_description = self._build_tools_description()
        
        # ReACT模板
        react_template = """
你必须使用以下格式进行推理：

Thought: 你应该总是思考你要做什么
Action: 要采取的行动，应该是以下之一：[{tool_names}]
Action Input: 行动的输入
Observation: 行动的结果
... (这个 Thought/Action/Action Input/Observation 可以重复N次)
Thought: 我现在知道最终答案
Final Answer: 原始输入问题的最终答案

开始！

{base_instruction}

{tools_description}
""".strip()

        tool_names = ", ".join([tool.name for tool in self._prompt_messages_tools])
        
        return react_template.format(
            tool_names=tool_names,
            base_instruction=base_instruction,
            tools_description=tools_description
        )

    def _build_tools_description(self) -> str:
        """
        构建工具描述
        
        Returns:
            str: 格式化的工具描述
        """
        if not self._prompt_messages_tools:
            return "当前没有可用的工具。"
        
        tool_descriptions = []
        for tool in self._prompt_messages_tools:
            description = f"- {tool.name}: {tool.description}"
            
            # 添加参数信息
            if tool.parameters and "properties" in tool.parameters:
                params = []
                for param_name, param_info in tool.parameters["properties"].items():
                    param_desc = f"{param_name}({param_info.get('type', 'any')})"
                    if param_name in tool.parameters.get("required", []):
                        param_desc += "*"  # 标记必填参数
                    params.append(param_desc)
                
                if params:
                    description += f" - 参数: {', '.join(params)}"
            
            tool_descriptions.append(description)
        
        return "可用工具：\n" + "\n".join(tool_descriptions)

    def _parse_react_response(self, response: str) -> Dict[str, Any]:
        """
        解析ReACT格式的模型响应
        
        Args:
            response: 模型生成的文本响应
            
        Returns:
            Dict[str, Any]: 解析结果
        """
        result = {}
        
        # 使用正则表达式提取各个部分
        patterns = {
            "thought": r"Thought:\s*(.*?)(?=\nAction:|\nFinal Answer:|$)",
            "action": r"Action:\s*(.*?)(?=\nAction Input:|\n|$)",
            "action_input": r"Action Input:\s*(.*?)(?=\nObservation:|\n|$)",
            "observation": r"Observation:\s*(.*?)(?=\nThought:|\n|$)",
            "final_answer": r"Final Answer:\s*(.*?)$"
        }
        
        for key, pattern in patterns.items():
            match = re.search(pattern, response, re.DOTALL | re.IGNORECASE)
            if match:
                result[key] = match.group(1).strip()
        
        # 处理action_input，尝试解析为JSON
        if "action_input" in result:
            try:
                # 尝试解析为JSON对象
                result["action_input"] = json.loads(result["action_input"])
            except json.JSONDecodeError:
                # 如果不是JSON，保持原始字符串
                pass
        
        return result

    def _execute_react_tool_call(
        self,
        action: str,
        action_input: Union[str, Dict[str, Any]],
        tool_instances: Dict[str, Tool],
        agent_thought_id: str
    ) -> Generator:
        """
        执行ReACT工具调用
        
        Args:
            action: 工具名称
            action_input: 工具输入参数
            tool_instances: 工具实例字典
            agent_thought_id: 思维记录ID
            
        Yields:
            工具执行过程事件
        """
        if action not in tool_instances:
            error_msg = f"未找到工具: {action}"
            yield self._create_tool_error_chunk(action, error_msg)
            self._add_scratchpad_observation(error_msg)
            return

        try:
            # 准备工具参数
            if isinstance(action_input, str):
                # 尝试解析字符串参数为字典
                try:
                    tool_params = json.loads(action_input)
                except json.JSONDecodeError:
                    tool_params = {"input": action_input}
            else:
                tool_params = action_input

            # 执行工具调用
            tool_instance = tool_instances[action]
            yield self._create_tool_start_chunk(action, tool_params)
            
            tool_result = tool_instance.invoke(
                user_id=self.user_id,
                tool_parameters=tool_params
            )
            
            # 格式化工具结果
            observation = self._format_tool_response(tool_result)
            yield self._create_tool_result_chunk(action, observation)
            
            # 添加到推理记录
            self._add_scratchpad_observation(observation)
            
        except Exception as e:
            error_msg = f"工具执行错误: {str(e)}"
            logger.exception(f"ReACT工具调用失败 - {action}: {e}")
            yield self._create_tool_error_chunk(action, error_msg)
            self._add_scratchpad_observation(error_msg)

    def _format_agent_scratchpad(self) -> str:
        """
        格式化Agent推理历史
        
        Returns:
            str: 格式化的推理历史
        """
        if not self._agent_scratchpad:
            return f"用户问题: {self._query}\n\nThought:"
        
        scratchpad_parts = [f"用户问题: {self._query}\n"]
        
        for unit in self._agent_scratchpad:
            if unit.thought:
                scratchpad_parts.append(f"Thought: {unit.thought}")
            if unit.action:
                scratchpad_parts.append(f"Action: {unit.action}")
            if unit.action_input is not None:
                if isinstance(unit.action_input, dict):
                    action_input_str = json.dumps(unit.action_input, ensure_ascii=False)
                else:
                    action_input_str = str(unit.action_input)
                scratchpad_parts.append(f"Action Input: {action_input_str}")
            if unit.observation:
                scratchpad_parts.append(f"Observation: {unit.observation}")
        
        # 添加下一步提示
        scratchpad_parts.append("Thought:")
        
        return "\n".join(scratchpad_parts)

    def _update_agent_scratchpad(self, react_result: Dict[str, Any]):
        """
        更新Agent推理历史
        
        Args:
            react_result: ReACT解析结果
        """
        unit = AgentScratchpadUnit(
            thought=react_result.get("thought", ""),
            action=react_result.get("action", ""),
            action_input=react_result.get("action_input"),
            observation=react_result.get("observation", ""),
            final_answer=react_result.get("final_answer", "")
        )
        
        self._agent_scratchpad.append(unit)

    def _add_scratchpad_observation(self, observation: str):
        """
        向最新的推理单元添加观察结果
        
        Args:
            observation: 观察结果
        """
        if self._agent_scratchpad:
            self._agent_scratchpad[-1].observation = observation

class AgentScratchpadUnit(BaseModel):
    """
    Agent推理单元
    记录一次完整的Thought-Action-Observation循环
    """
    
    # 思考过程
    thought: str = ""
    
    # 采取的行动（工具名称）
    action: str = ""
    
    # 行动输入参数
    action_input: Union[str, Dict[str, Any], None] = None
    
    # 观察结果（工具执行结果）
    observation: str = ""
    
    # 最终答案（如果有）
    final_answer: str = ""
    
    @property
    def is_final(self) -> bool:
        """
        判断是否为最终答案单元
        
        Returns:
            bool: 是否包含最终答案
        """
        return bool(self.final_answer)

class CotChatAgentRunner(CotAgentRunner):
    """
    基于Chat模型的思维链Agent运行器
    专门处理对话模型的ReACT推理
    """
    
    def _organize_react_prompt_messages(self) -> List[PromptMessage]:
        """
        为Chat模型组织ReACT提示消息
        Chat模型需要明确的对话角色分离
        
        Returns:
            List[PromptMessage]: 适合Chat模型的提示消息
        """
        messages: List[PromptMessage] = []

        # 1. 系统角色提示
        system_prompt = self._build_react_system_prompt()
        messages.append(SystemPromptMessage(content=system_prompt))

        # 2. 历史对话
        messages.extend(self.history_prompt_messages)

        # 3. 当前推理会话
        if self._agent_scratchpad and not self._is_first_iteration:
            # 构建包含推理历史的用户消息
            scratchpad_content = self._format_agent_scratchpad()
            messages.append(UserPromptMessage(content=scratchpad_content))
        else:
            # 第一次推理
            messages.append(UserPromptMessage(content=f"用户问题: {self._query}"))
            
        self._is_first_iteration = False
        
        return messages

class CotCompletionAgentRunner(CotAgentRunner):
    """
    基于Completion模型的思维链Agent运行器
    专门处理文本完成模型的ReACT推理
    """
    
    def _organize_react_prompt_messages(self) -> List[PromptMessage]:
        """
        为Completion模型组织ReACT提示消息
        Completion模型使用单一的连续文本提示
        
        Returns:
            List[PromptMessage]: 适合Completion模型的提示消息
        """
        # Completion模型使用单一的UserPromptMessage
        prompt_parts = []
        
        # 1. 系统指令和工具描述
        system_prompt = self._build_react_system_prompt()
        prompt_parts.append(system_prompt)
        
        # 2. 历史对话（如果有）
        if self.history_prompt_messages:
            for msg in self.history_prompt_messages:
                if isinstance(msg, UserPromptMessage):
                    prompt_parts.append(f"Human: {msg.content}")
                elif isinstance(msg, AssistantPromptMessage):
                    prompt_parts.append(f"Assistant: {msg.content}")
        
        # 3. 当前推理会话
        if self._agent_scratchpad:
            scratchpad_content = self._format_agent_scratchpad()
            prompt_parts.append(scratchpad_content)
        else:
            # 第一次推理
            prompt_parts.append(f"用户问题: {self._query}\n\nThought:")
        
        # 合并所有部分为单一提示
        full_prompt = "\n\n".join(prompt_parts)
        
        return [UserPromptMessage(content=full_prompt)]

class ToolManager:
    """
    工具管理器
    负责工具的注册、加载和执行管理
    """
    
    @staticmethod
    def get_agent_tool_runtime(
        tenant_id: str,
        app_id: str,
        agent_tool: AgentToolEntity,
        invoke_from: InvokeFrom
    ) -> Tool:
        """
        获取Agent工具运行时实例
        
        Args:
            tenant_id: 租户ID
            app_id: 应用ID
            agent_tool: Agent工具实体
            invoke_from: 调用来源
            
        Returns:
            Tool: 工具运行时实例
        """
        tool_type = agent_tool.tool_type
        
        if tool_type == AgentToolType.BUILTIN:
            # 内置工具
            return BuiltinToolManager.get_tool_runtime(
                provider=agent_tool.provider,
                tool_name=agent_tool.tool_name,
                tenant_id=tenant_id
            )
        elif tool_type == AgentToolType.API:
            # API工具
            return ApiToolManager.get_tool_runtime(
                tenant_id=tenant_id,
                tool_id=agent_tool.tool_id,
                invoke_from=invoke_from
            )
        elif tool_type == AgentToolType.WORKFLOW:
            # 工作流工具
            return WorkflowToolManager.get_tool_runtime(
                tenant_id=tenant_id,
                app_id=app_id,
                workflow_id=agent_tool.workflow_id,
                invoke_from=invoke_from
            )
        else:
            raise ValueError(f"不支持的工具类型: {tool_type}")

class AgentToolEntity(BaseModel):
    """
    Agent工具实体
    定义Agent可使用的工具配置
    """
    
    # 工具类型
    tool_type: AgentToolType
    
    # 工具提供者（用于内置工具）
    provider: Optional[str] = None
    
    # 工具名称
    tool_name: str
    
    # 工具ID（用于API工具）
    tool_id: Optional[str] = None
    
    # 工作流ID（用于工作流工具）
    workflow_id: Optional[str] = None
    
    # 工具配置参数
    tool_configuration: Dict[str, Any] = {}
    
    # 工具描述
    description: Optional[str] = None
    
    # 是否启用
    enabled: bool = True

class AgentToolType(Enum):
    """Agent工具类型枚举"""
    
    # 内置工具 - 平台提供的预定义工具
    BUILTIN = "builtin"
    
    # API工具 - 通过API调用的外部工具
    API = "api"
    
    # 工作流工具 - Dify工作流作为工具使用
    WORKFLOW = "workflow"
    
    # 数据集工具 - 知识库检索工具
    DATASET = "dataset"

# 工具类型特性对比
tool_type_features = {
    "builtin": {
        "description": "平台内置的预定义工具",
        "examples": ["web_search", "calculator", "weather", "translator"],
        "advantages": ["稳定可靠", "性能优化", "安全性高"],
        "limitations": ["功能固定", "定制能力有限"],
        "use_cases": ["通用功能", "基础工具需求"]
    },
    "api": {
        "description": "通过API接口调用的外部服务",
        "examples": ["第三方API", "企业内部API", "微服务接口"],
        "advantages": ["功能丰富", "高度定制", "集成灵活"],
        "limitations": ["依赖外部服务", "网络延迟", "可靠性问题"],
        "use_cases": ["专业服务", "企业集成", "定制功能"]
    },
    "workflow": {
        "description": "Dify工作流作为工具使用",
        "examples": ["复杂业务流程", "多步骤处理", "条件分支逻辑"],
        "advantages": ["逻辑复杂", "可视化配置", "复用性高"],
        "limitations": ["执行耗时", "资源消耗大"],
        "use_cases": ["复杂处理逻辑", "业务流程自动化"]
    },
    "dataset": {
        "description": "知识库检索和RAG功能",
        "examples": ["文档检索", "知识问答", "上下文增强"],
        "advantages": ["知识丰富", "上下文相关", "准确度高"],
        "limitations": ["依赖数据质量", "检索效果限制"],
        "use_cases": ["知识问答", "文档查询", "专业咨询"]
    }
}

class DatasetRetrieverTool:
    """
    数据集检索工具
    为Agent提供知识库检索能力
    """
    
    @staticmethod
    def get_dataset_tools(
        tenant_id: str,
        dataset_ids: List[str],
        retrieve_config: Optional[DatasetRetrievalConfig] = None,
        return_resource: bool = True,
        invoke_from: InvokeFrom = InvokeFrom.WEB_APP,
        hit_callback: Optional[DatasetIndexToolCallbackHandler] = None,
        user_id: str = "",
        inputs: Dict[str, Any] = None,
    ) -> List['DatasetRetrieverTool']:
        """
        创建数据集检索工具列表
        
        Args:
            tenant_id: 租户ID
            dataset_ids: 数据集ID列表
            retrieve_config: 检索配置
            return_resource: 是否返回资源信息
            invoke_from: 调用来源
            hit_callback: 检索回调处理器
            user_id: 用户ID
            inputs: 输入参数
            
        Returns:
            List[DatasetRetrieverTool]: 数据集工具列表
        """
        if not dataset_ids:
            return []
        
        tools = []
        for dataset_id in dataset_ids:
            try:
                # 验证数据集访问权限
                dataset = DatasetService.get_dataset(tenant_id, dataset_id)
                if not dataset:
                    logger.warning(f"数据集不存在: {dataset_id}")
                    continue
                
                # 创建检索工具实例
                tool = DatasetRetrieverTool(
                    tenant_id=tenant_id,
                    dataset_id=dataset_id,
                    retrieve_config=retrieve_config or DatasetRetrievalConfig(),
                    return_resource=return_resource,
                    invoke_from=invoke_from,
                    hit_callback=hit_callback,
                    user_id=user_id,
                    inputs=inputs or {}
                )
                
                tools.append(tool)
                
            except Exception as e:
                logger.exception(f"创建数据集工具失败 - {dataset_id}: {e}")
                continue
        
        return tools

    def __init__(
        self,
        tenant_id: str,
        dataset_id: str,
        retrieve_config: DatasetRetrievalConfig,
        return_resource: bool = True,
        invoke_from: InvokeFrom = InvokeFrom.WEB_APP,
        hit_callback: Optional[DatasetIndexToolCallbackHandler] = None,
        user_id: str = "",
        inputs: Dict[str, Any] = None,
    ):
        """
        初始化数据集检索工具
        
        Args:
            tenant_id: 租户ID
            dataset_id: 数据集ID
            retrieve_config: 检索配置
            return_resource: 是否返回资源信息
            invoke_from: 调用来源
            hit_callback: 检索回调处理器
            user_id: 用户ID
            inputs: 输入参数
        """
        self.tenant_id = tenant_id
        self.dataset_id = dataset_id
        self.retrieve_config = retrieve_config
        self.return_resource = return_resource
        self.invoke_from = invoke_from
        self.hit_callback = hit_callback
        self.user_id = user_id
        self.inputs = inputs or {}
        
        # 加载数据集信息
        self.dataset = DatasetService.get_dataset(tenant_id, dataset_id)
        if not self.dataset:
            raise ValueError(f"数据集不存在: {dataset_id}")
        
        # 初始化工具实体
        self.entity = ToolEntity(
            identity=ToolIdentity(
                name=f"dataset_{dataset_id}",
                label=I18nObject(
                    en_US=f"Query {self.dataset.name}",
                    zh_Hans=f"查询{self.dataset.name}"
                )
            ),
            description=ToolDescription(
                human=I18nObject(
                    en_US=f"Search and retrieve information from {self.dataset.name} dataset",
                    zh_Hans=f"从{self.dataset.name}数据集搜索和检索信息"
                ),
                llm=f"Use this tool to search for information in the {self.dataset.name} knowledge base. "
                    f"Provide a search query as input and get relevant documents and answers."
            )
        )

    def get_runtime_parameters(self) -> List[ToolParameter]:
        """
        获取运行时参数定义
        
        Returns:
            List[ToolParameter]: 参数列表
        """
        return [
            ToolParameter(
                name="query",
                label=I18nObject(
                    en_US="Search Query",
                    zh_Hans="搜索查询"
                ),
                human_description=I18nObject(
                    en_US="The search query to find relevant information",
                    zh_Hans="用于查找相关信息的搜索查询"
                ),
                llm_description="The search query string to find relevant documents and information",
                form=ToolParameter.ToolParameterForm.LLM,
                type=ToolParameter.ToolParameterType.STRING,
                required=True
            )
        ]

    def invoke(
        self,
        user_id: str,
        tool_parameters: Dict[str, Any]
    ) -> ToolInvokeMessage:
        """
        执行数据集检索
        
        Args:
            user_id: 用户ID
            tool_parameters: 工具参数
            
        Returns:
            ToolInvokeMessage: 检索结果
        """
        query = tool_parameters.get("query", "")
        if not query or not query.strip():
            return ToolInvokeMessage(
                type=ToolInvokeMessage.MessageType.TEXT,
                message="搜索查询不能为空"
            )

        try:
            # 执行数据集检索
            retrieval_service = RetrievalService()
            retrieve_result = retrieval_service.retrieve(
                tenant_id=self.tenant_id,
                dataset_id=self.dataset_id,
                query=query,
                top_k=self.retrieve_config.top_k,
                score_threshold=self.retrieve_config.score_threshold,
                reranking_config=self.retrieve_config.reranking_config,
                user_id=user_id
            )

            # 处理检索结果
            if not retrieve_result.documents:
                return ToolInvokeMessage(
                    type=ToolInvokeMessage.MessageType.TEXT,
                    message="未找到相关信息"
                )

            # 格式化检索结果
            result_parts = []
            
            # 添加文档内容
            for i, doc in enumerate(retrieve_result.documents[:self.retrieve_config.top_k]):
                doc_content = f"[文档 {i+1}]\n{doc.page_content}"
                if self.return_resource and doc.metadata:
                    # 添加资源信息
                    metadata_parts = []
                    if doc.metadata.get("source"):
                        metadata_parts.append(f"来源: {doc.metadata['source']}")
                    if doc.metadata.get("page"):
                        metadata_parts.append(f"页码: {doc.metadata['page']}")
                    if doc.metadata.get("score"):
                        metadata_parts.append(f"相关度: {doc.metadata['score']:.3f}")
                    
                    if metadata_parts:
                        doc_content += f"\n({', '.join(metadata_parts)})"
                
                result_parts.append(doc_content)

            # 合并结果
            final_result = "\n\n".join(result_parts)

            # 触发检索回调
            if self.hit_callback:
                self.hit_callback.on_retrieval_success(
                    query=query,
                    documents=retrieve_result.documents
                )

            return ToolInvokeMessage(
                type=ToolInvokeMessage.MessageType.TEXT,
                message=final_result
            )

        except Exception as e:
            error_message = f"数据集检索失败: {str(e)}"
            logger.exception(error_message)
            
            # 触发错误回调
            if self.hit_callback:
                self.hit_callback.on_retrieval_error(query=query, error=str(e))
            
            return ToolInvokeMessage(
                type=ToolInvokeMessage.MessageType.TEXT,
                message=error_message
            )

class DatasetRetrievalConfig(BaseModel):
    """数据集检索配置"""
    
    # 返回文档数量
    top_k: int = 3
    
    # 相关度阈值
    score_threshold: float = 0.5
    
    # 重排序配置
    reranking_config: Optional[RerankingConfig] = None
    
    # 检索模式
    retrieval_mode: RetrievalMode = RetrievalMode.SINGLE
    
    # 是否启用多轮检索
    multi_round_enabled: bool = False

class DatasetIndexToolCallbackHandler:
    """
    数据集检索工具回调处理器
    处理检索过程中的各种事件
    """
    
    def __init__(
        self,
        queue_manager: AppQueueManager,
        app_id: str,
        message_id: str,
        user_id: str,
        invoke_from: InvokeFrom
    ):
        """
        初始化回调处理器
        
        Args:
            queue_manager: 队列管理器
            app_id: 应用ID
            message_id: 消息ID
            user_id: 用户ID
            invoke_from: 调用来源
        """
        self.queue_manager = queue_manager
        self.app_id = app_id
        self.message_id = message_id
        self.user_id = user_id
        self.invoke_from = invoke_from

    def on_retrieval_success(self, query: str, documents: List[Document]):
        """
        检索成功回调
        
        Args:
            query: 搜索查询
            documents: 检索到的文档列表
        """
        # 发布检索资源事件
        retrieval_resources = [
            RetrievalResource(
                content=doc.page_content,
                metadata=doc.metadata,
                score=doc.metadata.get("score", 0.0) if doc.metadata else 0.0
            )
            for doc in documents
        ]
        
        self.queue_manager.publish(
            QueueRetrieverResourcesEvent(
                retrieval_resources=retrieval_resources
            ),
            PublishFrom.TOOL_ENGINE
        )

    def on_retrieval_error(self, query: str, error: str):
        """
        检索失败回调
        
        Args:
            query: 搜索查询
            error: 错误信息
        """
        logger.error(f"数据集检索失败 - 查询: {query}, 错误: {error}")
        
        # 可以发布错误事件或记录到监控系统
        # self.queue_manager.publish(
        #     QueueErrorEvent(error_message=f"数据集检索失败: {error}"),
        #     PublishFrom.TOOL_ENGINE
        # )