[Datawheel]利用Zigent框架编写智能体-2

书接上回

1. Zigent实现教程编写智能体

本节将通过 Zigent 框架实现一个教程编写智能体，其主要功能是输入教程主题，然后自动生成完整的教程内容。
设计思路： 先通过 LLM 大模型生成教程的目录，再对目录按照二级标题进行分块，对于每块目录按照标题生成详细内容，最后再将标题和内容进行拼接。分块的设计解决了 LLM 大模型长文本的限制问题。
首先是大模型配置：

import os
from dotenv import load_dotenv
from zigent.llm.agent_llms import LLM

# 加载环境变量
load_dotenv()
# 从环境变量中读取api_key
api_key = os.getenv('ZISHU_API_KEY')
base_url = "http://43.200.7.56:8008/v1"
chat_model = "glm-4-flash"

llm = LLM(api_key=api_key, base_url=base_url, model_name=chat_model)

导入zigent依赖

from typing import List, Dict
from zigent.llm.agent_llms import LLM
from zigent.actions import BaseAction, ThinkAct, FinishAct
from zigent.agents import BaseAgent
from zigent.commons import TaskPackage, AgentAct
from zigent.actions.InnerActions import INNER_ACT_KEY
from datetime import datetime
import json

定义生成教程的目录 Action 类

定义 WriteDirectoryAction 类，继承自 BaseAction。该类的主要功能是生成一个教程的目录结构。具体来说，它通过调用大语言模型（LLM）来根据给定的主题和语言生成一个符合特定格式的目录。

class WriteDirectoryAction(BaseAction):
    """Generate tutorial directory structure action"""
    def __init__(self) -> None:
        action_name = "WriteDirectory"
        action_desc = "Generate tutorial directory structure"
        params_doc = {
            "topic": "(Type: string): The tutorial topic name",
            "language": "(Type: string): Output language (default: 'Chinese')"
        }
        super().__init__(action_name, action_desc, params_doc)
        
    def __call__(self, **kwargs):
        topic = kwargs.get("topic", "")
        language = kwargs.get("language", "Chinese")
        
        directory_prompt = f"""
        请为主题"{topic}"生成教程目录结构,要求:
        1. 输出语言必须是{language}
        2. 严格按照以下字典格式输出: {{"title": "xxx", "directory": [{{"章节1": ["小节1", "小节2"]}}, {{"章节2": ["小节3", "小节4"]}}]}}
        3. 目录层次要合理,包含主目录和子目录
        4. 每个目录标题要有实际意义
        5. 不要有多余的空格或换行
        """
        
        # 调用 LLM 生成目录
        directory_data = llm.run(directory_prompt)
        try:
            directory_data = json.loads(directory_data)
        except:
            directory_data = {"title": topic, "directory": []}
            
        return {
            "topic": topic,
            "language": language,
            "directory_data": directory_data
        }
  

定义生成教程内容的 Action 类
WriteContentAction 类用于生成教程内容。它的 call 方法接收标题、章节、语言和目录数据，并构建一个内容提示，最后调用 LLM 生成相应的内容。

class WriteContentAction(BaseAction):
    """Generate tutorial content action"""
    def __init__(self) -> None:
        action_name = "WriteContent"
        action_desc = "Generate detailed tutorial content based on directory structure"
        params_doc = {
            "title": "(Type: string): The section title",
            "chapter": "(Type: string): The chapter title",
            "directory_data": "(Type: dict): The complete directory structure", 
            "language": "(Type: string): Output language (default: 'Chinese')"
        }
        super().__init__(action_name, action_desc, params_doc)
        
    def __call__(self, **kwargs):
        title = kwargs.get("title", "")
        chapter = kwargs.get("chapter", "")
        language = kwargs.get("language", "Chinese")
        directory_data = kwargs.get("directory_data", {})
        
        content_prompt = f"""
        请为教程章节生成详细内容:
        教程标题: {directory_data.get('title', '')}
        章节: {chapter}
        小节: {title}
        
        要求:
        1. 内容要详细且准确
        2. 如果需要代码示例,请按标准规范提供
        3. 使用 Markdown 格式
        4. 输出语言必须是{language}
        5. 内容长度适中,通常在500-1000字之间
        """
        
        # 调用 LLM 生成内容
        content = llm.run(content_prompt)
        return content

定义教程编写智能体
定义 TutorialAssistant 类，继承自 BaseAgent，用于生成教程内容。
主要功能包括：

• 初始化目录和内容生成的动作（WriteDirectoryAction 和 WriteContentAction）
• _generate_tutorial 方法根据目录数据生成完整的教程内容包括目录和每个章节的详细内容
• _add_tutorial_example 方法为助手添加一个示例任务并展示如何生成一个 Python 教程的目录和内容

最终调用 call 方法处理生成教程的任务。它从任务中提取主题，生成目录结构，然后生成完整的教程内容，并将结果保存到本地。

class TutorialAssistant(BaseAgent):
    """Tutorial generation assistant that manages directory and content creation"""
    def __init__(
        self,
        llm: LLM,
        language: str = "Chinese"
    ):
        name = "TutorialAssistant"
        role = """You are a professional tutorial writer. You can create well-structured, 
        comprehensive tutorials on various topics. You excel at organizing content logically 
        and explaining complex concepts clearly."""
        
        super().__init__(
            name=name,
            role=role,
            llm=llm,
        )
        
        self.language = language
        self.directory_action = WriteDirectoryAction()
        self.content_action = WriteContentAction()
    
        # Add example for the tutorial assistant
        self._add_tutorial_example()
        
    def _generate_tutorial(self, directory_data: Dict) -> str:
        """Generate complete tutorial content based on directory structure"""
        full_content = []
        title = directory_data["title"]
        full_content.append(f"# {title}\n")
        
        # Generate table of contents
        full_content.append("## 目录\n")
        for idx, chapter in enumerate(directory_data["directory"], 1):
            for chapter_title, sections in chapter.items():
                full_content.append(f"{idx}. {chapter_title}")
                for section_idx, section in enumerate(sections, 1):
                    full_content.append(f"   {idx}.{section_idx}. {section}")
        full_content.append("\n---\n")
        
        # Generate content for each section
        for chapter in directory_data["directory"]:
            for chapter_title, sections in chapter.items():
                for section in sections:
                    content = self.content_action(
                        title=section,
                        chapter=chapter_title,
                        directory_data=directory_data,
                        language=self.language
                    )
                    full_content.append(content)
                    full_content.append("\n---\n")
        
        return "\n".join(full_content)

    def __call__(self, task: TaskPackage):
        """Process the tutorial generation task"""
        # Extract topic from task
        topic = task.instruction.split("Create a ")[-1].split(" tutorial")[0]
        if not topic:
            topic = task.instruction
            
        # Generate directory structure
        directory_result = self.directory_action(
            topic=topic,
            language=self.language
        )

        print(directory_result)
        
        # Generate complete tutorial
        tutorial_content = self._generate_tutorial(directory_result["directory_data"])

        # Save the result
        task.answer = tutorial_content
        task.completion = "completed"
        
        return task

    def _add_tutorial_example(self):
        """Add an illustration example for the tutorial assistant"""
        exp_task = "Create a Python tutorial for beginners"
        exp_task_pack = TaskPackage(instruction=exp_task)
        topic = "Python基础教程"

        act_1 = AgentAct(
            name=ThinkAct.action_name,
            params={INNER_ACT_KEY: """First, I'll create a directory structure for the Python tutorial, 
            then generate detailed content for each section."""}
        )
        obs_1 = "OK. I'll start with the directory structure."

        act_2 = AgentAct(
            name=self.directory_action.action_name,
            params={
                "topic": topic, 
                "language": self.language
            }
        )
        obs_2 = """{"title": "Python基础教程", "directory": [
            {"第一章：Python介绍": ["1.1 什么是Python", "1.2 环境搭建"]},
            {"第二章：基础语法": ["2.1 变量和数据类型", "2.2 控制流"]}
        ]}"""

        act_3 = AgentAct(
            name=self.content_action.action_name,
            params={
                "title": "什么是Python",
                "chapter": "第一章：Python介绍",
                "directory_data": json.loads(obs_2),
                "language": self.language
            }
        )
        obs_3 = """# 第一章：Python介绍\n## 什么是Python\n\nPython是一种高级编程语言..."""

        act_4 = AgentAct(
            name=FinishAct.action_name,
            params={INNER_ACT_KEY: "Tutorial structure and content generated successfully."}
        )
        obs_4 = "Tutorial generation task completed successfully."

        exp_act_obs = [(act_1, obs_1), (act_2, obs_2), (act_3, obs_3), (act_4, obs_4)]
        
        self.prompt_gen.add_example(
            task=exp_task_pack,
            action_chain=exp_act_obs
        )

运行

if __name__ == "__main__":
    assistant = TutorialAssistant(llm=llm)

     # 交互式生成教程
    FLAG_CONTINUE = True
    while FLAG_CONTINUE:
        input_text = input("What tutorial would you like to create?\n")
        task = TaskPackage(instruction=input_text)
        result = assistant(task)
        print("\nGenerated Tutorial:\n")
        print(result.answer)

        # 创建输出目录
        output_dir = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        os.makedirs(output_dir, exist_ok=True)
        
        # 保存文件
        output_file = os.path.join(output_dir, f"{input_text}.md")
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(result.answer)
        if input("\nDo you want to create another tutorial? (y/n): ").lower() != "y":
            FLAG_CONTINUE = False

这里我运行的是

How to learn HTML?

生成的结果如链接所示How to learn html？基于chatGLM-b生成示例（仅供参考）运行时间比较长，中间也没有很多提示，记得耐心等待！

2. Zigent实现出题智能体

本节将通过Zigent框架实现一个出题智能体，其主要功能是根据指定的Markdown文件内容自动生成考卷。该智能体支持单选题、多选题和填空题三种题型，并能将生成的考卷保存为Markdown文件。

思路：

1. 从指定目录加载Markdown文件作为知识来源
2. 根据用户指定的受众群体和考察目的生成考卷
3. 支持多种题型（单选题、多选题、填空题）
4. 自动保存生成的考卷并返回结果路径

导入依赖

import os
from pathlib import Path
from datetime import datetime
from typing import Dict, List
import json

from zigent.llm.agent_llms import LLM
from zigent.actions import BaseAction, ThinkAct, FinishAct
from zigent.agents import BaseAgent
from zigent.commons import TaskPackage, AgentAct
from zigent.actions.InnerActions import INNER_ACT_KEY

定义出题Action

class QuizGenerationAction(BaseAction):
    """Generate quiz questions from markdown content"""
    def __init__(self, llm: LLM) -> None:
        action_name = "GenerateQuiz"
        action_desc = "Generate quiz questions from markdown content"
        params_doc = {
            "content": "(Type: string): The markdown content to generate questions from",
            "question_types": "(Type: list): List of question types to generate",
            "audience": "(Type: string): Target audience for the quiz",
            "purpose": "(Type: string): Purpose of the quiz"
        }
        super().__init__(action_name, action_desc, params_doc)
        self.llm = llm
        
    def __call__(self, **kwargs):
        content = kwargs.get("content", "")
        question_types = kwargs.get("question_types", [])
        audience = kwargs.get("audience", "")
        purpose = kwargs.get("purpose", "")
        
        prompt = f"""
        你是一个辅助设计考卷的机器人,全程使用中文。
        你的任务是帮助用户快速创建、设计考卷，考卷以markdown格式给出。
        
        要求：
        1. 受众群体：{audience}
        2. 考察目的：{purpose}
        3. 需要包含以下题型：{", ".join(question_types)}
        4. 考卷格式要求：
        """
        prompt += """
        # 问卷标题
        ---
        1. 这是判断题的题干?
            - (x) True
            - ( ) False
        # (x)为正确答案

        2. 这是单选题的题干
            - (x) 这是正确选项
            - ( ) 这是错误选项
        # (x)为正确答案

        3. 这是多选题的题干?
            - [x] 这是正确选项1
            - [x] 这是正确选项2
            - [ ] 这是错误选项1
            - [ ] 这是错误选项2
        # [x]为正确答案

        4. 这是填空题的题干?
            - R:= 填空题答案
        #填空题正确答案格式
        """
        
        prompt += f"\n请根据以下内容生成考卷：\n{content}"
        
        quiz_content = self.llm.run(prompt)
        return {
            "quiz_content": quiz_content,
            "audience": audience,
            "purpose": purpose,
            "question_types": question_types
        }

定义保存Action

class SaveQuizAction(BaseAction):
    """Save quiz to file and return URL"""
    def __init__(self) -> None:
        action_name = "SaveQuiz"
        action_desc = "Save quiz content to file and return URL"
        params_doc = {
            "quiz_content": "(Type: string): The quiz content to save",
            "quiz_title": "(Type: string): Title of the quiz"
        }
        super().__init__(action_name, action_desc, params_doc)
        
    def __call__(self, **kwargs):
        quiz_content = kwargs.get("quiz_content", "")
        quiz_title = kwargs.get("quiz_title", "quiz")
        
        output_dir = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        os.makedirs(output_dir, exist_ok=True)
        
        output_file = os.path.join(output_dir, f"{quiz_title}.md")
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(quiz_content)
            
        return {
            "file_path": output_file,
            "quiz_url": f"/{output_file}"
        }

定义出题智能体

class QuizGeneratorAgent(BaseAgent):
    """Quiz generation agent that manages quiz creation process"""
    def __init__(
        self,
        llm: LLM,
        markdown_dir: str
    ):
        name = "QuizGeneratorAgent"
        role = """你是一个专业的考卷生成助手。你可以根据提供的Markdown内容生成结构良好、
        内容全面的考卷。你擅长根据受众群体和考察目的设计合适的题目。"""
        
        super().__init__(
            name=name,
            role=role,
            llm=llm,
        )
        
        self.markdown_dir = markdown_dir
        self.quiz_action = QuizGenerationAction(llm)
        self.save_action = SaveQuizAction()
        
        self._add_quiz_example()
        
    def _load_markdown_content(self) -> str:
        """Load all markdown files from directory"""
        content = []
        for root, _, files in os.walk(self.markdown_dir):
            for file in files:
                if file.endswith(".md"):
                    with open(os.path.join(root, file), 'r', encoding='utf-8') as f:
                        content.append(f.read())
        return "\n".join(content)
        
    def __call__(self, task: TaskPackage):
        """Process the quiz generation task"""
        # Parse task parameters
        params = json.loads(task.instruction)
        audience = params.get("audience", "")
        purpose = params.get("purpose", "")
        question_types = params.get("question_types", [])
        
        # Load markdown content
        content = self._load_markdown_content()
        
        # Generate quiz
        quiz_result = self.quiz_action(
            content=content,
            question_types=question_types,
            audience=audience,
            purpose=purpose
        )
        
        # Save quiz
        save_result = self.save_action(
            quiz_content=quiz_result["quiz_content"],
            quiz_title="generated_quiz"
        )
        
        task.answer = {
            "quiz_content": quiz_result["quiz_content"],
            "quiz_url": save_result["quiz_url"]
        }
        task.completion = "completed"
        
        return task

使用

from dotenv import load_dotenv
load_dotenv()

api_key = os.getenv('ZISHU_API_KEY')
base_url = "http://43.200.7.56:8008/v1"
chat_model = "deepseek-chat"

llm = LLM(api_key=api_key, base_url=base_url, model_name=chat_model)

# 创建出题智能体
markdown_dir = "docs"  # 指定包含Markdown文件的目录
agent = QuizGeneratorAgent(llm=llm, markdown_dir=markdown_dir)

# 定义考卷参数
quiz_params = {
    "audience": "零基础", # 受众群体
    "purpose": "测试基础知识掌握情况", # 考察目的
    "question_types": ["单选题"] # 需要包含的题型
}

# 生成考卷
task = TaskPackage(instruction=json.dumps(quiz_params))
result = agent(task)

print("生成的考卷内容：")
print(result.answer["quiz_content"])
print(f"考卷路径: {result.answer['quiz_url']}")

输出