利用CrewAI框架打造高效自动化周报系统：告别手动报告，迎接智能协作

发布日期：2024-04-21 08:03:32 浏览次数： 4473

作者：加加笔记

微信搜一搜，关注“加加笔记”

在日常工作中，编写个人和团队的工作周报是一项既繁琐又枯燥的任务。幸运的是，这种场景非常适合利用人工智能（AI）技术来简化。尽管市面上已有多种基于GPT的智能体，如国内的文心一言、讯飞星火、智谱清言等，它们能够对用户提供的内容进行总结，但如果没有用户输入，它们便无法凭空创造内容。此外，对于团队周报而言，市场上尚未出现能够有效汇总部门成员工作流的解决方案。

为了解决这一问题，我采用了多智能体协作的方法，并选择了CrewAI框架来构建一个自动化的秘书团队。CrewAI是一个专注于多智能体协作的框架，它在团队协作能力、通信接口、内置算法、并行计算、扩展性和应用场景等方面展现出显著优势。

想了解CrewAI，可以看下这篇公众号文章。《跟着我的步骤，轻松打造出 AI 智能体》

接下来，我将分享如何利用CrewAI框架创建多智能体应用，以实现自动整理、汇总和发送部门成员的周报。

准备工作：

首先，为了实现这一工作流，我们需要明确智能体的角色、任务以及所需工具。由于我们的目标是汇总个人周报并整合成团队周报，我开发了自定义工具来操作语雀文档和阿里云邮箱，这些工具均按照CrewAI的规范构建。

核心文件与智能体定义：

在CrewAI的核心文件中，agents.yaml和tasks.yaml定义了智能体和任务，而crew.py和main.py则定义了智能体的逻辑和输入信息。

agents.yaml：在此文件中，我们定义了两个智能体，分别用于获取个人工作周报并汇总为团队工作周报，以及发送邮件和发布语雀文档。例如：

writer:
 role: >
  工作周报填写者
 goal: >
  汇总每周的部门成员的工作周报，形成部门的工作周报。
 backstory: >
  你是一个经理秘书，负责汇总每周的部门成员的工作周报，形成部门的工作周报。你需要对每个成员的工作周报进行审核，确保每个成员的工作周报都是真实的，然后将每个成员的工作周报进行汇总，形成部门的工作周报。

  
sender:
 role: >
  邮件和语雀发送者
 goal: >
  发送邮件以及语雀文档给领导
 backstory: >
  你是一个邮件和语雀发送者，负责发送邮件以及语雀文档给领导。你需要将部门的工作周报发送给领导，确保邮件的内容是正确的，不要单独创建附件。

tasks.yaml：在此文件中，我们定义了两个任务，一个用于汇总周报内容，并按照Markdown格式组织文档；另一个任务则是将整理好的周报发送给领导。

writing_task:
 description: >
  汇总本部门成员的工作周报，形成部门的工作周报。
  请确保最终形成的文档是中文的。
  请尽量不要做内容的压缩，要保证包含所有的工作内容
  不要把名字列进入，只需要把他们干的工作写进去就行。
  调用成员周报获取工具只需要执行一次，不要反复去调用，因为每次获取的都是一样的。
  请保证完成部门周报的整体创作以后询问人类是否要发送邮件和发布语雀文档。  
  
  最终形成的部门的工作周报，需要按照以下的样例进行填写，输出markdown格式：
  
  ## 本周重点工作
  
  ### 产品A 相关
  
  * xxxx 5.5.6：本周完成了后段bug的修复。
  
  * XXXX 6 & 6.2 开发：计划进行 XXX 6.0 测试提的问题修复及 XXX 6.2 开发。
  
  ### XXX 项目
  
  * 进行了多项优化，包括告警规则、日志下载接口、节点管理界面等。
  
  * 计划进行项目维护和版本适配。
  
  ### 前端逻辑
  
  ...
  
  ## 下周计划
  
  * 继续进行 XXX 6 相关开发和问题处理。
  
  * XXX 3.0 页面组件后续开发。
  
  ...

 expected_output: >
  一篇markdown格式的部门的工作周报。

  

send_task:
 description: >
  发送邮件和语雀文档给相关领导。
  没有汇总完成之前，请不要发送邮件和发布语雀文档。
  发送邮件和发布语雀的content参数请保证是markdown格式的部门工作周报的内容。
    
  邮件内容示例：
  
  本周主要完成……工作，以下为详细内容：
  
  [汇总的部门工作周报]
  ***********************************************************************
  
  张三 /Zhangsan
  
  Mobile：86-13122223333；
  
  Email：zhagsan@example.com
  
  XXXXXXX有限公司 / XXXXXXXX Technology Ltd.
  ***********************************************************************
  
  语雀文档内容示例：
  
  [汇总的markdown格式部门工作周报]
 
 expected_output: >
 
  任务是否完成

智能体逻辑编排：

在crew.py中，我们通过定义WeeklyReportCrew类来编排智能体的工作流程。每个智能体都有自己的角色和目标，例如，writer智能体的目标是汇总部门成员的周报，而sender智能体则负责将周报发送给领导。

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from weekly_report.tools.ali_mail import SendEmailTool
from weekly_report.tools.send_yuque_doc import SendYuqueDocTool
from weekly_report.utils.llms import LLMs
from langchain.agents import load_tools

# Uncomment the following line to use an example of a custom tool
from weekly_report.tools.weekly_reports import GetWeeklyReportsTool

# Check our tools documentations for more information on how to use them
# from crewai_tools import SerperDevTool

llm = LLMs(model_name="glm-3-turbo").get_llm()
function_calling_llm = LLMs(model_name="gpt-3.5-turbo-0125").get_llm()

get_weekly_reports_tool = GetWeeklyReportsTool()
send_email_tool = SendEmailTool()
send_yuque_doc_tool = SendYuqueDocTool()
human_tools = load_tools(["human"])

@CrewBase
class WeeklyReportCrew():
 """WeeklyReport crew"""
 agents_config = 'config/agents.yaml'
 tasks_config = 'config/tasks.yaml'
 
 @agent
 def writer(self) -> Agent:
  return Agent(
   config=self.agents_config['writer'],
   tools=[get_weekly_reports_tool], 
   verbose=True,
   llm=llm, 
   function_calling_llm=function_calling_llm,
   allow_delegation=True
  )
 
 @agent
 def sender(self) -> Agent:
  return Agent(
   config=self.agents_config['sender'],
   tools=[send_email_tool, send_yuque_doc_tool],
   verbose=True,
   llm=llm, 
   function_calling_llm=function_calling_llm,
   allow_delegation=True
  )

 @task
 def writing_task(self) -> Task:
  return Task(
   config=self.tasks_config['writing_task'],
   agent=self.writer(),
   # context=[self.manage_task()]
  )
 
 @task
 def send_task(self) -> Task:
  return Task(
   config=self.tasks_config['send_task'],
   agent=self.sender(),
   context=[self.writing_task()]
  )

 @crew
 def crew(self) -> Crew:
  """Creates the WeeklyReport crew"""
  return Crew(
   agents=self.agents, # Automatically created by the @agent decorator
   tasks=self.tasks, # Automatically created by the @task decorator
   # process=Process.sequential,
   verbose=2,
   process=Process.hierarchical, # In case you wanna use that instead https://docs.crewai.com/how-to/Hierarchical/
   manager_llm=function_calling_llm,
  )

自定义工具开发：

在CrewAI框架中，自定义工具是通过Python类来实现的，这些类继承自BaseTool，并根据需要完成特定的任务。例如，SendEmailTool和SendYuqueDocTool是两个用于发送邮件和发布语雀文档的工具。

SendEmailTool：这个工具负责发送包含周报内容的邮件。它使用Python的smtplib库来通过阿里云企业邮箱发送邮件。工具的输入是一个包含邮件内容的字符串，输出是发送成功或失败的信息。

import os

from dotenv import load_dotenv
from crewai_tools import BaseTool
from pydantic.v1 import BaseModel, Field
from typing import Type, List

class SendEmailInput(BaseModel):
    content: str = Field(..., title="邮件正文", description="周报的具体内容")

class SendEmailTool(BaseTool):
    name: str = "发送邮件工具"
    description: str = "用于发送部门周报邮件，标题和收件人都不需要填写，固定为AI&UI小组工作周报"
    args_schema: Type[BaseModel] = SendEmailInput

    def send_email(self, content: str):
        load_dotenv()
        # 配置阿里企业邮箱
        mail_host = "smtp.qiye.aliyun.com"
        sender = os.getenv("ALI_SENDER")
        passwd = os.getenv("ALI_PASSWD")
        import smtplib
        from email.mime.text import MIMEText
        from email.header import Header
        from email.utils import formataddr
        receivers = os.getenv("ALI_RECEIVERS")
        # 用markdown格式发送邮件
        content = f"""{content}"""
        import markdown2
        content = markdown2.markdown(content)
        print(content)
        print(sender)
        print(receivers)
        msg = MIMEText(content, 'html', 'utf-8')
        msg['From'] = formataddr(["xdfs", sender])
        msg['To'] = receivers
        msg['Subject'] = Header("AI&UI小组工作周报", 'utf-8').encode()
        print(msg)
        to_list = receivers.split(',')
        try:
            server = smtplib.SMTP_SSL(mail_host, 465)
            server.login(sender, passwd)
            server.sendmail(sender, to_list, msg.as_string())
            server.close()
            return '邮件发送成功'
        except Exception as e:
            return '邮件发送失败 %s' % e
    
    def _run(self, content: str) -> str:
        load_dotenv()
        result = self.send_email(content)
        return result

SendYuqueDocTool：这个工具用于将Markdown格式的周报发布到语雀知识库中。它通过语雀的API来创建新的文档，并将周报内容作为文档正文。工具的输入是Markdown格式的周报内容，输出是发布成功或失败的信息。

from crewai_tools import BaseTool
from dotenv import load_dotenv
import os
import requests

class SendYuqueDocTool(BaseTool):
    name: str = "发送语雀文档工具"
    description: str = "用于发送或者发布语雀文档到语雀知识库的工具"

    def get_this_friday(self):
        import datetime
        today = datetime.date.today()
        today_weekday = today.weekday()
        print(today_weekday)
        if today_weekday == 4:
            return today
        elif today_weekday > 4:
            return today - datetime.timedelta(days=(today_weekday-4) % 7)
        else:
            return today + datetime.timedelta(days=(4 - today_weekday) % 7)
    

    def _run(self, content: str) -> str:
        load_dotenv()
        auth_token = os.getenv("YUQUE_AUTH_TOKEN")
        login = os.getenv("YUQUE_LOGIN")
        slug = os.getenv("YUQUE_SLUG")
        # 获取周报所在目录的uuid
        get_toc_url = f"https://www.yuque.com/api/v2/repos/{login}/{slug}/toc"
        header = {"X-Auth-Token": auth_token}
        res_toc = requests.get(get_toc_url, headers=header)
        toc = res_toc.json()
        this_friday = self.get_this_friday().strftime("%Y%m%d")
        print(this_friday)
        target_uuid = ""
        for item in toc["data"]:
            if item["type"] == "TITLE" and item["title"] == this_friday:
                target_uuid = item["uuid"]
                break
        print(target_uuid)
        # 需要单独调用更新目录接口才能更新文档到目录
        if target_uuid == "":
            return "未找到周报目录"
        create_doc_url = f"https://www.yuque.com/api/v2/repos/{login}/{slug}/docs"
        header = {"X-Auth-Token": auth_token}
        data = {
            "title": "部门工作周报",
            "format": "markdown",
            "body": content
        }
        created_article = requests.post(create_doc_url, headers=header, data=data)
        update_toc_url = f"https://www.yuque.com/api/v2/repos/{login}/{slug}/toc"
        data = {
            "action": "appendNode",
            "action_mode": "child",
            "type": "DOC",
            "doc_ids": [created_article.json()["data"]["id"]],
            "target_uuid": target_uuid
        }
        response = requests.put(update_toc_url, headers=header, json=data)
        return "创建语雀文档成功！"

运行流程：

获取周报：writer智能体首先使用GetWeeklyReportsTool工具从语雀中获取团队成员的个人周报。
汇总周报：然后，writer智能体将这些信息汇总成一份完整的团队周报，按照预定义的Markdown模板进行格式化。
发送周报：sender智能体在确认周报内容无误后，使用SendEmailTool将周报通过邮件发送给领导，并通过SendYuqueDocTool将周报发布到语雀知识库中。

避坑指南：

如果你不使用OpenAI的LLM，需要明确指定所使用的LLM。
如果使用国内的LLM，务必指定一个国外的LLM作为function_calling_llm，以避免潜在的错误。
尽管目前的任务是串行的，但使用hierarchical进程模式会比sequential更有效。
对于生成内容，最好提供明确的模板，以确保内容的准确性和一致性。

通过上述步骤，我们成功构建了一个自动化的秘书团队，它能够高效地完成周报的整理、汇总和发送工作。这不仅节省了大量的人力资源，也提高了工作效率和准确性。希望这篇文章能够帮助你在实际工作中应用AI技术，实现工作流程的自动化。

53AI，企业落地大模型首选服务商

产品：场景落地咨询+大模型应用平台+行业解决方案

承诺：免费POC验证，效果达标后再合作。零风险落地应用大模型，已交付160+中大型企业