LlamaIndex ：企业级知识助理，万物可知

from llama_index.indices.managed.llama_cloud import LlamaCloudIndexfrom llama_parse import LlamaParsefrom typing import List, Optionalfrom pydantic import BaseModel, Fieldfrom llama_index.llms.openai import OpenAI
# Setup Indexindex = LlamaCloudIndex(  name="gdpr",   project_name="llamacloud_demo",  organization_id="cdcb3478-1348-492e-8aa0-25f47d1a3902",  # api_key="llx-...")
# Setup RAG proxy# similarity_top_k=2，相邻前2回进行提取retriever = index.as_retriever(similarity_top_k=2)
# Setup Parserparser = LlamaParse(result_type="markdown")
# Setup env，包含大模型，parser proxy，RAG proxyllm = OpenAI(model="gpt-4o")workflow = ContractReviewWorkflow(    parser=parser,    guideline_retriever=retriever,    llm=llm,    verbose=True,    timeout=None, )
# Setup 合同 Output 格式class ContractClause(BaseModel):    clause_text: str = Field(..., description="The exact text of the clause.")    mentions_data_processing: bool = Field(False, description="True if the clause involves personal data collection or usage.")    mentions_data_transfer: bool = Field(False, description="True if the clause involves transferring personal data, especially to third parties or across borders.")    requires_consent: bool = Field(False, description="True if the clause explicitly states that user consent is needed for data activities.")    specifies_purpose: bool = Field(False, description="True if the clause specifies a clear purpose for data handling or transfer.")    mentions_safeguards: bool = Field(False, description="True if the clause mentions security measures or other safeguards for data.")
class ContractExtraction(BaseModel):    vendor_name: Optional[str] = Field(None, description="The vendor's name if identifiable.")    effective_date: Optional[str] = Field(None, description="The effective date of the agreement, if available.")    governing_law: Optional[str] = Field(None, description="The governing law of the contract, if stated.")    clauses: List[ContractClause] = Field(..., description="List of extracted clauses and their relevant indicators.")

# Setup 合同检查内容class GuidelineMatch(BaseModel):    guideline_text: str = Field(..., description="The single most relevant guideline excerpt related to this clause.")    similarity_score: float = Field(..., description="Similarity score indicating how closely the guideline matches the clause, e.g., between 0 and 1.")    relevance_explanation: Optional[str] = Field(None, description="Brief explanation of why this guideline is relevant.")# 级联检查项class ClauseComplianceCheck(BaseModel):    clause_text: str = Field(..., description="The exact text of the clause from the contract.")    matched_guideline: Optional[GuidelineMatch] = Field(None, description="The most relevant guideline extracted via vector retrieval.")    compliant: bool = Field(..., description="Indicates whether the clause is considered compliant with the referenced guideline.")    notes: Optional[str] = Field(None, description="Additional commentary or recommendations.")
# Setup Contract Review Workflow# 1.从知识库协议中提取结构化数据。# 2.对于每个条款，根据 Setup 合同检查内容进行检索，看其是否符合准则。# 3.生成最终摘要和判断结果。from llama_index.core.workflow import (    Event,    StartEvent,    StopEvent,    Context,    Workflow,    step,)from llama_index.core.llms import LLMfrom typing import Optionalfrom pydantic import BaseModelfrom llama_index.core import SimpleDirectoryReaderfrom llama_index.core.schema import Documentfrom llama_index.core.agent import FunctionCallingAgentWorkerfrom llama_index.core.prompts import ChatPromptTemplatefrom llama_index.core.llms import ChatMessage, MessageRolefrom llama_index.core.retrievers import BaseRetrieverfrom pathlib import Pathimport loggingimport jsonimport os
# 设置日志_logger = logging.getLogger(__name__)_logger.setLevel(logging.INFO)# 开始设置 prompt# 提取内容 promptCONTRACT_EXTRACT_PROMPT = """\You are given contract data below. \Please extract out relevant information from the contract into the defined schema - the schema is defined as a function call.\
{contract_data}"""# 内容和知识库匹配 promptCONTRACT_MATCH_PROMPT = """\Given the following contract clause and the corresponding relevant guideline text, evaluate the compliance \and provide a JSON object that matches the ClauseComplianceCheck schema.
**Contract Clause:**{clause_text}
**Matched Guideline Text(s):**{guideline_text}"""# 级联检查项 promptCOMPLIANCE_REPORT_SYSTEM_PROMPT = """\You are a compliance reporting assistant. Your task is to generate a final compliance report \based on the results of clause compliance checks against \a given set of guidelines. 
Analyze the provided compliance results and produce a structured report according to the specified schema. Ensure that if there are no noncompliant clauses, the report clearly indicates full compliance."""# 报告输出格式 promptCOMPLIANCE_REPORT_USER_PROMPT = """\A set of clauses within a contract were checked against GDPR compliance guidelines for the following vendor: {vendor_name}. The set of noncompliant clauses are given below.
Each section includes:- **Clause:** The exact text of the contract clause.- **Guideline:** The relevant GDPR guideline text.- **Compliance Status:** Should be `False` for noncompliant clauses.- **Notes:** Additional information or explanations.
{compliance_results}
Based on the above compliance results, generate a final compliance report following the `ComplianceReport` schema below. If there are no noncompliant clauses, the report should indicate that the contract is fully compliant."""

class ContractExtractionEvent(Event):    contract_extraction: ContractExtractionclass MatchGuidelineEvent(Event):    clause: ContractClauseclass MatchGuidelineResultEvent(Event):    result: ClauseComplianceCheckclass GenerateReportEvent(Event):    match_results: List[ClauseComplianceCheck]class LogEvent(Event):    msg: str    delta: bool = False
# 工作流核心代码class ContractReviewWorkflow(Workflow):    """Contract review workflow."""
    def __init__(self, parser: LlamaParse, guideline_retriever: BaseRetriever,        llm: LLM | None = None, similarity_top_k: int = 20, output_dir: str = "data_out",        **kwargs,) -> None:        """Init params."""        super().__init__(**kwargs)
		# 拿前面设置好的 llamaIndex 组件和环境        self.parser = parser        self.guideline_retriever = guideline_retriever        self.llm = llm or OpenAI(model="gpt-4o-mini")        self.similarity_top_k = similarity_top_k
        # if not exists, create        out_path = Path(output_dir) / "workflow_output"        if not out_path.exists():            out_path.mkdir(parents=True, exist_ok=True)            os.chmod(str(out_path), 0o0777)        self.output_dir = out_path
    @step    async def parse_contract(self, ctx: Context, ev: StartEvent) -> ContractExtractionEvent:        # load output template file        contract_extraction_path = Path(            f"{self.output_dir}/contract_extraction.json"        )        if contract_extraction_path.exists():            if self._verbose:                ctx.write_event_to_stream(LogEvent(msg=">> Loading contract from cache"))            contract_extraction_dict = json.load(open(str(contract_extraction_path), "r"))            contract_extraction = ContractExtraction.model_validate(contract_extraction_dict)        else:            if self._verbose:                ctx.write_event_to_stream(LogEvent(msg=">> Reading contract"))
            # 设置 llamaParam 解析文档            docs = SimpleDirectoryReader(input_files=[ev.contract_path]).load_data()
            # 构造提取内容 prompt            prompt = ChatPromptTemplate.from_messages([                ("user", CONTRACT_EXTRACT_PROMPT)            ])			# 等待 LLM 返回结果，参数包含输入的文档，模型，prompt            contract_extraction = await llm.astructured_predict(                ContractExtraction,                prompt,                contract_data="\n".join([d.get_content(metadata_mode="all") for d in docs])            )            if not isinstance(contract_extraction, ContractExtraction):                raise ValueError(f"Invalid extraction from contract: {contract_extraction}")            # save output template to file            with open(contract_extraction_path, "w") as fp:                fp.write(contract_extraction.model_dump_json())        if self._verbose:            ctx.write_event_to_stream(LogEvent(msg=f">> Contract data: {contract_extraction.dict()}"))
        return ContractExtractionEvent(contract_extraction=contract_extraction)
    @step    async def dispatch_guideline_match(self, ctx: Context, ev: ContractExtractionEvent) -> MatchGuidelineEvent:        """For each clause in the contract, find relevant guidelines.        Use a map-reduce pattern.         """        await ctx.set("num_clauses", len(ev.contract_extraction.clauses))        await ctx.set("vendor_name", ev.contract_extraction.vendor_name)
        for clause in ev.contract_extraction.clauses:            ctx.send_event(MatchGuidelineEvent(clause=clause, vendor_name=ev.contract_extraction.vendor_name))
    # 匹配知识库内容    @step    async def handle_guideline_match(self, ctx: Context, ev: MatchGuidelineEvent) -> MatchGuidelineResultEvent:        """Handle matching clause against guideline."""
        # 构造查询 prompt        query = """Please find the relevant guideline from {ev.vendor_name} that aligns with the following contract clause:{ev.clause.clause_text}"""		# 查询知识库 Embedding        guideline_docs = self.guideline_retriever.retrieve(query)        guideline_text="\n\n".join([g.get_content() for g in guideline_docs])        if self._verbose:            ctx.write_event_to_stream(                LogEvent(msg=f">> Found guidelines: {guideline_text[:200]}...")            )
        # 提取知识库相关内容        prompt = ChatPromptTemplate.from_messages([("user", CONTRACT_MATCH_PROMPT)])		# 等待 LLM 处理知识库内容和输入内容正确性，参数包含检查代理，prompt，知识库dump出来的graph，输入需要匹配文本        compliance_output = await llm.astructured_predict(            ClauseComplianceCheck,            prompt,            clause_text=ev.clause.model_dump_json(),            guideline_text=guideline_text
        )
        if not isinstance(compliance_output, ClauseComplianceCheck):            raise ValueError(f"Invalid compliance check: {compliance_output}")
        return MatchGuidelineResultEvent(result=compliance_output)
    # 匹配结果    @step    async def gather_guideline_match(self, ctx: Context, ev: MatchGuidelineResultEvent) -> GenerateReportEvent:        """Handle matching clause against guideline."""        num_clauses = await ctx.get("num_clauses")        events = ctx.collect_events(ev, [MatchGuidelineResultEvent] * num_clauses)        if events is None:            return
        match_results = [e.result for e in events]        # save match results        match_results_path = Path(            f"{self.output_dir}/match_results.jsonl"        )        with open(match_results_path, "w") as fp:            for mr in match_results:                fp.write(mr.model_dump_json() + "\n")

        return GenerateReportEvent(match_results=[e.result for e in events])
    # 输出    @step    async def generate_output(self, ctx: Context, ev: GenerateReportEvent) -> StopEvent:        if self._verbose:            ctx.write_event_to_stream(LogEvent(msg=">> Generating Compliance Report"))
        # if all clauses are compliant, return a compliant result        non_compliant_results = [r for r in ev.match_results if not r.compliant]
        # generate compliance results string        result_tmpl = """1. **Clause**: {clause}2. **Guideline:** {guideline}3. **Compliance Status:** {compliance_status}4. **Notes:** {notes}"""        non_compliant_strings = []        for nr in non_compliant_results:            non_compliant_strings.append(                result_tmpl.format(                    clause=nr.clause_text,                    guideline=nr.matched_guideline.guideline_text,                    compliance_status=nr.compliant,                    notes=nr.notes                )            )        non_compliant_str = "\n\n".join(non_compliant_strings)
        prompt = ChatPromptTemplate.from_messages([            ("system", COMPLIANCE_REPORT_SYSTEM_PROMPT),            ("user", COMPLIANCE_REPORT_USER_PROMPT)        ])        compliance_report = await llm.astructured_predict(            ComplianceReport,            prompt,            compliance_results=non_compliant_str,            vendor_name=await ctx.get("vendor_name")        )
        return StopEvent(result={"report": compliance_report, "non_compliant_results": non_compliant_results})