from agentscope.message import Msg

msg = Msg(
    name="游戏主持人",
    role="system",
    content="游戏开始！",
)

from agentscope.message import (
    Msg,
    Base64Source,
    TextBlock,
    ThinkingBlock,
    ImageBlock,
    AudioBlock,
    VideoBlock,
    ToolUseBlock,
    ToolResultBlock,
)

msg = Msg(
    name="Jarvis",
    role="assistant",
    content=[
        TextBlock(
            type="text",
            text="这是一个包含 base64 编码数据的多模态消息。",
        ),
        ImageBlock(
            type="image",
            source=Base64Source(
                type="base64",
                media_type="image/jpeg",
                data="/9j/4AAQSkZ...",
            ),
        ),
        AudioBlock(
            type="audio",
            source=Base64Source(
                type="base64",
                media_type="audio/mpeg",
                data="SUQzBAAAAA...",
            ),
        ),
        VideoBlock(
            type="video",
            source=Base64Source(
                type="base64",
                media_type="video/mp4",
                data="AAAAIGZ0eX...",
            ),
        ),
    ],
)

TextBlock(
   type="text",
   text="Hello, world!"
)

# 普通示例
ImageBlock(
   type="image",
   source=URLSource(
       type="url",
       url="https://example.com/image.jpg"
   )
)
# base64示例
ImageBlock(
    type="image",
    source=Base64Source(
        type="base64",
        media_type="image/jpeg",
        data="/9j/4AAQSkZ...",
    ),
)

# 普通示例
AudioBlock(
   type="audio",
   source=URLSource(
       type="url",
       url="https://example.com/audio.mp3"
   )
)
# base64示例
AudioBlock(
    type="audio",
    source=Base64Source(
        type="base64",
        media_type="audio/mpeg",
        data="SUQzBAAAAA...",
    ),
)

VideoBlock(
   type="video",
   source=URLSource(
       type="url",
       url="https://example.com/video.mp4"
   )
)

msg_thinking = Msg(
    name="Jarvis",
    role="assistant",
    content=[
        ThinkingBlock(
            type="thinking",
            thinking="我正在为 AgentScope 构建一个思考块的示例。",
        ),
        TextBlock(
            type="text",
            text="这是一个思考块的示例。",
        ),
    ],
)

msg_tool_call = Msg(
    name="Jarvis",
    role="assistant",
    content=[
        ToolUseBlock(
            type="tool_use",
            id="343",
            name="get_weather",
            input={
                "location": "Beijing",
            },
        ),
    ],
)

msg_tool_res = Msg(
    name="system",
    role="system",
    content=[
        ToolResultBlock(
            type="tool_result",
            id="343",
            name="get_weather",
            output="北京的天气是晴天，温度为 25°C。",
        ),
    ],
)

serialized_msg = msg.to_dict()

print(type(serialized_msg))
print(json.dumps(serialized_msg, indent=4, ensure_ascii=False))

<class 'dict'>
{
    "id": "gH8zDp9YSfLSKyGppA2QkL",
    "name": "Jarvis",
    "role": "assistant",
    "content": [
        {
            "type": "text",
            "text": "这是一个包含 base64 编码数据的多模态消息。"
        },
        {
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": "image/jpeg",
                "data": "/9j/4AAQSkZ..."
            }
        },
        {
            "type": "audio",
            "source": {
                "type": "base64",
                "media_type": "audio/mpeg",
                "data": "SUQzBAAAAA..."
            }
        },
        {
            "type": "video",
            "source": {
                "type": "base64",
                "media_type": "video/mp4",
                "data": "AAAAIGZ0eX..."
            }
        }
    ],
    "metadata": {},
    "timestamp": "2026-03-12 03:27:29.431"
}

new_msg = Msg.from_dict(serialized_msg)

print(type(new_msg))
print(f'消息的发送者: "{new_msg.name}"')
print(f'发送者的角色: "{new_msg.role}"')
print(f'消息的内容: "{json.dumps(new_msg.content, indent=4, ensure_ascii=False)}"')

<class 'agentscope.message._message_base.Msg'>
消息的发送者: "Jarvis"
发送者的角色: "assistant"
消息的内容: "[
    {
        "type": "text",
        "text": "这是一个包含 base64 编码数据的多模态消息。"
    },
    {
        "type": "image",
        "source": {
            "type": "base64",
            "media_type": "image/jpeg",
            "data": "/9j/4AAQSkZ..."
        }
    },
    {
        "type": "audio",
        "source": {
            "type": "base64",
            "media_type": "audio/mpeg",
            "data": "SUQzBAAAAA..."
        }
    },
    {
        "type": "video",
        "source": {
            "type": "base64",
            "media_type": "video/mp4",
            "data": "AAAAIGZ0eX..."
        }
    }
]"

def tool_function(a: int, b: str) -> ToolResponse:
    """{函数描述}

    Args:
        a (int):
            {第一个参数的描述}
        b (str):
            {第二个参数的描述}
    """

# 准备一个自定义工具函数
async def my_search(query: str, api_key: str) -> ToolResponse:
    """一个简单的示例工具函数。

    Args:
        query (str):
            搜索查询。
        api_key (str):
            用于身份验证的 API 密钥。
    """
    return ToolResponse(
        content=[
            TextBlock(
                type="text",
                text=f"正在使用 API 密钥 '{api_key}' 搜索 '{query}'",
            ),
        ],
    )


# 在工具模块中注册工具函数
toolkit = Toolkit()
toolkit.register_tool_function(my_search)

print("工具 JSON Schemas：")
print(json.dumps(toolkit.get_json_schemas(), indent=4, ensure_ascii=False))

工具 JSON Schemas：
[
    {
        "type": "function",
        "function": {
            "name": "my_search",
            "parameters": {
                "properties": {
                    "query": {
                        "description": "搜索查询。",
                        "type": "string"
                    },
                    "api_key": {
                        "description": "用于身份验证的 API 密钥。",
                        "type": "string"
                    }
                },
                "required": [
                    "query",
                    "api_key"
                ],
                "type": "object"
            },
            "description": "一个简单的示例工具函数。"
        }
    }
]

import asyncio
import os

from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit, execute_python_code


async def creating_react_agent() -> None:
    """创建一个 ReAct 智能体并运行一个简单任务。"""
    # 准备工具
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)  # 内置工具函数execute_python_code

    jarvis = ReActAgent(
        name="Jarvis",
        sys_prompt="你是一个名为 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="你好！Jarvis，用 Python 运行 Hello World。",
        role="user",
    )

    await jarvis(msg)


asyncio.run(creating_react_agent())

def 前置钩子名(self, kwargs: dict[str, Any]) -> dict[str, Any] | None:
    # 你的自定义逻辑
    pass

def pre_reply_filter(self, kwargs: dict[str, Any]) -> dict[str, Any] | None:
    # 拿到用户输入的消息
    user_msg = kwargs["msg"]
    # 自定义逻辑：敏感词替换
    user_msg.content = user_msg.content.replace("脏话", "**")
    # 返回修改后的参数
    return {**kwargs, "msg": user_msg}

def 后置钩子名(self, kwargs: dict[str, Any], output: Any) -> Any | None:
    # 你的自定义逻辑
    pass

def post_reply_add_disclaimer(self, kwargs: dict[str, Any], output: Any) -> Any | None:
    # 拿到智能体的原始回复
    reply_msg = output
    # 加免责声明
    reply_msg.content += "\n\n以上内容仅为AI生成，不构成任何建议"
    # 返回修改后的输出
    return reply_msg

调用核心函数（比如reply）
↓
【前置钩子阶段】先执行实例级前置钩子（按注册顺序，先注册先执行）
↓
【前置钩子阶段】再执行类级前置钩子（按注册顺序，先注册先执行）
↓
执行核心函数本身的逻辑
↓
【后置钩子阶段】先执行实例级后置钩子（按注册顺序，先注册先执行）
↓
【后置钩子阶段】再执行类级后置钩子（按注册顺序，先注册先执行）
↓
返回最终结果

async def in_memory_example():
    """使用InMemoryMemory在内存中存储消息的示例。"""
    memory = InMemoryMemory()
    await memory.add(
        Msg("Alice", "生成一份关于AgentScope的报告", "user"),
    )

    # 添加一条带有标记"hint"的提示消息
    await memory.add(
        [
            Msg(
                "system",
                "<system-hint>首先创建一个计划来收集信息，然后逐步生成报告。</system-hint>",
                "system",
            ),
        ],
        marks="hint",   # 标记一次性提示消息
    )

    msgs = await memory.get_memory(mark="hint")
    print("带有标记'hint'的消息：")
    for msg in msgs:
        print(f"- {msg}")

    # 所有存储的消息都可以通过 ``state_dict`` 和 ``load_state_dict`` 方法导出和加载。
    state = memory.state_dict()
    print("记忆的状态字典：")
    print(json.dumps(state, indent=2, ensure_ascii=False))

    # 通过标记删除消息
    deleted_count = await memory.delete_by_mark("hint")
    print(f"删除了 {deleted_count} 条带有标记'hint'的消息。")

    print("删除后的记忆状态字典：")
    state = memory.state_dict()
    print(json.dumps(state, indent=2, ensure_ascii=False))


asyncio.run(in_memory_example())

带有标记'hint'的消息：
- Msg(id='XqLgHiuTEPEYtN3pM5EUDV', name='system', content='<system-hint>首先创建一个计划来收集信息，然后逐步生成报告。</system-hint>', role='system', metadata={}, timestamp='2026-03-12 03:30:48.641', invocation_id='None')
记忆的状态字典：
{
  "_compressed_summary": "",
  "content": [
    [
      {
        "id": "Agmg28QmfUuyqBcc2sMucP",
        "name": "Alice",
        "role": "user",
        "content": "生成一份关于AgentScope的报告",
        "metadata": {},
        "timestamp": "2026-03-12 03:30:48.641"
      },
      []
    ],
    [
      {
        "id": "XqLgHiuTEPEYtN3pM5EUDV",
        "name": "system",
        "role": "system",
        "content": "<system-hint>首先创建一个计划来收集信息，然后逐步生成报告。</system-hint>",
        "metadata": {},
        "timestamp": "2026-03-12 03:30:48.641"
      },
      [
        "hint"
      ]
    ]
  ]
}
删除了 1 条带有标记'hint'的消息。
删除后的记忆状态字典：
{
  "_compressed_summary": "",
  "content": [
    [
      {
        "id": "Agmg28QmfUuyqBcc2sMucP",
        "name": "Alice",
        "role": "user",
        "content": "生成一份关于AgentScope的报告",
        "metadata": {},
        "timestamp": "2026-03-12 03:30:48.641"
      },
      []
    ]
  ]
}

async def sqlalchemy_example() -> None:
    """使用 AsyncSQLAlchemyMemory 在 SQLite 数据库中存储消息的示例。"""

    # 首先创建一个异步 SQLAlchemy 引擎
    engine = create_async_engine("sqlite+aiosqlite:///./test_memory.db")

    # 然后使用该引擎创建记忆
    memory = AsyncSQLAlchemyMemory(
        engine_or_session=engine,
        # 可选传入指定user_id和session_id
        user_id="user_1",
        session_id="session_1",
    )

    await memory.add(
        Msg("Alice", "生成一份关于AgentScope的报告", "user"),
    )

    await memory.add(
        [
            Msg(
                "system",
                "<system-hint>首先创建一个计划来收集信息，然后逐步生成报告。</system-hint>",
                "system",
            ),
        ],
        marks="hint",
    )

    msgs = await memory.get_memory(mark="hint")
    print("带有标记'hint'的消息：")
    for msg in msgs:
        print(f"- {msg}")

    # 完成后关闭引擎
    await memory.close()


asyncio.run(sqlalchemy_example())

带有标记'hint'的消息：
- Msg(id='BWboEGSoABPMkkseBtoLZD', name='system', content='<system-hint>首先创建一个计划来收集信息，然后逐步生成报告。</system-hint>', role='system', metadata={}, timestamp='2026-03-12 03:30:48.683', invocation_id='None')

async def redis_memory_example() -> None:
    """使用 RedisMemory 在 Redis 中存储消息的示例。"""
    # 使用fakeredis进行内存测试，无需真实的 Redis 服务器
    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    # 创建 Redis 记忆
    memory = RedisMemory(
        # 使用fake redis进行演示
        connection_pool=fake_redis.connection_pool,
        # 也可以通过指定主机和端口连接到真实的Redis服务器
        # host="localhost",
        # port=6379,
        # 可选地指定 user_id 和 session_id
        user_id="user_1",
        session_id="session_1",
    )

    # 向记忆中添加消息
    await memory.add(
        Msg(
            "Alice",
            "生成一份关于AgentScope的报告",
            "user",
        ),
    )

    # 添加一条带有标记"hint"的提示消息
    await memory.add(
        Msg(
            "system",
            "<system-hint>首先创建一个计划来收集信息，然后逐步生成报告。</system-hint>",
            "system",
        ),
        marks="hint",
    )

    # 检索带有标记"hint"的消息
    msgs = await memory.get_memory(mark="hint")
    print("带有标记'hint'的消息：")
    for msg in msgs:
        print(f"- {msg}")


asyncio.run(redis_memory_example())

带有标记'hint'的消息：
- Msg(id='ZkPMW3LtzGnKEq6LEBTDfK', name='system', content='<system-hint>首先创建一个计划来收集信息，然后逐步生成报告。</system-hint>', role='system', metadata={}, timestamp='2026-03-12 03:30:48.710', invocation_id='None')

# 让Alice、Bob、Charlie自我介绍
async def example_broadcast_message():
    """使用 MsgHub 广播消息的示例。"""

    # 创建消息中心
    async with MsgHub(
        participants=[alice, bob, charlie],
        announcement=Msg(
            "user",
            "现在请简要介绍一下自己，包括你的姓名、年龄和职业。",
            "user",
        ),
    ) as hub:
        # 无需手动消息传递的群聊
        await alice()
        await bob()
        await charlie()


asyncio.run(example_broadcast_message())


# 询问Alice之外的其他人是否知道Alice信息
async def check_broadcast_message():
    """检查消息是否正确广播。"""
    user_msg = Msg(
        "user",
        "你知道 Alice 是谁吗，她是做什么的？",
        "user",
    )

    await bob(user_msg)
    await charlie(user_msg)
    await david(user_msg)


asyncio.run(check_broadcast_message())

async with MsgHub(participants=[alice]) as hub:
    # 添加新参与者
    hub.add(david)

    # 移除参与者
    hub.delete(alice)

    # 向所有当前参与者广播
    await hub.broadcast(
        Msg("system", "现在我们开始...", "system"),
    )

from agentscope.pipeline import sequential_pipeline

msg = await sequential_pipeline(
    # 按顺序执行的智能体列表
    agents=[alice, bob, charlie, david],
    # 第一个输入消息，可以是 None
    msg=None
)

from agentscope.pipeline import SequentialPipeline

# 先创建管道对象，定义好执行顺序
pipeline = SequentialPipeline(agents=[alice, bob, charlie, david])

# 第一次调用，初始输入为None
msg = await pipeline(msg=None)

# 换个输入，复用同一个管道，不用重新定义顺序
msg2 = await pipeline(msg=Msg("user", "你好！", "user"))

from agentscope.pipeline import fanout_pipeline

msgs = await fanout_pipeline(
    # 要执行的智能体列表
    agents=[alice, bob, charlie, david],
    # 输入消息，可以是 None
    msg=None,
    enable_gather=False,
)

from agentscope.pipeline import FanoutPipeline

# 创建可复用的管道对象
pipeline = FanoutPipeline(agents=[alice, bob, charlie, david])

# 多次调用，换不同的输入
msgs = await pipeline(msg=None)
msgs2 = await pipeline(msg=Msg("user", "你对AI发展怎么看？", "user"))

async def run_example_pipeline() -> None:
    """运行流式打印消息的示例。"""
    agent = create_agent("Alice", 20, "student")

    # 我们关闭agent的终端打印以避免输出混乱
    agent.set_console_output_enabled(False)

    async for msg, last in stream_printing_messages(
        agents=[agent],
        coroutine_task=agent(
            Msg("user", "你好，你是谁？", "user"),
        ),
    ):
        print(msg, last)
        if last:
            print()


asyncio.run(run_example_pipeline())