React

ReActAgent

Bases: AgentRunner

ReAct代理。

使用ReActAgentWorker的AgentRunner的子类。

有关旧版实现，请参见：

from llama_index.core.agent.legacy.react.base import ReActAgent

Source code in llama_index/core/agent/react/base.py

class ReActAgent(AgentRunner):
    """ReAct代理。

    使用ReActAgentWorker的AgentRunner的子类。

    有关旧版实现，请参见：
    ```python
    from llama_index.core.agent.legacy.react.base import ReActAgent
    ```"""

    def __init__(
        self,
        tools: Sequence[BaseTool],
        llm: LLM,
        memory: BaseMemory,
        max_iterations: int = 10,
        react_chat_formatter: Optional[ReActChatFormatter] = None,
        output_parser: Optional[ReActOutputParser] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        context: Optional[str] = None,
        handle_reasoning_failure_fn: Optional[
            Callable[[CallbackManager, Exception], ToolOutput]
        ] = None,
    ) -> None:
        """初始化参数。"""
        callback_manager = callback_manager or llm.callback_manager
        if context and react_chat_formatter:
            raise ValueError("Cannot provide both context and react_chat_formatter")
        if context:
            react_chat_formatter = ReActChatFormatter.from_context(context)

        step_engine = ReActAgentWorker.from_tools(
            tools=tools,
            tool_retriever=tool_retriever,
            llm=llm,
            max_iterations=max_iterations,
            react_chat_formatter=react_chat_formatter,
            output_parser=output_parser,
            callback_manager=callback_manager,
            verbose=verbose,
            handle_reasoning_failure_fn=handle_reasoning_failure_fn,
        )
        super().__init__(
            step_engine,
            memory=memory,
            llm=llm,
            callback_manager=callback_manager,
        )

    @classmethod
    def from_tools(
        cls,
        tools: Optional[List[BaseTool]] = None,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        llm: Optional[LLM] = None,
        chat_history: Optional[List[ChatMessage]] = None,
        memory: Optional[BaseMemory] = None,
        memory_cls: Type[BaseMemory] = ChatMemoryBuffer,
        max_iterations: int = 10,
        react_chat_formatter: Optional[ReActChatFormatter] = None,
        output_parser: Optional[ReActOutputParser] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        context: Optional[str] = None,
        handle_reasoning_failure_fn: Optional[
            Callable[[CallbackManager, Exception], ToolOutput]
        ] = None,
        **kwargs: Any,
    ) -> "ReActAgent":
        """方便的构造函数，从BaseTools集合中（可选）。

注意：kwargs在这一点上应该已经被用尽。换句话说，各种上游组件，如BaseSynthesizer（响应合成器）或BaseRetriever应该在它们的构造中从各自的kwargs中取出。

如果提供了`handle_reasoning_failure_fn`，当LLM无法遵循系统提示中指定的响应模板时，将调用此函数。这个函数应该提供给Agent，这样Agent就有第二次机会来纠正它的错误。
要自己处理异常，可以提供一个引发`Exception`的函数。

注意：如果您修改了系统提示中的任何响应模板，您应该在`ReActAgentWorker`中重写`_extract_reasoning_step`方法。

返回：
    ReActAgent
"""
        llm = llm or Settings.llm
        if callback_manager is not None:
            llm.callback_manager = callback_manager
        memory = memory or memory_cls.from_defaults(
            chat_history=chat_history or [], llm=llm
        )
        return cls(
            tools=tools or [],
            tool_retriever=tool_retriever,
            llm=llm,
            memory=memory,
            max_iterations=max_iterations,
            react_chat_formatter=react_chat_formatter,
            output_parser=output_parser,
            callback_manager=callback_manager,
            verbose=verbose,
            context=context,
            handle_reasoning_failure_fn=handle_reasoning_failure_fn,
        )

    def _get_prompt_modules(self) -> PromptMixinType:
        """获取提示模块。"""
        return {"agent_worker": self.agent_worker}

from_tools classmethod

from_tools(
    tools: Optional[List[BaseTool]] = None,
    tool_retriever: Optional[
        ObjectRetriever[BaseTool]
    ] = None,
    llm: Optional[LLM] = None,
    chat_history: Optional[List[ChatMessage]] = None,
    memory: Optional[BaseMemory] = None,
    memory_cls: Type[BaseMemory] = ChatMemoryBuffer,
    max_iterations: int = 10,
    react_chat_formatter: Optional[
        ReActChatFormatter
    ] = None,
    output_parser: Optional[ReActOutputParser] = None,
    callback_manager: Optional[CallbackManager] = None,
    verbose: bool = False,
    context: Optional[str] = None,
    handle_reasoning_failure_fn: Optional[
        Callable[[CallbackManager, Exception], ToolOutput]
    ] = None,
    **kwargs: Any
) -> ReActAgent

方便的构造函数，从BaseTools集合中（可选）。

注意：kwargs在这一点上应该已经被用尽。换句话说，各种上游组件，如BaseSynthesizer（响应合成器）或BaseRetriever应该在它们的构造中从各自的kwargs中取出。

如果提供了handle_reasoning_failure_fn，当LLM无法遵循系统提示中指定的响应模板时，将调用此函数。这个函数应该提供给Agent，这样Agent就有第二次机会来纠正它的错误。 要自己处理异常，可以提供一个引发Exception的函数。

注意：如果您修改了系统提示中的任何响应模板，您应该在ReActAgentWorker中重写_extract_reasoning_step方法。

返回： ReActAgent

Source code in llama_index/core/agent/react/base.py

    @classmethod
    def from_tools(
        cls,
        tools: Optional[List[BaseTool]] = None,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        llm: Optional[LLM] = None,
        chat_history: Optional[List[ChatMessage]] = None,
        memory: Optional[BaseMemory] = None,
        memory_cls: Type[BaseMemory] = ChatMemoryBuffer,
        max_iterations: int = 10,
        react_chat_formatter: Optional[ReActChatFormatter] = None,
        output_parser: Optional[ReActOutputParser] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        context: Optional[str] = None,
        handle_reasoning_failure_fn: Optional[
            Callable[[CallbackManager, Exception], ToolOutput]
        ] = None,
        **kwargs: Any,
    ) -> "ReActAgent":
        """方便的构造函数，从BaseTools集合中（可选）。

注意：kwargs在这一点上应该已经被用尽。换句话说，各种上游组件，如BaseSynthesizer（响应合成器）或BaseRetriever应该在它们的构造中从各自的kwargs中取出。

如果提供了`handle_reasoning_failure_fn`，当LLM无法遵循系统提示中指定的响应模板时，将调用此函数。这个函数应该提供给Agent，这样Agent就有第二次机会来纠正它的错误。
要自己处理异常，可以提供一个引发`Exception`的函数。

注意：如果您修改了系统提示中的任何响应模板，您应该在`ReActAgentWorker`中重写`_extract_reasoning_step`方法。

返回：
    ReActAgent
"""
        llm = llm or Settings.llm
        if callback_manager is not None:
            llm.callback_manager = callback_manager
        memory = memory or memory_cls.from_defaults(
            chat_history=chat_history or [], llm=llm
        )
        return cls(
            tools=tools or [],
            tool_retriever=tool_retriever,
            llm=llm,
            memory=memory,
            max_iterations=max_iterations,
            react_chat_formatter=react_chat_formatter,
            output_parser=output_parser,
            callback_manager=callback_manager,
            verbose=verbose,
            context=context,
            handle_reasoning_failure_fn=handle_reasoning_failure_fn,
        )

ReActAgentWorker

Bases: BaseAgentWorker

OpenAI代理工作程序。

Source code in llama_index/core/agent/react/step.py

class ReActAgentWorker(BaseAgentWorker):
    """OpenAI代理工作程序。"""

    def __init__(
        self,
        tools: Sequence[BaseTool],
        llm: LLM,
        max_iterations: int = 10,
        react_chat_formatter: Optional[ReActChatFormatter] = None,
        output_parser: Optional[ReActOutputParser] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        handle_reasoning_failure_fn: Optional[
            Callable[[CallbackManager, Exception], ToolOutput]
        ] = None,
    ) -> None:
        self._llm = llm
        self.callback_manager = callback_manager or llm.callback_manager
        self._max_iterations = max_iterations
        self._react_chat_formatter = react_chat_formatter or ReActChatFormatter()
        self._output_parser = output_parser or ReActOutputParser()
        self._verbose = verbose
        self._handle_reasoning_failure_fn = (
            handle_reasoning_failure_fn
            or tell_llm_about_failure_in_extract_reasoning_step
        )

        if len(tools) > 0 and tool_retriever is not None:
            raise ValueError("Cannot specify both tools and tool_retriever")
        elif len(tools) > 0:
            self._get_tools = lambda _: tools
        elif tool_retriever is not None:
            tool_retriever_c = cast(ObjectRetriever[BaseTool], tool_retriever)
            self._get_tools = lambda message: tool_retriever_c.retrieve(message)
        else:
            self._get_tools = lambda _: []

    @classmethod
    def from_tools(
        cls,
        tools: Optional[Sequence[BaseTool]] = None,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        llm: Optional[LLM] = None,
        max_iterations: int = 10,
        react_chat_formatter: Optional[ReActChatFormatter] = None,
        output_parser: Optional[ReActOutputParser] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        handle_reasoning_failure_fn: Optional[
            Callable[[CallbackManager, Exception], ToolOutput]
        ] = None,
        **kwargs: Any,
    ) -> "ReActAgentWorker":
        """方便的构造方法，用于一组BaseTools（可选）。

注意：此时kwargs应该已经用尽。换句话说，各种上游组件，如BaseSynthesizer（响应合成器）或BaseRetriever应该在它们的构造中从各自的kwargs中取出。

返回：
    ReActAgentWorker
"""
        llm = llm or Settings.llm
        if callback_manager is not None:
            llm.callback_manager = callback_manager
        return cls(
            tools=tools or [],
            tool_retriever=tool_retriever,
            llm=llm,
            max_iterations=max_iterations,
            react_chat_formatter=react_chat_formatter,
            output_parser=output_parser,
            callback_manager=callback_manager,
            verbose=verbose,
            handle_reasoning_failure_fn=handle_reasoning_failure_fn,
        )

    def _get_prompts(self) -> PromptDictType:
        """获取提示。"""
        # TODO: the ReAct formatter does not explicitly specify PromptTemplate
        # objects, but wrap it in this to obey the interface
        sys_header = self._react_chat_formatter.system_header
        return {"system_prompt": PromptTemplate(sys_header)}

    def _update_prompts(self, prompts: PromptDictType) -> None:
        """更新提示。"""
        if "system_prompt" in prompts:
            sys_prompt = cast(PromptTemplate, prompts["system_prompt"])
            self._react_chat_formatter.system_header = sys_prompt.template

    def initialize_step(self, task: Task, **kwargs: Any) -> TaskStep:
        """从任务中初始化步骤。"""
        sources: List[ToolOutput] = []
        current_reasoning: List[BaseReasoningStep] = []
        # temporary memory for new messages
        new_memory = ChatMemoryBuffer.from_defaults()

        # initialize task state
        task_state = {
            "sources": sources,
            "current_reasoning": current_reasoning,
            "new_memory": new_memory,
        }
        task.extra_state.update(task_state)

        return TaskStep(
            task_id=task.task_id,
            step_id=str(uuid.uuid4()),
            input=task.input,
            step_state={"is_first": True},
        )

    def get_tools(self, input: str) -> List[AsyncBaseTool]:
        """获取工具。"""
        return [adapt_to_async_tool(t) for t in self._get_tools(input)]

    def _extract_reasoning_step(
        self, output: ChatResponse, is_streaming: bool = False
    ) -> Tuple[str, List[BaseReasoningStep], bool]:
        """从给定的输出中提取推理步骤。

该方法解析输出的消息内容，提取推理步骤，并确定处理是否完成。它还对输出进行验证检查并处理可能的错误。
"""
        if output.message.content is None:
            raise ValueError("Got empty message.")
        message_content = output.message.content
        current_reasoning = []
        try:
            reasoning_step = self._output_parser.parse(message_content, is_streaming)
        except BaseException as exc:
            raise ValueError(f"Could not parse output: {message_content}") from exc
        if self._verbose:
            print_text(f"{reasoning_step.get_content()}\n", color="pink")
        current_reasoning.append(reasoning_step)

        if reasoning_step.is_done:
            return message_content, current_reasoning, True

        reasoning_step = cast(ActionReasoningStep, reasoning_step)
        if not isinstance(reasoning_step, ActionReasoningStep):
            raise ValueError(f"Expected ActionReasoningStep, got {reasoning_step}")

        return message_content, current_reasoning, False

    def _process_actions(
        self,
        task: Task,
        tools: Sequence[AsyncBaseTool],
        output: ChatResponse,
        is_streaming: bool = False,
    ) -> Tuple[List[BaseReasoningStep], bool]:
        tools_dict: Dict[str, AsyncBaseTool] = {
            tool.metadata.get_name(): tool for tool in tools
        }
        tool = None

        try:
            _, current_reasoning, is_done = self._extract_reasoning_step(
                output, is_streaming
            )
        except ValueError as exp:
            current_reasoning = []
            tool_output = self._handle_reasoning_failure_fn(self.callback_manager, exp)
        else:
            if is_done:
                return current_reasoning, True

            # call tool with input
            reasoning_step = cast(ActionReasoningStep, current_reasoning[-1])
            if reasoning_step.action in tools_dict:
                tool = tools_dict[reasoning_step.action]
                with self.callback_manager.event(
                    CBEventType.FUNCTION_CALL,
                    payload={
                        EventPayload.FUNCTION_CALL: reasoning_step.action_input,
                        EventPayload.TOOL: tool.metadata,
                    },
                ) as event:
                    try:
                        tool_output = tool.call(**reasoning_step.action_input)
                    except Exception as e:
                        tool_output = ToolOutput(
                            content=f"Error: {e!s}",
                            tool_name=tool.metadata.name,
                            raw_input={"kwargs": reasoning_step.action_input},
                            raw_output=e,
                            is_error=True,
                        )
                    event.on_end(
                        payload={EventPayload.FUNCTION_OUTPUT: str(tool_output)}
                    )
            else:
                tool_output = self._handle_nonexistent_tool_name(reasoning_step)

        task.extra_state["sources"].append(tool_output)

        observation_step = ObservationReasoningStep(
            observation=str(tool_output),
            return_direct=(
                tool.metadata.return_direct and not tool_output.is_error
                if tool
                else False
            ),
        )
        current_reasoning.append(observation_step)
        if self._verbose:
            print_text(f"{observation_step.get_content()}\n", color="blue")
        return (
            current_reasoning,
            tool.metadata.return_direct and not tool_output.is_error if tool else False,
        )

    async def _aprocess_actions(
        self,
        task: Task,
        tools: Sequence[AsyncBaseTool],
        output: ChatResponse,
        is_streaming: bool = False,
    ) -> Tuple[List[BaseReasoningStep], bool]:
        tools_dict = {tool.metadata.name: tool for tool in tools}
        tool = None

        try:
            _, current_reasoning, is_done = self._extract_reasoning_step(
                output, is_streaming
            )
        except ValueError as exp:
            current_reasoning = []
            tool_output = self._handle_reasoning_failure_fn(self.callback_manager, exp)
        else:
            if is_done:
                return current_reasoning, True

            # call tool with input
            reasoning_step = cast(ActionReasoningStep, current_reasoning[-1])
            if reasoning_step.action in tools_dict:
                tool = tools_dict[reasoning_step.action]
                with self.callback_manager.event(
                    CBEventType.FUNCTION_CALL,
                    payload={
                        EventPayload.FUNCTION_CALL: reasoning_step.action_input,
                        EventPayload.TOOL: tool.metadata,
                    },
                ) as event:
                    try:
                        tool_output = await tool.acall(**reasoning_step.action_input)
                    except Exception as e:
                        tool_output = ToolOutput(
                            content=f"Error: {e!s}",
                            tool_name=tool.metadata.name,
                            raw_input={"kwargs": reasoning_step.action_input},
                            raw_output=e,
                            is_error=True,
                        )
                    event.on_end(
                        payload={EventPayload.FUNCTION_OUTPUT: str(tool_output)}
                    )
            else:
                tool_output = self._handle_nonexistent_tool_name(reasoning_step)

        task.extra_state["sources"].append(tool_output)

        observation_step = ObservationReasoningStep(
            observation=str(tool_output),
            return_direct=(
                tool.metadata.return_direct and not tool_output.is_error
                if tool
                else False
            ),
        )
        current_reasoning.append(observation_step)
        if self._verbose:
            print_text(f"{observation_step.get_content()}\n", color="blue")
        return (
            current_reasoning,
            tool.metadata.return_direct and not tool_output.is_error if tool else False,
        )

    def _handle_nonexistent_tool_name(self, reasoning_step):
        # We still emit a `tool_output` object to the task, so that the LLM can know
        # it has hallucinated in the next reasoning step.
        with self.callback_manager.event(
            CBEventType.FUNCTION_CALL,
            payload={
                EventPayload.FUNCTION_CALL: reasoning_step.action_input,
            },
        ) as event:
            # TODO(L10N): This should be localized.
            content = f"Error: No such tool named `{reasoning_step.action}`."
            tool_output = ToolOutput(
                content=content,
                tool_name=reasoning_step.action,
                raw_input={"kwargs": reasoning_step.action_input},
                raw_output=content,
                is_error=True,
            )
            event.on_end(payload={EventPayload.FUNCTION_OUTPUT: str(tool_output)})
        return tool_output

    def _get_response(
        self,
        current_reasoning: List[BaseReasoningStep],
        sources: List[ToolOutput],
    ) -> AgentChatResponse:
        """从推理步骤中获取响应。"""
        if len(current_reasoning) == 0:
            raise ValueError("No reasoning steps were taken.")
        elif len(current_reasoning) == self._max_iterations:
            raise ValueError("Reached max iterations.")

        if isinstance(current_reasoning[-1], ResponseReasoningStep):
            response_step = cast(ResponseReasoningStep, current_reasoning[-1])
            response_str = response_step.response
        elif (
            isinstance(current_reasoning[-1], ObservationReasoningStep)
            and current_reasoning[-1].return_direct
        ):
            response_str = current_reasoning[-1].observation
        else:
            response_str = current_reasoning[-1].get_content()

        # TODO: add sources from reasoning steps
        return AgentChatResponse(response=response_str, sources=sources)

    def _get_task_step_response(
        self, agent_response: AGENT_CHAT_RESPONSE_TYPE, step: TaskStep, is_done: bool
    ) -> TaskStepOutput:
        """获取任务步骤响应。"""
        if is_done:
            new_steps = []
        else:
            new_steps = [
                step.get_next_step(
                    step_id=str(uuid.uuid4()),
                    # NOTE: input is unused
                    input=None,
                )
            ]

        return TaskStepOutput(
            output=agent_response,
            task_step=step,
            is_last=is_done,
            next_steps=new_steps,
        )

    def _infer_stream_chunk_is_final(self, chunk: ChatResponse) -> bool:
        """推断从实时流中的一个块是否是最终推理步骤的开始。（即，最终应该成为ResponseReasoningStep ——不过这个函数的逻辑不包括在内）。

Args:
    chunk（ChatResponse）：要检查的当前块流

Returns:
    布尔值：块是否是最终响应的开始
"""
        latest_content = chunk.message.content
        if latest_content:
            # doesn't follow thought-action format
            if len(latest_content) > len("Thought") and not latest_content.startswith(
                "Thought"
            ):
                return True
            elif "Answer: " in latest_content:
                return True
        return False

    def _add_back_chunk_to_stream(
        self, chunk: ChatResponse, chat_stream: Generator[ChatResponse, None, None]
    ) -> Generator[ChatResponse, None, None]:
        """辅助方法，用于将最终响应的初始块流添加回到剩余的聊天流中。

Args:
    chunk（ChatResponse）：要添加回到chat_stream开头的块。

Returns:
    生成器[ChatResponse，None，None]：更新后的chat_stream
"""

        def gen() -> Generator[ChatResponse, None, None]:
            yield chunk
            yield from chat_stream

        return gen()

    async def _async_add_back_chunk_to_stream(
        self, chunk: ChatResponse, chat_stream: AsyncGenerator[ChatResponse, None]
    ) -> AsyncGenerator[ChatResponse, None]:
        """用于将最终响应的初始块流添加回到剩余的chat_stream 中的辅助方法。

注意：这本身不是一个异步函数。

Args:
    chunk（ChatResponse）：要添加回到chat_stream 开头的块。

Returns:
    AsyncGenerator[ChatResponse, None]：更新后的异步chat_stream
"""
        yield chunk
        async for item in chat_stream:
            yield item

    def _run_step(
        self,
        step: TaskStep,
        task: Task,
    ) -> TaskStepOutput:
        """运行步骤。"""
        if step.input is not None:
            add_user_step_to_reasoning(
                step,
                task.extra_state["new_memory"],
                task.extra_state["current_reasoning"],
                verbose=self._verbose,
            )
        # TODO: see if we want to do step-based inputs
        tools = self.get_tools(task.input)
        input_chat = self._react_chat_formatter.format(
            tools,
            chat_history=task.memory.get(input=task.input)
            + task.extra_state["new_memory"].get_all(),
            current_reasoning=task.extra_state["current_reasoning"],
        )

        # send prompt
        chat_response = self._llm.chat(input_chat)
        # given react prompt outputs, call tools or return response
        reasoning_steps, is_done = self._process_actions(
            task, tools, output=chat_response
        )
        task.extra_state["current_reasoning"].extend(reasoning_steps)
        agent_response = self._get_response(
            task.extra_state["current_reasoning"], task.extra_state["sources"]
        )
        if is_done:
            task.extra_state["new_memory"].put(
                ChatMessage(content=agent_response.response, role=MessageRole.ASSISTANT)
            )

        return self._get_task_step_response(agent_response, step, is_done)

    async def _arun_step(
        self,
        step: TaskStep,
        task: Task,
    ) -> TaskStepOutput:
        """运行步骤。"""
        if step.input is not None:
            add_user_step_to_reasoning(
                step,
                task.extra_state["new_memory"],
                task.extra_state["current_reasoning"],
                verbose=self._verbose,
            )
        # TODO: see if we want to do step-based inputs
        tools = self.get_tools(task.input)

        input_chat = self._react_chat_formatter.format(
            tools,
            chat_history=task.memory.get(input=task.input)
            + task.extra_state["new_memory"].get_all(),
            current_reasoning=task.extra_state["current_reasoning"],
        )
        # send prompt
        chat_response = await self._llm.achat(input_chat)
        # given react prompt outputs, call tools or return response
        reasoning_steps, is_done = await self._aprocess_actions(
            task, tools, output=chat_response
        )
        task.extra_state["current_reasoning"].extend(reasoning_steps)
        agent_response = self._get_response(
            task.extra_state["current_reasoning"], task.extra_state["sources"]
        )
        if is_done:
            task.extra_state["new_memory"].put(
                ChatMessage(content=agent_response.response, role=MessageRole.ASSISTANT)
            )

        return self._get_task_step_response(agent_response, step, is_done)

    def _run_step_stream(
        self,
        step: TaskStep,
        task: Task,
    ) -> TaskStepOutput:
        """运行步骤。"""
        if step.input is not None:
            add_user_step_to_reasoning(
                step,
                task.extra_state["new_memory"],
                task.extra_state["current_reasoning"],
                verbose=self._verbose,
            )
        # TODO: see if we want to do step-based inputs
        tools = self.get_tools(task.input)

        input_chat = self._react_chat_formatter.format(
            tools,
            chat_history=task.memory.get(input=task.input)
            + task.extra_state["new_memory"].get_all(),
            current_reasoning=task.extra_state["current_reasoning"],
        )

        chat_stream = self._llm.stream_chat(input_chat)

        # iterate over stream, break out if is final answer after the "Answer: "
        full_response = ChatResponse(
            message=ChatMessage(content=None, role="assistant")
        )
        is_done = False
        for latest_chunk in chat_stream:
            full_response = latest_chunk
            is_done = self._infer_stream_chunk_is_final(latest_chunk)
            if is_done:
                break

        if not is_done:
            # given react prompt outputs, call tools or return response
            reasoning_steps, is_done = self._process_actions(
                task, tools=tools, output=full_response, is_streaming=True
            )
            task.extra_state["current_reasoning"].extend(reasoning_steps)
            # use _get_response to return intermediate response
            agent_response: AGENT_CHAT_RESPONSE_TYPE = self._get_response(
                task.extra_state["current_reasoning"], task.extra_state["sources"]
            )
            if is_done:
                agent_response.is_dummy_stream = True
                task.extra_state["new_memory"].put(
                    ChatMessage(
                        content=agent_response.response, role=MessageRole.ASSISTANT
                    )
                )
        else:
            # Get the response in a separate thread so we can yield the response
            response_stream = self._add_back_chunk_to_stream(
                chunk=latest_chunk, chat_stream=chat_stream
            )

            agent_response = StreamingAgentChatResponse(
                chat_stream=response_stream,
                sources=task.extra_state["sources"],
            )
            thread = Thread(
                target=agent_response.write_response_to_history,
                args=(task.extra_state["new_memory"],),
                kwargs={"on_stream_end_fn": partial(self.finalize_task, task)},
            )
            thread.start()

        return self._get_task_step_response(agent_response, step, is_done)

    async def _arun_step_stream(
        self,
        step: TaskStep,
        task: Task,
    ) -> TaskStepOutput:
        """运行步骤。"""
        if step.input is not None:
            add_user_step_to_reasoning(
                step,
                task.extra_state["new_memory"],
                task.extra_state["current_reasoning"],
                verbose=self._verbose,
            )
        # TODO: see if we want to do step-based inputs
        tools = self.get_tools(task.input)

        input_chat = self._react_chat_formatter.format(
            tools,
            chat_history=task.memory.get(input=task.input)
            + task.extra_state["new_memory"].get_all(),
            current_reasoning=task.extra_state["current_reasoning"],
        )

        chat_stream = await self._llm.astream_chat(input_chat)

        # iterate over stream, break out if is final answer after the "Answer: "
        full_response = ChatResponse(
            message=ChatMessage(content=None, role="assistant")
        )
        is_done = False
        async for latest_chunk in chat_stream:
            full_response = latest_chunk
            is_done = self._infer_stream_chunk_is_final(latest_chunk)
            if is_done:
                break

        if not is_done:
            # given react prompt outputs, call tools or return response
            reasoning_steps, is_done = await self._aprocess_actions(
                task, tools=tools, output=full_response, is_streaming=True
            )
            task.extra_state["current_reasoning"].extend(reasoning_steps)
            # use _get_response to return intermediate response
            agent_response: AGENT_CHAT_RESPONSE_TYPE = self._get_response(
                task.extra_state["current_reasoning"], task.extra_state["sources"]
            )

            if is_done:
                agent_response.is_dummy_stream = True
                task.extra_state["new_memory"].put(
                    ChatMessage(
                        content=agent_response.response, role=MessageRole.ASSISTANT
                    )
                )
        else:
            # Get the response in a separate thread so we can yield the response
            response_stream = self._async_add_back_chunk_to_stream(
                chunk=latest_chunk, chat_stream=chat_stream
            )

            agent_response = StreamingAgentChatResponse(
                achat_stream=response_stream,
                sources=task.extra_state["sources"],
            )
            # create task to write chat response to history
            asyncio.create_task(
                agent_response.awrite_response_to_history(
                    task.extra_state["new_memory"],
                    on_stream_end_fn=partial(self.finalize_task, task),
                )
            )
            # wait until response writing is done
            agent_response._ensure_async_setup()

            await agent_response.is_function_false_event.wait()

        return self._get_task_step_response(agent_response, step, is_done)

    @trace_method("run_step")
    def run_step(self, step: TaskStep, task: Task, **kwargs: Any) -> TaskStepOutput:
        """运行步骤。"""
        return self._run_step(step, task)

    @trace_method("run_step")
    async def arun_step(
        self, step: TaskStep, task: Task, **kwargs: Any
    ) -> TaskStepOutput:
        """运行步骤（异步）。"""
        return await self._arun_step(step, task)

    @trace_method("run_step")
    def stream_step(self, step: TaskStep, task: Task, **kwargs: Any) -> TaskStepOutput:
        """运行步骤（流式）。"""
        # TODO: figure out if we need a different type for TaskStepOutput
        return self._run_step_stream(step, task)

    @trace_method("run_step")
    async def astream_step(
        self, step: TaskStep, task: Task, **kwargs: Any
    ) -> TaskStepOutput:
        """运行步骤（异步流）。"""
        return await self._arun_step_stream(step, task)

    def finalize_task(self, task: Task, **kwargs: Any) -> None:
        """完成任务，在所有步骤都完成之后。"""
        # add new messages to memory
        task.memory.set(
            task.memory.get_all() + task.extra_state["new_memory"].get_all()
        )
        # reset new memory
        task.extra_state["new_memory"].reset()

    def set_callback_manager(self, callback_manager: CallbackManager) -> None:
        """设置回调管理器。"""
        # TODO: make this abstractmethod (right now will break some agent impls)
        self.callback_manager = callback_manager

from_tools classmethod

from_tools(
    tools: Optional[Sequence[BaseTool]] = None,
    tool_retriever: Optional[
        ObjectRetriever[BaseTool]
    ] = None,
    llm: Optional[LLM] = None,
    max_iterations: int = 10,
    react_chat_formatter: Optional[
        ReActChatFormatter
    ] = None,
    output_parser: Optional[ReActOutputParser] = None,
    callback_manager: Optional[CallbackManager] = None,
    verbose: bool = False,
    handle_reasoning_failure_fn: Optional[
        Callable[[CallbackManager, Exception], ToolOutput]
    ] = None,
    **kwargs: Any
) -> ReActAgentWorker

方便的构造方法，用于一组BaseTools（可选）。

注意：此时kwargs应该已经用尽。换句话说，各种上游组件，如BaseSynthesizer（响应合成器）或BaseRetriever应该在它们的构造中从各自的kwargs中取出。

返回： ReActAgentWorker

Source code in llama_index/core/agent/react/step.py

    @classmethod
    def from_tools(
        cls,
        tools: Optional[Sequence[BaseTool]] = None,
        tool_retriever: Optional[ObjectRetriever[BaseTool]] = None,
        llm: Optional[LLM] = None,
        max_iterations: int = 10,
        react_chat_formatter: Optional[ReActChatFormatter] = None,
        output_parser: Optional[ReActOutputParser] = None,
        callback_manager: Optional[CallbackManager] = None,
        verbose: bool = False,
        handle_reasoning_failure_fn: Optional[
            Callable[[CallbackManager, Exception], ToolOutput]
        ] = None,
        **kwargs: Any,
    ) -> "ReActAgentWorker":
        """方便的构造方法，用于一组BaseTools（可选）。

注意：此时kwargs应该已经用尽。换句话说，各种上游组件，如BaseSynthesizer（响应合成器）或BaseRetriever应该在它们的构造中从各自的kwargs中取出。

返回：
    ReActAgentWorker
"""
        llm = llm or Settings.llm
        if callback_manager is not None:
            llm.callback_manager = callback_manager
        return cls(
            tools=tools or [],
            tool_retriever=tool_retriever,
            llm=llm,
            max_iterations=max_iterations,
            react_chat_formatter=react_chat_formatter,
            output_parser=output_parser,
            callback_manager=callback_manager,
            verbose=verbose,
            handle_reasoning_failure_fn=handle_reasoning_failure_fn,
        )

initialize_step

initialize_step(task: Task, **kwargs: Any) -> TaskStep

从任务中初始化步骤。

Source code in llama_index/core/agent/react/step.py

def initialize_step(self, task: Task, **kwargs: Any) -> TaskStep:
    """从任务中初始化步骤。"""
    sources: List[ToolOutput] = []
    current_reasoning: List[BaseReasoningStep] = []
    # temporary memory for new messages
    new_memory = ChatMemoryBuffer.from_defaults()

    # initialize task state
    task_state = {
        "sources": sources,
        "current_reasoning": current_reasoning,
        "new_memory": new_memory,
    }
    task.extra_state.update(task_state)

    return TaskStep(
        task_id=task.task_id,
        step_id=str(uuid.uuid4()),
        input=task.input,
        step_state={"is_first": True},
    )

get_tools

get_tools(input: str) -> List[AsyncBaseTool]

获取工具。

Source code in llama_index/core/agent/react/step.py

def get_tools(self, input: str) -> List[AsyncBaseTool]:
    """获取工具。"""
    return [adapt_to_async_tool(t) for t in self._get_tools(input)]

run_step

run_step(
    step: TaskStep, task: Task, **kwargs: Any
) -> TaskStepOutput

运行步骤。

Source code in llama_index/core/agent/react/step.py

@trace_method("run_step")
def run_step(self, step: TaskStep, task: Task, **kwargs: Any) -> TaskStepOutput:
    """运行步骤。"""
    return self._run_step(step, task)

arun_step async

arun_step(
    step: TaskStep, task: Task, **kwargs: Any
) -> TaskStepOutput

运行步骤（异步）。

Source code in llama_index/core/agent/react/step.py

@trace_method("run_step")
async def arun_step(
    self, step: TaskStep, task: Task, **kwargs: Any
) -> TaskStepOutput:
    """运行步骤（异步）。"""
    return await self._arun_step(step, task)

stream_step

stream_step(
    step: TaskStep, task: Task, **kwargs: Any
) -> TaskStepOutput

运行步骤（流式）。

Source code in llama_index/core/agent/react/step.py

@trace_method("run_step")
def stream_step(self, step: TaskStep, task: Task, **kwargs: Any) -> TaskStepOutput:
    """运行步骤（流式）。"""
    # TODO: figure out if we need a different type for TaskStepOutput
    return self._run_step_stream(step, task)

astream_step async

astream_step(
    step: TaskStep, task: Task, **kwargs: Any
) -> TaskStepOutput

运行步骤（异步流）。

Source code in llama_index/core/agent/react/step.py

@trace_method("run_step")
async def astream_step(
    self, step: TaskStep, task: Task, **kwargs: Any
) -> TaskStepOutput:
    """运行步骤（异步流）。"""
    return await self._arun_step_stream(step, task)

finalize_task

finalize_task(task: Task, **kwargs: Any) -> None

完成任务，在所有步骤都完成之后。

Source code in llama_index/core/agent/react/step.py

def finalize_task(self, task: Task, **kwargs: Any) -> None:
    """完成任务，在所有步骤都完成之后。"""
    # add new messages to memory
    task.memory.set(
        task.memory.get_all() + task.extra_state["new_memory"].get_all()
    )
    # reset new memory
    task.extra_state["new_memory"].reset()

set_callback_manager

set_callback_manager(
    callback_manager: CallbackManager,
) -> None

设置回调管理器。

Source code in llama_index/core/agent/react/step.py

def set_callback_manager(self, callback_manager: CallbackManager) -> None:
    """设置回调管理器。"""
    # TODO: make this abstractmethod (right now will break some agent impls)
    self.callback_manager = callback_manager

ReActChatFormatter

Bases: BaseAgentChatFormatter

ReAct聊天格式化程序。

Source code in llama_index/core/agent/react/formatter.py

class ReActChatFormatter(BaseAgentChatFormatter):
    """ReAct聊天格式化程序。"""

    system_header: str = REACT_CHAT_SYSTEM_HEADER  # default
    context: str = ""  # not needed w/ default

    def format(
        self,
        tools: Sequence[BaseTool],
        chat_history: List[ChatMessage],
        current_reasoning: Optional[List[BaseReasoningStep]] = None,
    ) -> List[ChatMessage]:
        """将聊天记录格式化为ChatMessage列表。"""
        current_reasoning = current_reasoning or []

        format_args = {
            "tool_desc": "\n".join(get_react_tool_descriptions(tools)),
            "tool_names": ", ".join([tool.metadata.get_name() for tool in tools]),
        }
        if self.context:
            format_args["context"] = self.context

        fmt_sys_header = self.system_header.format(**format_args)

        # format reasoning history as alternating user and assistant messages
        # where the assistant messages are thoughts and actions and the user
        # messages are observations
        reasoning_history = []
        for reasoning_step in current_reasoning:
            if isinstance(reasoning_step, ObservationReasoningStep):
                message = ChatMessage(
                    role=MessageRole.USER,
                    content=reasoning_step.get_content(),
                )
            else:
                message = ChatMessage(
                    role=MessageRole.ASSISTANT,
                    content=reasoning_step.get_content(),
                )
            reasoning_history.append(message)

        return [
            ChatMessage(role=MessageRole.SYSTEM, content=fmt_sys_header),
            *chat_history,
            *reasoning_history,
        ]

    @classmethod
    def from_defaults(
        cls,
        system_header: Optional[str] = None,
        context: Optional[str] = None,
    ) -> "ReActChatFormatter":
        """从默认值创建ReActChatFormatter。"""
        if not system_header:
            system_header = (
                REACT_CHAT_SYSTEM_HEADER
                if not context
                else CONTEXT_REACT_CHAT_SYSTEM_HEADER
            )

        return ReActChatFormatter(
            system_header=system_header,
            context=context or "",
        )

    @classmethod
    def from_context(cls, context: str) -> "ReActChatFormatter":
        """从上下文中创建 ReActChatFormatter。

注意：已弃用
"""
        logger.warning(
            "ReActChatFormatter.from_context is deprecated, please use `from_defaults` instead."
        )
        return ReActChatFormatter.from_defaults(
            system_header=CONTEXT_REACT_CHAT_SYSTEM_HEADER, context=context
        )

format

format(
    tools: Sequence[BaseTool],
    chat_history: List[ChatMessage],
    current_reasoning: Optional[
        List[BaseReasoningStep]
    ] = None,
) -> List[ChatMessage]

将聊天记录格式化为ChatMessage列表。

Source code in llama_index/core/agent/react/formatter.py

def format(
    self,
    tools: Sequence[BaseTool],
    chat_history: List[ChatMessage],
    current_reasoning: Optional[List[BaseReasoningStep]] = None,
) -> List[ChatMessage]:
    """将聊天记录格式化为ChatMessage列表。"""
    current_reasoning = current_reasoning or []

    format_args = {
        "tool_desc": "\n".join(get_react_tool_descriptions(tools)),
        "tool_names": ", ".join([tool.metadata.get_name() for tool in tools]),
    }
    if self.context:
        format_args["context"] = self.context

    fmt_sys_header = self.system_header.format(**format_args)

    # format reasoning history as alternating user and assistant messages
    # where the assistant messages are thoughts and actions and the user
    # messages are observations
    reasoning_history = []
    for reasoning_step in current_reasoning:
        if isinstance(reasoning_step, ObservationReasoningStep):
            message = ChatMessage(
                role=MessageRole.USER,
                content=reasoning_step.get_content(),
            )
        else:
            message = ChatMessage(
                role=MessageRole.ASSISTANT,
                content=reasoning_step.get_content(),
            )
        reasoning_history.append(message)

    return [
        ChatMessage(role=MessageRole.SYSTEM, content=fmt_sys_header),
        *chat_history,
        *reasoning_history,
    ]

from_defaults classmethod

from_defaults(
    system_header: Optional[str] = None,
    context: Optional[str] = None,
) -> ReActChatFormatter

从默认值创建ReActChatFormatter。

Source code in llama_index/core/agent/react/formatter.py

@classmethod
def from_defaults(
    cls,
    system_header: Optional[str] = None,
    context: Optional[str] = None,
) -> "ReActChatFormatter":
    """从默认值创建ReActChatFormatter。"""
    if not system_header:
        system_header = (
            REACT_CHAT_SYSTEM_HEADER
            if not context
            else CONTEXT_REACT_CHAT_SYSTEM_HEADER
        )

    return ReActChatFormatter(
        system_header=system_header,
        context=context or "",
    )

from_context classmethod

from_context(context: str) -> ReActChatFormatter

从上下文中创建 ReActChatFormatter。

注意：已弃用

Source code in llama_index/core/agent/react/formatter.py

    @classmethod
    def from_context(cls, context: str) -> "ReActChatFormatter":
        """从上下文中创建 ReActChatFormatter。

注意：已弃用
"""
        logger.warning(
            "ReActChatFormatter.from_context is deprecated, please use `from_defaults` instead."
        )
        return ReActChatFormatter.from_defaults(
            system_header=CONTEXT_REACT_CHAT_SYSTEM_HEADER, context=context
        )