手把手搭建：WorkBuddy 文档审阅 + 纠错闭环工作流

你是企业内部专职技术文档审阅专员，仅依托绑定的知识库规则对待校验文档做合规审查，禁止自由创作、全文润色、主观化评价。
校验一共包含四个维度：
1.格式层级校验：检查标题层级混乱、Markdown语法错误、代码块未标注语言、表格格式不规范问题；
2.文字语法校验：排查错别字、标点符号错误、语句不通顺、口语化网络用语；
3.专业术语校验：比对知识库术语白名单，检查缩写不规范、自定义别名不统一、禁用词汇使用情况；
4.业务逻辑校验：核对参数范围、状态码、字段说明是否存在前后描述冲突、数值矛盾。

输出严格遵循以下要求：
1.最终结果只允许出现一张Markdown表格，无任何前置说明、结尾总结、问候语句；
2.表格固定表头：错误片段位置、错误原文、错误分类、修改建议、规范依据；
3.每一条错误必须精准标注所在章节大致位置，规范依据必须写明知识库对应的条款编号；
4.未发现任何错误时，表格仅保留表头，表格内容为空，禁止输出“文档合规无错误”这类文字描述。

本次待校验文档片段内容：
{{document_content}}

本次需要严格遵循的约束规则来源：绑定的三份企业内部知识库文档，不允许脱离知识库自定义规则进行审核。
请按照系统设定的四个校验维度完成全量检查，输出固定格式的Markdown纠错表格。

# 所需第三方依赖库安装命令
# pip install requests pandas openpyxl markdown python-dotenv

# WorkBuddy开放平台密钥配置
WB_ACCESS_KEY=你的平台访问密钥
WB_SECRET_KEY=你的平台密钥
WB_AGENT_ID=自定义文档审阅智能体ID
# 文件存储路径配置
ORIGIN_DOC_PATH=./docs/origin/login_api.md
CORRECT_REPORT_PATH=./report/doc_check_result.xlsx
FINAL_DOC_PATH=./docs/final/login_api_final.md
# 单次提交最大文本字符限制
MAX_CONTENT_LENGTH=3000

import os
import time
import pandas as pd
import requests
from dotenv import load_dotenv
from markdown import markdown

# 加载环境配置
load_dotenv()

class DocAuditWorkFlow:
    def __init__(self):
        self.access_key = os.getenv("WB_ACCESS_KEY")
        self.secret_key = os.getenv("WB_SECRET_KEY")
        self.agent_id = os.getenv("WB_AGENT_ID")
        self.origin_path = os.getenv("ORIGIN_DOC_PATH")
        self.report_path = os.getenv("CORRECT_REPORT_PATH")
        self.final_path = os.getenv("FINAL_DOC_PATH")
        self.max_len = int(os.getenv("MAX_CONTENT_LENGTH"))
        self.api_host = "https://open.workbuddy.qq.com/v1"

    def split_markdown_content(self):
        """文档预处理：将长文档按照标题分段，避免文本超限"""
        with open(self.origin_path, "r", encoding="utf-8") as f:
            content = f.read()
        sections = content.split("# ")
        content_list = []
        temp_text = ""
        for section in sections:
            if not section.strip():
                continue
            if len(temp_text + section) > self.max_len:
                content_list.append(temp_text)
                temp_text = "# " + section
            else:
                temp_text = temp_text + "# " + section
        if temp_text:
            content_list.append(temp_text)
        return content_list, content

    def create_audit_task(self, doc_segment):
        """调用WorkBuddy API创建异步文档校验任务"""
        url = f"{self.api_host}/agent/async_task/create"
        prompt_template = """
你是企业内部专职技术文档审阅专员，仅依托绑定的三份知识库规则做合规审查，禁止主观润色全文。
待校验文档片段：
{doc_content}
严格按照系统预设四个校验维度检查，仅输出固定五列表格格式的校验结果，无多余文字。
"""
        params = {
            "agent_id": self.agent_id,
            "access_key": self.access_key,
            "secret_key": self.secret_key,
            "user_prompt": prompt_template.format(doc_content=doc_segment)
        }
        resp = requests.post(url, json=params)
        return resp.json().get("data", {}).get("task_id")

    def get_task_result(self, task_id, wait_times=60, interval=2):
        """轮询监听任务执行状态，直到任务完成返回校验结果"""
        url = f"{self.api_host}/agent/async_task/get"
        for _ in range(wait_times):
            params = {
                "access_key": self.access_key,
                "secret_key": self.secret_key,
                "task_id": task_id
            }
            resp = requests.get(url, params=params)
            res_data = resp.json().get("data", {})
            status = res_data.get("task_status")
            if status == "SUCCESS":
                return res_data.get("task_result")
            elif status in ["FAIL", "CANCEL"]:
                return None
            time.sleep(interval)
        return None

    def merge_check_result(self, result_list):
        """合并多段文档校验结果，转换为结构化Excel纠错报告"""
        all_rows = []
        for md_table in result_list:
            try:
                df = pd.read_html(markdown(md_table))[0]
                all_rows.append(df)
            except Exception:
                continue
        if all_rows:
            total_df = pd.concat(all_rows, ignore_index=True)
            total_df.to_excel(self.report_path, index=False)
            return total_df
        return pd.DataFrame()

    def batch_fix_document(self, origin_content, audit_df):
        """读取纠错报告，批量替换原始文档错误内容"""
        fixed_content = origin_content
        for idx, row in audit_df.iterrows():
            error_text = str(row["错误原文"])
            fix_text = str(row["修改建议"])
            if pd.notna(error_text) and pd.notna(fix_text):
                fixed_content = fixed_content.replace(error_text, fix_text, 1)
        with open(self.final_path, "w", encoding="utf-8") as f:
            f.write(fixed_content)
        return fixed_content

    def run_workflow(self):
        """启动完整文档审阅纠错闭环工作流"""
        print("=====开始执行文档预处理分段=====")
        seg_list, origin_full = self.split_markdown_content()
        task_result_arr = []
        print(f"文档拆分完成，一共{len(seg_list)}个待校验片段")

        for index, seg in enumerate(seg_list):
            print(f"正在提交第{index+1}段文档校验任务")
            tid = self.create_audit_task(seg)
            if not tid:
                print(f"第{index+1}段任务创建失败，跳过该片段")
                continue
            res = self.get_task_result(tid)
            if res:
                task_result_arr.append(res)
            print(f"第{index+1}段文档校验执行完成")

        print("=====开始合并校验结果，生成纠错报告=====")
        audit_data = self.merge_check_result(task_result_arr)
        error_total = len(audit_data)
        print(f"本次文档一共检测到{error_total}处合规错误")

        if error_total > 0:
            print("=====开始批量自动修正原始文档=====")
            self.batch_fix_document(origin_full, audit_data)
            print(f"文档修正完成，终版文档存储路径：{self.final_path}")
        else:
            with open(self.final_path, "w", encoding="utf-8") as f:
                f.write(origin_full)
            print("文档无合规错误，直接归档原始文档")

if __name__ == "__main__":
    workflow = DocAuditWorkFlow()
    workflow.run_workflow()