创建ReAct智能体¶

AgentScope 提供了开箱即用的 ReAct 智能体 ReActAgent，位于 agentscope.agent 中，可以直接使用。

它同时支持以下功能:

✨ 基础功能
- 支持对 reply, observe, print, _reasoning 和 _acting 函数周围的钩子支持
- 支持结构化输出
✋ 实时调控
- 支持用户中断<./p>
- 支持自定义的中断处理
🛠️ 工具
- 支持同步/异步两种工具函数
- 支持流式工具响应
- 支持有状态工具管理
- 支持并行工具调用
- 支持 MCP 服务器
💾 记忆
- 支持智能体控制的长期记忆管理
- 支持静态长期记忆管理

提示

有关这些功能的更多细节，请参阅智能体部分。在快速入门中，我们将重点放在如何创建一个 ReAct 智能体并运行它。

from agentscope.agent import ReActAgent, AgentBase
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
import asyncio
import os

from agentscope.tool import Toolkit, execute_python_code

创建ReAct智能体¶

为了提高灵活性，ReActAgent 类在其构造函数中公开了以下参数：

Initialization parameters of `ReActAgent` class¶
参数	进一步阅读	描述
`name` (必填)		智能体的名称
`sys_prompt` (required)		智能体的系统提示
`model` (必需)	模型	智能体用于生成响应的模型
`formatter` (必选)	提示格式化器	提示词构建策略，应与模型保持一致。
`工具包`	工具	用于注册/调用工具函数的工具包.
`内存`	内存	用于存储对话历史的短期记忆。
`long_term_memory`	长期记忆	长期记忆
`long_term_memory_mode`	长期记忆	长期记忆的模式： `agent_control`: 允许智能体自行控制长期记忆 `static_control`: 每次回复的开始/结束时将进行长期记忆的检索和记录。 `both`: 同时激活以上两种模式
`enable_meta_tool`	工具	是否启用元工具，使得智能体能够自行管理工具
`parallel_tool_calls`	智能体	是否允许并行工具调用
`max_iters`		智能体生成响应的最大迭代次数

以DashScope API为例, 我们创建一个智能体对象如下:

async def creating_react_agent() -> None:
    """Create a ReAct agent and run a simple task."""
    # Prepare tools
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)

    jarvis = ReActAgent(
        name="Jarvis",
        sys_prompt="You're a helpful assistant named Jarvis",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=True,
            enable_thinking=False,
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
        memory=InMemoryMemory(),
    )

    msg = Msg(
        name="user",
        content="Hi! Jarvis, run Hello World in Python.",
        role="user",
    )

    await jarvis(msg)


asyncio.run(creating_react_agent())

Jarvis: {
    "type": "tool_use",
    "id": "call_b89e982fa75a40dead08e8",
    "name": "execute_python_code",
    "input": {
        "code": "print('Hello World')",
        "timeout": 300
    }
}
system: {
    "type": "tool_result",
    "id": "call_b89e982fa75a40dead08e8",
    "name": "execute_python_code",
    "output": [
        {
            "type": "text",
            "text": "<returncode>0</returncode><stdout>Hello World\n</stdout><stderr></stderr>"
        }
    ]
}
Jarvis: The Python code has been successfully executed, and it printed:

```
Hello World
```

从零开始创建¶

你可能希望从头开始创建一个智能体，AgentScope为你提供了两个可供继承的基础类：

类	抽象方法	描述
`AgentBase`	`reply` `observe` `handle_interrupt`	所有智能体的基类，支持环绕`reply`、`observe`和`print`函数的前置与后置钩子。在`__call__`方法中实现实时指导。
`ReActAgentBase`	`reply` `observe` `handle_interrupt` `_reasoning` `_acting`	在`AgentBase`的基础上添加两个抽象函数`_reasoning`和`_acting`，以及它们的钩子。

请参考智能体部分了解更多关于agent类别的详情信息。

以AgentBase类为例，我们可以通过继承它并实现reply方法来创建自定义智能体类。

class MyAgent(AgentBase):
    """A custom agent class"""

    def __init__(self) -> None:
        """Initialize the agent"""
        super().__init__()

        self.name = "Friday"
        self.sys_prompt = "You're a helpful assistant named Friday."
        self.model = DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=False,
        )
        self.formatter = DashScopeChatFormatter()
        self.memory = InMemoryMemory()

    async def reply(self, msg: Msg | list[Msg] | None) -> Msg:
        """Reply to the message."""
        await self.memory.add(msg)

        # Prepare the prompt
        prompt = await self.formatter.format(
            [
                Msg("system", self.sys_prompt, "system"),
                *await self.memory.get_memory(),
            ],
        )

        # Call the model
        response = await self.model(prompt)

        msg = Msg(
            name=self.name,
            content=response.content,
            role="assistant",
        )

        # Record the response in memory
        await self.memory.add(msg)

        # Print the message
        await self.print(msg)
        return msg

    async def observe(self, msg: Msg | list[Msg] | None) -> None:
        """Observe the message."""
        # Store the message in memory
        await self.memory.add(msg)

    async def handle_interrupt(self) -> Msg:
        """Postprocess the interrupt."""
        # Taking a fixed response as example
        return Msg(
            name=self.name,
            content="I noticed you interrupted me, how can I help you?",
            role="assistant",
        )


async def run_custom_agent() -> None:
    """Run the custom agent."""
    agent = MyAgent()
    msg = Msg(
        name="user",
        content="Who are you?",
        role="user",
    )
    await agent(msg)


asyncio.run(run_custom_agent())

Friday: Hello! I'm Friday, an AI assistant designed to help with a wide range of tasks and provide information on various topics. How can I assist you today?

扩展阅读¶

脚本的总运行时间：（0分钟9.584秒）

Gallery generated by Sphinx-Gallery