# 1. 用户的原始请求
user_request = "你好，请帮我搜集一些关于 Transformer 模型的最新资料。"

# 2. “硬编码”执行工具函数
# 这里的 web_search 函数是模拟的，以聚焦于核心逻辑
def web_search(query: str):
    """模拟执行网络搜索并返回JSON格式的结果"""
    print(f"--- [工具执行中] 正在搜索: {query} ---")
    # 真实场景中，这里会调用一个真正的搜索引擎API
    return '''{
        "results": [
            {"title": "Attention Is All You Need (Transformer 论文原文)", "url": "https://arxiv.org/abs/1706.03762", "snippet": "The dominant sequence transduction models are based on complex recurrent or convolutional neural networks... We propose a new simple network architecture, the Transformer, based solely on attention mechanisms..."},
            {"title": "The Illustrated Transformer – Jay Alammar", "url": "https://jalammar.github.io/illustrated-transformer/", "snippet": "A visual and intuitive explanation of the Transformer model."}
        ]
    }'''

tool_result = web_search(query=user_request)

print(f"用户请求: {user_request}")
print(f"工具结果: {tool_result}\n")


# 3. 将用户请求和工具结果拼接成一个提示，发送给大模型
#    目标是让模型基于结构化的工具输出，生成一句人类易于理解的回复。
completion = client.chat.completions.create(
    model="qwen-plus", 
    messages=[
        {'role': 'system', 'content': '你是一位课程研究助理，你的任务是根据工具的执行结果，向用户生成一句友好和清晰的回复。'},
        {'role': 'user', 'content': f'用户原始请求: "{user_request}"\n工具执行结果: {tool_result}'}
    ]
)

# 4. 输出模型的最终回复
final_response = completion.choices[0].message.content
print(f"模型生成的最终回复:\n{final_response}")

# 伪代码：一个脆弱的、基于关键字的工具路由器
def route_to_tool(user_input):
    if "论文" in user_input or "arxiv" in user_input:
        return "search_arxiv_paper"
    elif "搜索" in user_input or "查找" in user_input or "资料" in user_input:
        return "web_search"
    elif "总结" in user_input or "内容" in user_input or "http" in user_input:
        return "fetch_webpage_content"
    # ... 你需要在这里不断添加 elif 判断
    else:
        return "no_tool_needed"

def get_tool_decision_from_llm(user_request):
    from textwrap import dedent

    # 使用 dedent 移除为了代码美观而添加的缩进，
    # 避免这些格式化用的tab/空格被传递给大模型
    prompt = dedent(f"""
        你是一个智能助理的路由模块。你的任务是根据用户的请求，从下面的工具列表中选择最合适的工具来解决问题，并给出入参。

        [可用工具列表]
        1. web_search: 用于在互联网上搜索通用信息。
        2. search_arxiv_paper: 用于在 Arxiv.org 上搜索学术论文。
        3. fetch_webpage_content: 用于获取指定URL的网页内容。

        [用户请求]
        "{user_request}"

        [决策]
        请告诉用户你的决定。
    """)
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=[
            {'role': 'user', 'content': prompt}
        ],
        temperature=0.0 # 使用低 temperature 以获得更确定的决策
    )
    decision = completion.choices[0].message.content
    return decision

# --- 测试用例 ---
request = "帮我找一下那篇经典的 Transformer 论文，标题是 'Attention Is All You Need'"
decision = get_tool_decision_from_llm(request)
print(f"用户请求: \"{request}\"")
print(f"模型决策: {decision}\n")

{
  "tool_name": "search_arxiv_paper",
  "parameters": {
    "query": "Attention Is All You Need"
  }
}

import json
from textwrap import dedent
from typing import Union, Literal
from pydantic import BaseModel, Field, ValidationError, TypeAdapter

# 1. 为与课程研究相关的工具定义 Pydantic 参数模型
class WebSearchParams(BaseModel):
    query: str = Field(description="用于网络搜索的关键词。")

class SearchArxivParams(BaseModel):
    query: str = Field(description="用于在 Arxiv.org 上搜索的论文标题或关键词。")

# 2. 定义工具调用模型，将工具名与对应的参数模型绑定
class WebSearchCall(BaseModel):
    tool_name: Literal["web_search"]
    parameters: WebSearchParams

class SearchArxivCall(BaseModel):
    tool_name: Literal["search_arxiv_paper"]
    parameters: SearchArxivParams

# 使用 Union 类型，表示模型最终的决策是这两种调用中的一种
ToolCall = Union[WebSearchCall, SearchArxivCall]

# 3. 构建提示词，包含清晰的指令、工具定义和示例
def build_prompt(user_request: str) -> str:
    return dedent(f"""
        你的任务是根据用户的请求，从可用工具列表中选择最合适的工具，并以严格的 JSON 格式输出调用信息。

        # 可用工具:
        - `web_search(query: str)`: 当需要搜索通用信息、新闻或非学术性内容时使用。
        - `search_arxiv_paper(query: str)`: 当需要搜索学术论文，特别是来自 Arxiv.org 的论文时使用。

        # 输出格式要求:
        你必须严格按照以下 JSON 结构输出，不要包含任何额外的自然语言解释。

        {{
        "tool_name": "工具名称",
        "parameters": {{
            "参数名": "参数值"
        }}
        }}

        # 示例:
        用户请求: "最近AI领域有什么好玩的新闻？"
        你的输出:

        {{
        "tool_name": "web_search",
        "parameters": {{
            "query": "AI领域最新新闻"
        }}
        }}


        # 用户请求:
        "{user_request}"

        # 你的输出:
    """)

# 4. 调用并验证（带重试）
def get_structured_output(user_request: str, max_retries: int = 2):
    messages = [{'role': 'user', 'content': build_prompt(user_request)}]
    adapter = TypeAdapter(ToolCall)

    for attempt in range(max_retries):
        # 此处为对大模型服务的API调用，为保证课程的通用性，具体实现已省略
        # 你可以替换成自己的代码，例如 client.chat.completions.create(...)
        response = client.chat.completions.create(
            model="qwen-plus", messages=messages, temperature=0
        )
        raw_output = response.choices[0].message.content

        try:
            # 解析并验证
            data = json.loads(raw_output.strip('```json').strip('```'))
            # 使用Pydantic模型验证解析后的数据
            validated_data = adapter.validate_python(data)
            return validated_data.model_dump()
        except (json.JSONDecodeError, ValidationError) as e:
            # 如果失败，则将错误信息和原始输出都加入到对话历史中，以便模型进行修正
            messages.extend([
                {'role': 'assistant', 'content': raw_output},
                {'role': 'user', 'content': f"格式错误: {e}，请严格按照JSON格式重新输出"}
            ])

    return None

# 使用
# 假设 client 变量已在别处初始化
user_request = "帮我找一下那篇经典的 Transformer 论文，标题是 'Attention Is All You Need'"
result = get_structured_output(user_request)

if result:
    print(json.dumps(result, indent=2, ensure_ascii=False))

import json
# 1. 定义工具列表，包含每个函数的JSON Schema描述
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_arxiv_paper",
            "description": "在 Arxiv.org 上搜索学术论文",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "论文的标题或关键词"},
                },
                "required": ["query"],
            },
        }
    }
]
# 2. 发起第一次API调用，让模型决策
messages = [{"role": "user", "content": "帮我找一下那篇经典的 Transformer 论文 'Attention Is All You Need'"}]
response = client.chat.completions.create(
    model="qwen-plus", messages=messages, tools=tools, tool_choice="auto"
)
response_message = response.choices[0].message
# 3. 检查模型是否决定调用工具并执行
if response_message.tool_calls:
    tool_call = response_message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    print(f"模型决定调用工具: `{function_name}`")
    print(f"参数: {function_args}")

    # 这里我们模拟函数执行结果
    tool_result = json.dumps({"paper_id": "1706.03762", "url": "https://arxiv.org/abs/1706.03762", "title": "Attention Is All You Need"})
    print(f"工具执行结果: {tool_result}")
    # 4. 将模型的决策和工具的执行结果一起传回，让模型生成最终答复
    messages.append(response_message)
    messages.append(
        {"tool_call_id": tool_call.id, "role": "tool", "name": function_name, "content": tool_result}
    )

    final_response = client.chat.completions.create(model="qwen-plus", messages=messages)
    print("\n模型的最终回复:")
    print(final_response.choices[0].message.content)

import asyncio
from agentscope.agent import ReActAgent
from agentscope.tool import Toolkit, ToolResponse
from agentscope.model import DashScopeChatModel
from agentscope.message import Msg, TextBlock
from agentscope.formatter import DashScopeChatFormatter

# 1. 定义一个工具，使用 ToolResponse 返回结果
def search_arxiv_paper(query: str) -> ToolResponse:
    """在 Arxiv.org 上搜索学术论文。

    Args:
        query (str): 搜索关键词
    """
    print(f"--- [工具执行中] 正在 Arxiv 搜索: {query} ---")
    # 此处模拟搜索结果
    paper_url = "https://arxiv.org/abs/1706.03762"
    return ToolResponse(
        content=[
            TextBlock(
                type="text",
                text=f"已成功找到论文 '{query}'，你可以在这里访问：{paper_url}",
            )
        ]
    )

async def run_agentscope_example():
    # 2. 将工具函数注册到工具箱(Toolkit)中
    toolkit = Toolkit()
    toolkit.register_tool_function(search_arxiv_paper)

    # 3. 创建一个 ReActAgent 并为其配备工具箱
    agent = ReActAgent(
        name="Course Research Agent",
        sys_prompt="你是一个课程研究助理，擅长帮人搜集和整理学习资料。",
        model=DashScopeChatModel(
            model_name="qwen-plus", 
            api_key=os.environ.get("DASHSCOPE_API_KEY")
        ),
        toolkit=toolkit,
        formatter=DashScopeChatFormatter()
    )

    # 4. 向 Agent 发送消息，它会自动完成所有步骤
    user_request = "帮我找一下那篇经典的 Transformer 论文 'Attention Is All You Need'"
    msg = Msg(name="user", content=user_request, role="user")

    print(f"用户请求: {user_request}\n")
    await agent(msg)

# 运行示例
try:
    # 在Jupyter Notebook环境中，可以直接await协程
    await run_agentscope_example()
except NameError:
    # 在普通Python脚本中，需要使用asyncio.run()来运行异步函数
    asyncio.run(run_agentscope_example())

import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.mcp import HttpStatelessClient
from agentscope.tool import Toolkit
from agentscope.model import DashScopeChatModel
from agentscope.message import Msg
from agentscope.formatter import DashScopeChatFormatter
async def run_mcp_example():
    # 1. 配置 MCP 客户端，指向远程工具服务
    #    这里连接的是阿里云DashScope公开的联网搜索MCP服务
    web_search_client = HttpStatelessClient(
        name="web_search_service", # 为这个客户端起一个名字
        transport="sse",
        url="https://dashscope.aliyuncs.com/api/v1/mcps/WebSearch/sse",
        headers={"Authorization": "Bearer " + os.environ.get("DASHSCOPE_API_KEY")},
    )
    # 2. 将 MCP 客户端注册到工具箱
    #    Agent在启动时会自动通过客户端“发现”远程服务提供的所有工具
    toolkit = Toolkit()
    await toolkit.register_mcp_client(web_search_client)
    # 3. 创建 Agent，并配备包含 MCP 工具的工具箱
    agent = ReActAgent(
        name="Research Assistant Agent",
        sys_prompt="你是一个课程研究助理，擅长使用工具搜集和整理最新的教学素材。",
        model=DashScopeChatModel(
            model_name="qwen-plus", api_key=os.environ.get("DASHSCOPE_API_KEY")
        ),
        toolkit=toolkit,
        formatter=DashScopeChatFormatter()
    )

    # 4. 提出一个需要远程工具才能回答的问题
    user_request = "我正在为'大模型原理'课程搜集素材，需要一个调用外部实时数据的例子，比如帮我搜索一下最近关于'大型语言模型'的最新进展。"
    msg = Msg(name="user", content=user_request, role="user")

    print(f"用户请求: {user_request}\n")
    response_msg = await agent(msg)

try:
    # 在Jupyter Notebook环境中，可以直接await协程
    await run_mcp_example()
except NameError:
    # 在普通Python脚本中，需要使用asyncio.run()来运行异步函数
    asyncio.run(run_mcp_example())

# 课程的原始代码
def get_user_data(usr_id: str):
  # ... some logic ...
  return f"Data for {usr_id}"

# 正常执行
print(get_user_data(usr_id="u-123"))

# Agent 润色后的代码
def get_user_data(user_id: str):
  # ... some logic ...
  return f"Data for {user_id}"

# 运行时报错！
# print(get_user_data(usr_id="u-123")) 
# TypeError: get_user_data() got an unexpected keyword argument 'usr_id'

你是一位顶尖的课程作家，负责润色 Jupyter Notebook 格式的课程。
请优化文本，使其更具吸引力。
**重要：请绝对不要修改任何代码块中的内容，包括变量名、参数名和函数调用。**

## 角色
你是一位顶尖的课程作家。

## 任务
1. 润色课程的语言表达，输出润色后的全文内容。

2. 反思润色后的内容：
    ⁃ 是否符合写作规范。
    ⁃ 除了语言表达，其他内容是否被意外修改。
输出反思的结果和修改建议。

3. 对照建议修改课程，输出修改后的全文内容。

## 课程初稿
{original_notebook_content}

# 第一次调用 (写作 Agent)
你是一位顶尖的课程作家，负责润色 Jupyter Notebook 格式的课程。请优化文本，使其更具吸引力。

【课程初稿】:
{original_notebook_content}

# 第二次调用 (技术审查 Agent)
你是一个严苛的技术审查员。你的任务是审查润色后的课程内容，确保：
1. 润色后的课程符合写作规范。
2. 代码、配置、数据等技术内容没有被意外修改。

请比对【原始内容】和【润色后内容】：
- 如果只修改了文案表达，代码等技术内容完全一致（或只有注释、格式等非功能性变化），就回答"通过"。
- 如果发现任何技术内容被修改（如代码逻辑、变量名、参数、配置项等），就回答"不通过"，并指出具体是哪处内容被意外修改了。

【原始内容】:
{original_notebook_content}

【润色后内容】:
{draft_content}

prompt = """请完成以下课程审核任务：
1. 验证所有Python代码能正确运行
2. 检查技术概念的准确性（特别是pandas和numpy的用法说明）
3. 评估难度曲线是否适合编程新手
4. 按照我们的风格指南调整用词（避免"很简单""超级"等口语）
请给出完整的审核报告。"""
response = agent.run(prompt + course_content)

"""
模式一：流水线（Pipeline） - 场景：课程快速检查流程
目标：用户提交一篇课程初稿，系统按顺序完成
1) 提取课程中的代码 -> 2) 验证代码可执行性 -> 3) 生成代码检查报告
"""
import asyncio
from agentscope.message import Msg
from agentscope.pipeline import sequential_pipeline
from chatbot.agent import create_agent, disable_console_output

async def run_pipeline() -> None:
    # 节点A：代码提取 Agent
    # multi_agent=True 是 AgentScope 中的一个配置，用于确保 Agent 之间的通信格式兼容，
    # 是构建工作流和多智能体系统的推荐设置。
    code_extractor = create_agent(
        name="代码提取器",
        sys_prompt=(
            "你是代码提取专家。请从用户提供的课程文本中，精确地提取出所有 Python 代码块。"
            "只输出代码，不要有任何其他解释。"
        ),
        model_name="qwen-flash",
        multi_agent=True,
    )

    # 节点B：代码验证 Agent (可调用外部工具)
    code_validator = create_agent(
        name="代码验证器",
        sys_prompt=(
            "你是代码执行与验证专家。你将接收到代码文本。请使用代码解释器来执行它。"
            "报告代码是否能成功运行，如果不能，请指出错误。"
        ),
        # 此处可以配置 code_validator 使用你在第二节学过的代码解释器工具
        model_name="qwen-plus",
        multi_agent=True,
    )

    # 节点C：报告生成 Agent
    report_generator = create_agent(
        name="报告生成器",
        sys_prompt=(
            "你是审阅报告撰写助理。根据上一步的代码验证结果，为课程设计师生成一份简洁明了的检查报告。"
        ),
        model_name="qwen-max",
        multi_agent=True,
    )

    agents = [code_extractor, code_validator, report_generator]
    disable_console_output(agents)

    course_draft = (
        "这是我们的新课程。第一部分是`print('Hello, World!')`。"
        "第二部分是一个有问题的代码`x = 1 / 0`。"
    )
    result = await sequential_pipeline(
        agents=agents,
        msg=Msg("user", course_draft, "user"),
    )

    print("=" * 50)
    print("流水线输出：")
    print(result.content)
    print("=" * 50)

async def main() -> None:
    await run_pipeline()

await main()

"""
模式二：分支选择（Branching） - 场景：课程审阅任务分发
路由 Agent 读取审阅请求，选择以下分支之一：
1) code_check: 仅快速检查代码
2) style_guide: 按风格指南润色语言
3) full_review: 进行全面的多维度评审
"""
import asyncio
from typing import Literal
from pydantic import BaseModel, Field
from agentscope.message import Msg
from chatbot.agent import create_agent, disable_console_output

class RouteChoice(BaseModel):
    choice: Literal["code_check", "style_guide", "full_review", None] = Field(
        description="根据用户意图选择分支：code_check/style_guide/full_review/None"
    )
    extra: str | None = Field(default=None, description="对任务的简要说明")

async def branch_code_check(user_msg: Msg) -> Msg:
    agent = create_agent(
        name="代码快检专家",
        sys_prompt="你是代码快检专家。根据用户需求，快速验证课程中的代码片段是否能运行。",
        model_name="qwen-plus",
        multi_agent=True,
    )
    disable_console_output([agent])
    return await agent(user_msg)

async def branch_style_guide(user_msg: Msg) -> Msg:
    agent = create_agent(
        name="语言润色专家",
        sys_prompt="你是语言润色专家。请根据公司风格指南，改写和润色用户提供的课程文本。",
        model_name="qwen-max",
        multi_agent=True,
    )
    disable_console_output([agent])
    return await agent(user_msg)

async def branch_full_review(user_msg: Msg) -> Msg:
    agent = create_agent(
        name="首席评审",
        sys_prompt="你是首席评审。告知用户，你将启动一个包含代码、事实和教学法在内的全面评审流程。",
        model_name="qwen-flash", # 使用轻量模型模拟启动流程的告知动作
        multi_agent=True,
    )
    disable_console_output([agent])
    return await agent(user_msg)


async def run_branching() -> None:
    router = create_agent(
        name="审阅任务分发员",
        sys_prompt=(
            "你是课程审阅任务的分发员，根据用户输入选择分支：\n"
            "- 如果只是想检查代码，输出 code_check\n"
            "- 如果是想润色文笔，输出 style_guide\n"
            "- 如果是需要完整、全面的评审，输出 full_review\n"
            "仅通过结构化输出来表达你的选择，不要正文回答。"
        ),
        model_name="qwen-plus",
        multi_agent=False,
    )

    user_text = "这篇课程写的差不多了，帮我全面检查一下，特别是代码和难度。"

    res = await router(
        Msg("user", user_text, "user"),
        structured_model=RouteChoice,
    )
    choice = res.metadata.get("choice")

    if choice == "code_check":
        out = await branch_code_check(Msg("user", user_text, "user"))
    elif choice == "style_guide":
        out = await branch_style_guide(Msg("user", user_text, "user"))
    elif choice == "full_review":
        out = await branch_full_review(Msg("user", user_text, "user"))
    else:
        # 默认走全面评审，保证示例可运行
        out = await branch_full_review(Msg("user", user_text, "user"))

    print("=" * 50)
    print(f"分支选择：{choice}")
    print(out.content)
    print("=" * 50)

async def main() -> None:
    await run_branching()

await main()

"""
模式三：并行执行（Parallel Execution） - 场景：课程完整审阅
将独立的审阅子任务并行执行：代码检查、事实核对、教学法评估、语言风格检查。
示例使用 fanout_pipeline 并行收集报告，再由合成 Agent 汇总。
"""
import asyncio
from typing import List
from agentscope.message import Msg
from agentscope.pipeline import fanout_pipeline
from chatbot.agent import create_agent, disable_console_output

async def run_parallel() -> None:
    # 四个独立子任务的“专家”Agent
    code_checker = create_agent(
        name="代码检查员",
        sys_prompt="验证课程中的代码是否正确无误，并给出修复建议。",
        model_name="qwen-plus",
        multi_agent=True,
    )
    fact_checker = create_agent(
        name="事实核查员",
        sys_prompt="核对课程中的技术概念、函数解释是否准确，引用是否规范。",
        model_name="qwen-plus",
        multi_agent=True,
    )
    pedagogy_evaluator = create_agent(
        name="教学法评估师",
        sys_prompt="评估课程的难度曲线、案例趣味性和练习有效性。",
        model_name="qwen-flash",
        multi_agent=True,
    )
    style_editor = create_agent(
        name="风格编辑",
        sys_prompt="根据公司风格指南，检查并报告语言风格、术语一致性问题。",
        model_name="qwen-flash",
        multi_agent=True,
    )

    experts = [code_checker, fact_checker, pedagogy_evaluator, style_editor]
    disable_console_output(experts)

    course_content = "这是我们新开发的 Python 数据分析入门课..."
    msgs = await fanout_pipeline(
        agents=experts,
        msg=Msg("user", course_content, "user"),
        enable_gather=True,
    )

    # 汇总 Agent
    summarizer = create_agent(
        name="总编辑",
        sys_prompt="将来自多位专家的审阅意见汇总成一份结构清晰、条理分明的总审阅报告。",
        model_name="qwen-max",
        multi_agent=True,
    )
    disable_console_output([summarizer])

    merged_text: List[str] = [m.content for m in msgs]
    prompt = "\n\n".join(merged_text)
    summary = await summarizer(Msg("user", prompt, "user"))

    print("=" * 50)
    print("并行执行输出：")
    print(summary.content)
    print("=" * 50)

async def main() -> None:
    await run_parallel()

await main()

"""
模式四：混合专家（Mixture-of-Agents, MoA） - 场景：课程核心卖点提炼
使用三个不同的大语言模型并行处理同一任务，通过聚合模型融合它们的输出，
利用模型间的协作性(Collaborativeness)产生更高质量的结果。
"""
import asyncio
from agentscope.message import Msg
from agentscope.pipeline import fanout_pipeline
from chatbot.agent import create_agent, disable_console_output

async def run_moa() -> None:
    # 使用三个不同的模型作为提议者(Proposer)
    # 每个模型有其独特的优势，但处理相同的任务
    proposer1 = create_agent(
        name="Qwen3-Max",
        sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
        model_name="qwen3-max",
        multi_agent=True,
    )
    proposer2 = create_agent(
        name="DeepSeek-V3.2",
        sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
        model_name="deepseek-v3.2-exp",
        multi_agent=True,
    )
    proposer3 = create_agent(
        name="Kimi-K2",
        sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
        model_name="kimi-k2-thinking",
        multi_agent=True,
    )

    proposers = [proposer1, proposer2, proposer3]
    disable_console_output(proposers)

    task = (
        "这是一门新的'面向Web开发者的AI大模型应用'课程，请为其提炼核心卖点和宣传文案。"
    )
    msgs = await fanout_pipeline(
        agents=proposers,
        msg=Msg("user", task, "user"),
        enable_gather=True,
    )

    # 聚合器(Aggregator)接收所有模型的输出，综合产生最佳结果
    aggregator = create_agent(
        name="聚合器",
        sys_prompt=(
            "你的任务是综合多个大语言模型对同一问题的回答。"
            "这些回答来自不同的模型，各有优劣。请批判性地评估这些回答，"
            "识别其中的优点和不足，然后融合这些信息，生成一个高质量、准确、全面的最终回答。"
            "确保你的回答结构清晰、逻辑连贯，并达到最高的准确性和可靠性标准。"
        ),
        model_name="qwen3-max",
        multi_agent=True,
    )
    disable_console_output([aggregator])

    # 将所有提议者的输出合并，传递给聚合器
    merged = "\n\n".join([
        f"模型 {i+1} 的回答：\n{m.content}"
        for i, m in enumerate(msgs)
    ])
    final = await aggregator(Msg("user", merged, "user"))

    print("=" * 50)
    print("MoA 聚合输出：")
    print(final.content)
    print("=" * 50)

async def main() -> None:
    await run_moa()

await main()

async def run_multi_layer_moa() -> None:
    """
    多层 MoA 示例：3层架构提炼课程营销方案
    Layer 1: 3个不同模型作为提议者并行生成方案
    Layer 2: 3个不同模型对第一层的聚合输出进行优化和改进
    Layer 3: 最终聚合器综合所有信息，输出最佳方案
    """
    # 通用的聚合提示词
    aggregate_prompt = (
        "你的任务是综合多个大语言模型对同一问题的回答。"
        "请批判性地评估这些回答，识别其中的优点和不足，"
        "然后融合这些信息，生成一个高质量、准确、全面的最终回答。"
    )

    # Layer 1: 初始提议者层(3个不同模型)
    layer1_proposers = [
        create_agent(
            name="Proposer-L1-1",
            sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
            model_name="qwen3-max",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L1-2",
            sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
            model_name="deepseek-v3.2-exp",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L1-3",
            sys_prompt="你是一个专业的课程分析师，请为给定的课程提炼核心卖点和宣传文案。",
            model_name="kimi-k2-thinking",
            multi_agent=True
        ),
    ]

    task = "这是一门新的'面向Web开发者的AI大模型应用'课程，请为其提炼核心卖点和宣传文案。"
    layer1_outputs = await fanout_pipeline(
        agents=layer1_proposers,
        msg=Msg("user", task, "user"),
        enable_gather=True,
    )

    # Layer 1 聚合器
    layer1_aggregator = create_agent(
        name="Aggregator-L1",
        sys_prompt=aggregate_prompt,
        model_name="qwen3-max",
        multi_agent=True,
    )
    layer1_merged = "\n\n".join([f"模型 {i+1}:\n{m.content}" for i, m in enumerate(layer1_outputs)])
    layer1_result = await layer1_aggregator(Msg("user", layer1_merged, "user"))

    # Layer 2: 第二轮提议者层(3个不同模型，基于第一层的聚合结果进行优化)
    layer2_proposers = [
        create_agent(
            name="Proposer-L2-1",
            sys_prompt="你是一个专业的课程分析师。请审视给定的营销方案，并提出改进建议或优化版本。",
            model_name="qwen3-max",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L2-2",
            sys_prompt="你是一个专业的课程分析师。请审视给定的营销方案，并提出改进建议或优化版本。",
            model_name="deepseek-v3.2-exp",
            multi_agent=True
        ),
        create_agent(
            name="Proposer-L2-3",
            sys_prompt="你是一个专业的课程分析师。请审视给定的营销方案，并提出改进建议或优化版本。",
            model_name="kimi-k2-thinking",
            multi_agent=True
        ),
    ]

    layer2_prompt = f"以下是第一轮分析生成的营销方案：\n\n{layer1_result.content}\n\n请在此基础上提出改进建议或优化版本。"
    layer2_outputs = await fanout_pipeline(
        agents=layer2_proposers,
        msg=Msg("user", layer2_prompt, "user"),
        enable_gather=True,
    )

    # Layer 3: 最终聚合层
    final_aggregator = create_agent(
        name="Final-Aggregator",
        sys_prompt=aggregate_prompt,
        model_name="qwen3-max",
        multi_agent=True,
    )

    layer2_merged = "\n\n".join([f"模型 {i+1}:\n{m.content}" for i, m in enumerate(layer2_outputs)])
    final_output = await final_aggregator(Msg("user", layer2_merged, "user"))

    print("多层 MoA 最终输出：")
    print(final_output.content)

await run_multi_layer_moa()

"""
模式五：人机协作（HITL） - 场景：课程疑难点审核（Human as a Tool）
AI 先对课程中的一个潜在疑难点给出修改建议；随后由具备“人类咨询”工具的改写 Agent 自主决定何时调用该工具向人类请示，最终完成修改。
"""
import asyncio
import os
from agentscope.agent import ReActAgent, UserAgent
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.tool import Toolkit, ToolResponse
from chatbot.agent import create_agent, disable_console_output


# 将人类介入封装为一个工具：ask_human_decision
async def ask_human_decision(question: str) -> ToolResponse:
    """向人类专家征求决策或意见。

    Args:
        question (str): 想要请人类确认或补充的具体问题。
    """
    human_expert = UserAgent(name="教学专家")
    reply = await human_expert(
        Msg(
            "assistant",
            question,
            "assistant",
        )
    )
    return ToolResponse(
        content=[
            TextBlock(type="text", text=reply.get_text_content()),
        ]
    )


async def run_hitl() -> None:
    # AI：给出修改建议
    suggester = create_agent(
        name="疑难点分析师",
        sys_prompt=(
            "你是一名资深教学设计师。请找出课程中对初学者可能最难理解的一个概念，"
            "并提供一个更通俗易懂的解释作为修改建议。"
        ),
        model_name="qwen-plus",
        multi_agent=True,
    )
    disable_console_output([suggester])

    course_content = "在Python中，装饰器本质上是一个接收函数作为参数并返回一个新函数的函数..."
    suggestion = await suggester(Msg("user", course_content, "user"))

    print("AI 建议如下：\n")
    print(suggestion.content)

    # 将“人类介入”作为工具交给改写 Agent，自主决定是否调用
    toolkit = Toolkit()
    toolkit.register_tool_function(ask_human_decision)

    rewriter = ReActAgent(
        name="内容改写器",
        sys_prompt=(
            "你是课程内容改写器。基于提供的 AI 建议完成最终修改。\n"
            "- 若你有把握，请直接完成修改并给出确认信息；\n"
            "- 若存在不确定、歧义或高风险，请调用工具 ask_human_decision 先向人类专家请示，"
            "再据此完成修改；\n"
            "- 在最终结果中简要说明是否咨询了人类及原因。"
        ),
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY", "your-api-key"),
            stream=False,
        ),
        formatter=DashScopeMultiAgentFormatter(),
        toolkit=toolkit,
    )
    disable_console_output([rewriter])

    # 将课程内容与 AI 建议一并提供给改写 Agent
    task = (
        "下面是课程摘录与 AI 的修改建议。根据系统提示完成最终修改：\n\n"
        f"[课程内容]\n{course_content}\n\n"
        f"[AI 建议]\n{suggestion.get_text_content()}\n"
    )
    final_action = await rewriter(Msg("user", task, "user"))

    print("=" * 50)
    print("HITL 最终输出：")
    print(final_action.content)
    print("=" * 50)


async def main() -> None:
    await run_hitl()

await main()

"""AgentScope - Agent自主规划与执行示例（简化版）"""
import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.message import Msg, TextBlock
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit, ToolResponse
from agentscope.plan import PlanNotebook


# 模拟业务工具
async def analyze_competitor_course(url: str) -> ToolResponse:
    """分析竞品课程页面的大纲"""
    # 模拟因网站改版导致解析失败
    return ToolResponse(content=[
        TextBlock(type="text", text=f"❌ 错误：因 {url} 网站布局更新，无法解析课程大纲。")
    ])

async def search_industry_demand(topic: str) -> ToolResponse:
    """查询行业的技能需求"""
    return ToolResponse(content=[
        TextBlock(type="text", text=f"✅ 报告：关于“{topic}”的行业需求分析已完成。")
    ])

async def google_search(query: str) -> ToolResponse:
    """谷歌网页搜索"""
    if "syllabus" in query:
        return ToolResponse(content=[
            TextBlock(type="text", text="搜索结果：找到了'Python入门课程'的大纲PDF，地址 a.com/syllabus.pdf")
        ])
    return ToolResponse(content=[TextBlock(type="text", text="未找到相关信息")])

async def extract_text_from_pdf(url: str) -> ToolResponse:
    """从PDF链接中提取文本"""
    return ToolResponse(content=[
        TextBlock(type="text", text=f"✅ 已从 {url} 提取大纲文本：1. 变量与数据类型... 2. ...")
    ])


# 用于监控计划变化的钩子函数
plan_snapshots = []

def capture_plan_snapshot(notebook, plan):
    """捕获计划快照"""
    if plan:
        plan_snapshots.append({
            "name": plan.name,
            "description": plan.description,
            "state": plan.state,
            "subtasks": [
                {
                    "name": st.name,
                    "state": st.state,
                    "outcome": st.outcome
                }
                for st in plan.subtasks
            ]
        })


async def main():

    print("=" * 60)
    print("🤖 Agent自主规划演示")
    print("=" * 60)

    # 创建PlanNotebook并注册钩子
    plan_notebook = PlanNotebook()
    plan_notebook.register_plan_change_hook("capture", capture_plan_snapshot)

    # 创建工具箱
    toolkit = Toolkit()
    toolkit.register_tool_function(analyze_competitor_course)
    toolkit.register_tool_function(search_industry_demand)
    toolkit.register_tool_function(google_search)
    toolkit.register_tool_function(extract_text_from_pdf)

    # 创建Agent
    agent = ReActAgent(
        name="CourseResearcherAgent",
        sys_prompt=(
            "你是课程调研助手。遇到复杂任务时：\n"
            "1. 用create_plan创建计划\n"
            "2. 逐步执行，用finish_subtask标记完成\n"
            "3. 遇到问题灵活调整，例如使用google_search寻找替代方案\n"
            "4. 完成后用finish_plan结束"
        ),
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
        plan_notebook=plan_notebook,
    )

    # 用户请求
    print("\n💬 用户: 请帮我完成一门新的 Python 入门课程的前期调研。\n")
    print("-" * 60)

    await agent(Msg("user", "请帮我完成一门新的 Python 入门课程的前期调研，竞品是 some-site.com 的课程。", "user"))

    # 显示结果（从快照中获取最后的完整计划）
    print("\n" + "=" * 60)
    print("📊 执行结果")
    print("=" * 60)

    if plan_snapshots:
        final_plan = plan_snapshots[-1]
        finished = sum(1 for st in final_plan["subtasks"] if st["state"] == "finished")

        print(f"\n✅ 计划: {final_plan['name']}")
        print(f"📊 进度: {finished}/{len(final_plan['subtasks'])}")
        print(f"🎯 状态: {final_plan['state']}\n")

        print("子任务详情:")
        for i, subtask in enumerate(final_plan["subtasks"], 1):
            icon = "✅" if subtask["state"] == "finished" else "⏳"
            print(f"  {icon} {i}. {subtask['name']}")


await main()

{
  "plan": [
    {
      "step": 1,
      "thought": "我首先需要分析行业需求，这是课程定位的关键。",
      "tool_name": "search_industry_demand",
      "tool_params": {"topic": "Python 基础"}
    },
    {
      "step": 2,
      "thought": "接下来，我尝试分析竞品课程的大纲。",
      "tool_name": "analyze_competitor_course",
      "tool_params": {"url": "some-site.com/python-course"}
    }
  ]
}

# Plan generated by LLM
def execute_research_plan():
    # Step 1: Analyze industry demand
    demand_result = search_industry_demand(topic="Python 基础")
    print(demand_result)

    # Step 2: Analyze competitor course
    competitor_result = analyze_competitor_course(url="some-site.com/python-course")

    # Step 3: Handle analysis failure
    if not competitor_result.success and "无法解析" in competitor_result.message:
        print("竞品分析工具失效，正在寻找备选方案...")
        search_results = google_search(query="some-site.com python course syllabus")
        # Assume search_results gives a PDF link
        pdf_url = extract_pdf_link(search_results) 
        if pdf_url:
            syllabus_text = extract_text_from_pdf(url=pdf_url)
            print(syllabus_text)
    else:
        print(competitor_result)

execute_research_plan()

"""AgentScope Agent自主创建工具 - 精简版"""
import asyncio
import os
import sys
from io import StringIO

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

# 全局工具箱
toolkit = None


async def code_exec(code: str) -> ToolResponse:
    """代码解释器 - 用于创建和注册新工具"""
    global toolkit

    namespace = {
        'ToolResponse': ToolResponse,
        'TextBlock': TextBlock,
        'asyncio': asyncio,
        'agent_toolkit': toolkit,
        'math': __import__('math'),
    }

    stdout, sys.stdout = sys.stdout, StringIO()

    try:
        exec(code, namespace)
        output = sys.stdout.getvalue()
        sys.stdout = stdout
        return ToolResponse(content=[TextBlock(
            type="text", 
            text=output or "✅ 执行成功"
        )])
    except Exception as e:
        sys.stdout = stdout
        return ToolResponse(content=[TextBlock(
            type="text",
            text=f"❌ 错误: {e}"
        )])


async def add(a: float, b: float) -> ToolResponse:
    """加法工具"""
    return ToolResponse(content=[TextBlock(
        type="text", 
        text=f"{a} + {b} = {a + b}"
    )])


async def main():
    if "DASHSCOPE_API_KEY" not in os.environ:
        print("❌ 请设置 DASHSCOPE_API_KEY")
        return

    global toolkit
    toolkit = Toolkit()
    toolkit.register_tool_function(add)
    toolkit.register_tool_function(code_exec)

    agent = ReActAgent(
        name="ToolMaker",
        sys_prompt=(
            "你可以通过 code_exec 创建新工具。\n"
            "模板:\n"
            "async def tool_name(param: type) -> ToolResponse:\n"
            "    '''描述'''\n"
            "    result = ...\n"
            "    return ToolResponse(content=[TextBlock(type='text', text=f'{result}')])\n"
            "agent_toolkit.register_tool_function(tool_name)\n"
            "print('✅ 已注册 tool_name')"
        ),
        model=DashScopeChatModel(
            model_name="qwen-plus",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
        ),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
        memory=InMemoryMemory(),
    )

    print("=" * 60)
    print("🚀 Agent 自主创建工具演示")
    print("=" * 60)

    # 使用现有工具
    print("\n▶️ 场景1: 使用现有工具")
    await agent(Msg("user", "计算 30 + 45", "user"))

    # 创建新工具
    print("\n▶️ 场景2: 创建阶乘工具")
    await agent(Msg("user", "创建 factorial 工具计算阶乘", "user"))

    # 使用新工具
    print("\n▶️ 场景3: 使用新工具")
    await agent(Msg("user", "用 factorial 计算 5 的阶乘", "user"))

    # 显示工具箱
    print("\n📦 最终工具箱:")
    for i, s in enumerate(toolkit.get_json_schemas(), 1):
        print(f"{i}. {s['function']['name']}")


if __name__ == "__main__":
    await main()

import os
from typing import Any

from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.tool import ToolResponse, Toolkit

# ---- 1. 定义统一的专家 Agent 角色和提示 -
---

DESIGNER_LI_PROMPT = """
你是李老师，一位经验丰富的教学设计师。你的任务是为“Pandas 数据分析入门”课程设计出清晰、有逻辑的教学大纲。专注于：1. 定义每个模块清晰的学习目标。2. 确保知识点由浅入深，循序渐进。3. 提出互动性的练习和项目来巩固学习效果。
"""

SCIENTIST_WANG_PROMPT = """
你是王工，一位资深数据科学家，也是 Pandas 的实战专家。你的任务是为课程提供准确、实用的技术内容。专注于：1. 提供最核心、最常用的 Pandas 知识点。2. 设计源于真实工作场景的案例和数据集。3. 编写简洁、规范、易于理解的代码示例。
"""

WRITER_ZHANG_PROMPT = """
你是小张，一位充满创意的课程内容编写者。你的任务是把技术内容讲得通俗易懂、但不失严谨性、用词冷静克制的课程文稿。专注于：1. 用通俗易懂的语言和比喻来解释复杂概念。2. 设计真实性高的案例场景和模块标题。3. 确保课程的整体基调是鼓励性和启发性的。
"""

LEADER_PROMPT = """
你是一个课程项目主管，负责协调团队完成“Pandas入门课程”的初稿开发。
你有三名团队成员可以作为工具调用，他们每个人的工作都依赖于前一个人的输出。

你的工作流程必须严格遵循以下顺序：

1. **首先，调用 invoke_designer_li**，让他为课程创建一个初步的大纲和学习目标。
2. **其次，调用 invoke_scientist_wang**。将李老师生成的大纲作为 `context` 参数传递给他，要求他根据这个大纲填充技术要点和代码示例。
3. **接着，调用 invoke_writer_zhang**。将李老师和王工的全部产出合并后作为 `context` 参数传递给她，要求她在此基础上撰写完整的、对学习者友好的课程文稿。
4. **最后**，在收到所有专家的最终结果后，将它们整合成一份格式统一、内容完整的最终课程文档，然后作为你的最终回复。
   """

# ---- 2. 统一模型和 Agent 配置 ----

def get_model_instance() -> DashScopeChatModel:
    """获取一个统一配置的模型实例。"""
    return DashScopeChatModel(
        model_name="qwen-plus",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
    )

def create_member_agent(name: str, sys_prompt: str) -> ReActAgent:
    """根据给定的名称和系统提示创建一个团队成员 Agent。"""
    return ReActAgent(
        name=name,
        sys_prompt=sys_prompt,
        model=get_model_instance(),
        formatter=DashScopeMultiAgentFormatter(),
    )

# ---- 3. 定义作为工具的“团队成员” Agent (Handoffs 模式) ----

async def invoke_designer_li(task_description: str, context: str = "") -> ToolResponse:
    """
    当需要设计课程大纲、学习目标或教学活动时，调用教学设计师李老师。

    Args:
        task_description (str): 清晰地描述你需要李老师完成的设计任务。
        context (str): 可选。传递相关的背景信息或先前的工作成果。
    """
    print("\n--- 任务分派：正在调用教学设计师李老师 ---")
    agent = create_member_agent("DesignerLi", DESIGNER_LI_PROMPT)
    
    content_for_agent = task_description
    if context:
        content_for_agent = f"背景信息：\n{context}\n\n你的任务：{task_description}"
    
    result_msg = await agent(Msg(name="user", role="user", content=content_for_agent))
    return ToolResponse(content=result_msg.get_text_content())

async def invoke_scientist_wang(task_description: str, context: str = "") -> ToolResponse:
    """
    当需要提供专业技术知识、代码示例或真实案例时，调用数据科学家王工。

    Args:
        task_description (str): 清晰地描述你需要王工完成的技术任务。
        context (str): 可选。传递课程大纲等先前的工作成果，以便他在此基础上工作。
    """
    print("\n--- 任务分派：正在调用数据科学家王工 ---")
    agent = create_member_agent("ScientistWang", SCIENTIST_WANG_PROMPT)
    
    content_for_agent = task_description
    if context:
        content_for_agent = f"请基于以下课程大纲和背景信息来完成你的任务：\n{context}\n\n你的具体任务是：{task_description}"
    
    result_msg = await agent(Msg(name="user", role="user", content=content_for_agent))
    return ToolResponse(content=result_msg.get_text_content())

async def invoke_writer_zhang(task_description: str, context: str = "") -> ToolResponse:
    """
    当需要将技术内容转化为易于理解的文稿时，调用内容编写者小张。

    Args:
        task_description (str): 清晰地描述你需要小张完成的写作任务。
        context (str): 可选。传递大纲和技术要点等先前的工作成果，作为写作基础。
    """
    print("\n--- 任务分派：正在调用内容编写者小张 ---")
    agent = create_member_agent("WriterZhang", WRITER_ZHANG_PROMPT)
    
    content_for_agent = task_description
    if context:
        content_for_agent = f"请基于以下课程的草稿（包含大纲和技术点）来完成你的写作任务：\n{context}\n\n你的具体任务是：{task_description}"
    
    result_msg = await agent(Msg(name="user", role="user", content=content_for_agent))
    return ToolResponse(content=result_msg.get_text_content())

# ---- 4. 组织“分层规划”工作流 ----

async def main() -> None:
    """主执行函数，负责编排整个工作流。"""

    # 4.1 创建主管的工具包并注册团队成员
    leader_toolkit = Toolkit()
    leader_toolkit.register_tool_function(invoke_designer_li)
    leader_toolkit.register_tool_function(invoke_scientist_wang)
    leader_toolkit.register_tool_function(invoke_writer_zhang)
    
    # 4.2 创建主管 Agent
    leader_agent = ReActAgent(
        name="ProjectLeader",
        sys_prompt=LEADER_PROMPT,
        model=get_model_instance(),
        toolkit=leader_toolkit,
        formatter=DashScopeMultiAgentFormatter(),
    )
    
    # 4.3 定义顶层任务
    top_level_task = (
        "请为初学者创建一节关于Pandas 数据分析的简短课程初稿。"
    )
    
    print(f"项目主管收到的顶层任务：\n{top_level_task}\n" + "="*50)
    
    # 4.4 将任务交给主管 Agent 执行
    final_response_msg = await leader_agent(Msg(name="user", role="user", content=top_level_task))
    
    # 4.5 展示最终成果
    print("\n" + "="*50)
    print("  项目主管最终的汇总报告：")
    print("="*50 + "\n")
    print(final_response_msg.get_text_content())

# ---- 5. 运行主程序 ----

await main()

import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.pipeline import MsgHub
from textwrap import dedent

# ---- 1. 定义统一的专家 Agent 角色和提示 ----

DESIGNER_LI_PROMPT = """
你是李老师，一位经验丰富的教学设计师。你的任务是为“Pandas 数据分析入门”课程设计出清晰、有逻辑的教学大纲。在讨论中，你专注于：1. 定义每个模块清晰的学习目标。2. 确保知识点由浅入深，循序渐进。3. 提出互动性的练习和项目来巩固学习效果。
"""

SCIENTIST_WANG_PROMPT = """
你是王工，一位资深数据科学家，也是 Pandas 的实战专家。你的任务是为课程提供准确、实用的技术内容。在讨论中，你专注于：1. 提供最核心、最常用的 Pandas 知识点。2. 设计源于真实工作场景的案例和数据集。3. 编写简洁、规范、易于理解的代码示例。
"""

WRITER_ZHANG_PROMPT = """
你是小张，一位充满创意的课程内容编写者。你的任务是把技术内容讲得通俗易懂、但不失严谨性、用词冷静克制的课程文稿。在讨论中，你专注于：1. 用通俗易懂的语言和比喻来解释复杂概念。2. 设计真实性高的案例场景和模块标题。3. 确保课程的整体基调是鼓励性和启发性的。
"""

# ---- 2. 创建专家 Agent 的辅助函数 ----
def create_expert_agent(name: str, sys_prompt: str) -> ReActAgent:
    """根据给定的名称和系统提示创建一个专家 Agent。"""
    return ReActAgent(
        name=name,
        sys_prompt=sys_prompt,
        model=DashScopeChatModel(
            model_name="qwen-plus",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
        ),
        formatter=DashScopeMultiAgentFormatter(),
    )

# ---- 3. 组织协作流程的主函数 ----
async def main() -> None:
    """运行 Pandas 课程开发的共创协作模式，并生成最终成果。"""
    print("=== 开始课程开发会议：构思 'Pandas 入门' 课程大纲和案例 ===")

    # 创建课程开发团队
    designer_li = create_expert_agent("李老师 (教学设计师)", DESIGNER_LI_PROMPT)
    scientist_wang = create_expert_agent("王工 (数据科学家)", SCIENTIST_WANG_PROMPT)
    writer_zhang = create_expert_agent("小张 (内容编写者)", WRITER_ZHANG_PROMPT)

    # 定义会议开场白
    announcement = Msg(
        "system",
        (
            "团队好，我们今天的目标是共同协作，为“Pandas 数据分析入门”课程制定一个完整的、吸引人的**课程大纲和核心案例**。"
            "请大家集思广益，从教学设计师李老师开始，提出你的第一轮建议。"
        ),
        "system",
    )

    # 启动多轮讨论
    async with MsgHub(
        participants=[designer_li, scientist_wang, writer_zhang],
        announcement=announcement,
    ) as hub:
        for i in range(2):
            print(f"\n--- 第 {i + 1} 轮协作 ---")
            # 按照发言顺序依次调用
            await designer_li()
            await scientist_wang()
            await writer_zhang()

    print("\n=== 会议结束 ===")

    # ==================== 汇总阶段 ====================
    print("\n=== 开始生成最终团队成果（课程大纲初稿） ===")

    # 4.1 定义一个“会议秘书” Agent 来整理会议纪要
    secretary_prompt = dedent("""
        你是一位专业的会议秘书，非常擅长整理会议纪要。
        你的任务是阅读下面的团队讨论记录，然后根据讨论内容，以清晰的 Markdown 格式，
        生成“Pandas 入门课程”的**课程大纲初稿**。

        大纲应包含以下部分：
        - **模块标题**：一个吸引人的标题。
        - **学习目标**：清晰列出学生学完本模块后能做什么。
        - **核心概念**：涵盖的关键技术点。
        - **核心案例**：贯穿本模块的实践案例和数据集。
        - **代码示例**：需要包含的关键代码演示。
        - **课后练习**：一个具体的动手练习任务。
    """)
    secretary_agent = create_expert_agent("会议秘书", secretary_prompt)

    # 4.2 准备完整的讨论记录
    full_transcript_msgs = await designer_li.memory.get_memory()

    transcript_text = "以下是团队的讨论记录：\n\n"
    for msg in full_transcript_msgs:
        if msg.role != "system":
            transcript_text += f"[{msg.name}]: {msg.content}\n"

    # 4.3 指派汇总任务
    final_task_prompt = dedent(
        f"{transcript_text}\n"
        "请根据以上讨论记录，整理出课程大纲初稿。"
    )

    # 调用秘书 Agent 来完成任务
    final_output_msg = await secretary_agent(Msg("user", final_task_prompt, "user"))

    # 4.4 展示最终成果
    print("\n" + "="*25)
    print("  最终团队成果：课程大纲初稿")
    print("="*25 + "\n")
    print(final_output_msg.content)

await main()

# 创建一个课程编写 Agent
writing_agent = create_agent(
    name="Writer",
    sys_prompt="你是一个课程内容编写员。你的任务是编写一篇 Pandas 数据分析课程。"
)

async def run_stateless_test():
    # 第一次对话：设定教学风格
    msg1 = Msg("user", "我们的教学风格要严谨克制，请记住这一点。", "user")
    print(f"[{msg1.name}]: {msg1.content}")

    # Agent 会将这次对话存入它的 InMemoryMemory
    reply1 = await writing_agent(msg1)
    print(f"[{reply1.name}]: {reply1.content}")

    print("\n" + "="*20 + "\n")

    # 第二次对话：基于之前的设定提出新要求
    # 在调用时，writing_agent 会自动将 InMemoryMemory 中的历史记录和新消息一起发给模型
    msg2 = Msg("user", "很好，现在请把第二部分写得更专业一些。", "user")
    print(f"[{msg2.name}]: {msg2.content}")

    reply2 = await writing_agent(msg2)
    print(f"[{reply2.name}]: {reply2.content}")

    print("\n" + "="*20 + "\n")
    print("Agent 的短期记忆内容：")
    # 打印 Agent 的记忆，可以看到包含了全部两轮对话
    for m in await writing_agent.memory.get_memory():
        print(f"- [{m.role}] {m.name}: {m.content}")


await run_stateless_test()

# 这是一个伪代码示例，用于演示滚动摘要的核心逻辑
# 它不能直接运行，需要你继承 agentscope.memory.MemoryBase 并实现完整逻辑

class SummaryMemory: # (MemoryBase)
    def __init__(self, buffer_size=10, summary_ratio=0.5):
        self.history = []
        self.buffer_size = buffer_size
        self.summary_ratio = summary_ratio

    def add(self, message):
        self.history.append(message)
        self.try_summarize()

    def try_summarize(self):
        if len(self.history) > self.buffer_size:
            # 1. 确定要摘要的部分
            num_to_summarize = int(len(self.history) * self.summary_ratio)
            messages_to_summarize = self.history[:num_to_summarize]

            # 2. 调用大模型生成摘要 (伪代码)
            # summary_text = llm.call("请将以下对话总结为一段话：", messages_to_summarize)
            summary_text = "用户设定了教学风格为风趣幽默，并要求内容生动。"

            summary_message = Msg("system", f"【历史摘要】{summary_text}", "system")

            # 3. 用摘要替换原始对话
            self.history = [summary_message] + self.history[num_to_summarize:]
            print(f"--- 记忆已压缩，当前长度 {len(self.history)} ---")

    def get_memory(self):
        return self.history

# 使用示例
# summary_mem = SummaryMemory()
# summary_mem.add(Msg("user", "我们的教学风格要风趣幽默。", "user"))
# ... 经过多轮对话后 ...
# summary_mem.add(Msg("user", "再增加一个关于成本的案例。", "user")) # 此时可能会触发摘要

# 1. 初始化长期记忆模块
# 它需要一个语言模型（用于内部处理）和一个嵌入模型
from mem0.vector_stores.configs import VectorStoreConfig

# 为 Qdrant 本地向量库指定与 DashScope Embedding 一致的维度（2048）
vector_store = VectorStoreConfig(
    config={
        "on_disk": False,
        "embedding_model_dims": 2048,
    }
)

long_term_memory = Mem0LongTermMemory(
    agent_name="Writer",
    user_name="user",
    model=DashScopeChatModel(
        model_name="qwen-plus",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
        stream=False,
    ),
    embedding_model=DashScopeTextEmbedding(
        model_name="text-embedding-v4",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
        dimensions=2048
    ),
    vector_store_config=vector_store,
)

# 2. 创建一个装备了长期记忆的 Agent
ltm_agent_static = create_agent(
    name="LTM_Writer_Static",
    sys_prompt="你是一个拥有长期记忆的课程编写员。",
    long_term_memory=long_term_memory,
    long_term_memory_mode="static_control", # 关键参数：设置为静态控制模式
)


async def run_ltm_static_test():
    # 对话一：存入一个关键信息
    msg1 = Msg("user", "记住，我们正在写一篇 Pandas 数据分析课程，目前已经写完初稿。", "user")
    print(f"[{msg1.name}]: {msg1.content}")
    reply1 = await ltm_agent_static(msg1)
    print(f"[{reply1.name}]: {reply1.content}")
    # 在这一步之后，对话内容会被自动存入长期记忆

    print("\n" + "="*20 + " 模拟新的一次会话 " + "="*20 + "\n")

    # 清空 Agent 的短期记忆，模拟一次全新的会话
    await ltm_agent_static.memory.clear()
    print("Agent 的短期记忆已被清空。")

    # 对话二：提出一个相关问题
    # Agent 在回复前，会用问题“这门课适合谁？”去长期记忆中检索
    msg2 = Msg("user", "我们上次的工作进度到哪儿了？", "user")
    print(f"[{msg2.name}]: {msg2.content}")
    reply2 = await ltm_agent_static(msg2)
    print(f"[{reply2.name}]: {reply2.content}")
    print("\n注意: 即使短期记忆被清空，Agent 依然能回答正确，因为它从长期记忆中检索到了相关信息。")

await run_ltm_static_test()

# 复用之前创建的 long_term_memory 实例
# 重新创建一个 Agent，这次使用 agent_control 模式
from textwrap import dedent
ltm_agent_active = create_agent(
    name="LTM_Writer_Active",
    sys_prompt=dedent(
        "你是一个拥有主动记忆管理能力的课程编写员。\n"
        "你可以使用以下工具来管理你的长期记忆：\n"
        "- `record_to_memory(data: str)`: 将一段重要的信息记录到长期记忆中。\n"
        "- `retrieve_from_memory(query: str) -> str`: 根据查询从长期记忆中检索相关信息。\n"
        "在回答问题前，先思考是否需要检索记忆。在对话结束后，思考是否有关键信息需要记录。"
    ),
    long_term_memory=long_term_memory,
    long_term_memory_mode="agent_control", # 关键参数：设置为 Agent 控制模式
)


async def run_ltm_active_test():
    # 对话一：Agent 自主决定记录信息
    msg1 = Msg("user", "课程的写作风格必须非常严谨和学术化，这是一个核心要求。", "user")
    print(f"[{msg1.name}]: {msg1.content}")
    reply1 = await ltm_agent_active(msg1)
    # Agent 在这里的思考过程中，会判断“核心要求”是重要信息，并调用 record_to_memory 工具
    print(f"[{reply1.name}]: {reply1.content}")

    print("\n" + "="*20 + " 模拟一次新的会话 " + "="*20 + "\n")
    await ltm_agent_active.memory.clear()

    # 对话二：Agent 自主决定检索信息
    msg2 = Msg("user", "我忘了，我们课程的写作风格是什么来着？", "user")
    print(f"[{msg2.name}]: {msg2.content}")
    reply2 = await ltm_agent_active(msg2)
    # Agent 在这里的思考过程中，会先调用 retrieve_from_memory(query="写作风格")，再根据检索结果生成回答
    print(f"[{reply2.name}]: {reply2.content}")

await run_ltm_active_test()