#!/usr/bin/env python

-- coding: utf-8 --

”“” 信息提取Agent完整示例 “””

import os import json from dotenv import load_dotenv from xparse_client import create_pipeline_from_config, Pipeline, LocalSource from langchain_core.tools import Tool from langchain.agents import create_agent from langchain_community.chat_models import ChatTongyi

load_dotenv()

class InformationExtractionAgent:

GPT plus 代充 只需 145"""信息提取Agent""" def __init__(self): self.setup_pipeline() self.setup_llm() self.setup_agent() # 文档文本存储（实际应用中可以使用更持久化的存储） self._document_texts = {} def setup_pipeline(self): """配置Pipeline""" self.pipeline_config = { "source": { "type": "local", "directory": "./extraction_documents", "pattern": ["*.pdf", "*.docx", "*.xlsx", "*.xls", "*.png", "*.jpg"] }, "destination": { "type": "local", "output_dir": "./extraction_results" }, "api_base_url": "https://api.textin.com/api/xparse", "api_headers": , "stages": [ { "type": "parse", "config": {"provider": "textin"} } ] } def setup_llm(self): """初始化大模型""" self.llm = ChatTongyi( model="qwen-max", dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"), temperature=0, # 使用较低温度以获得更确定性的输出 ) def parse_document(self, file_path: str) -> list: """ 解析单个文档，返回 elements 列表 Args: file_path: 文档路径 Returns: list: elements 列表 """ import json import os from copy import deepcopy # 创建临时配置，使用包含该文件的目录作为source file_dir = os.path.dirname(os.path.abspath(file_path)) file_name_pattern = os.path.basename(file_path) # 创建临时Pipeline配置 temp_config = deepcopy(self.pipeline_config) temp_config["source"] = { "type": "local", "directory": file_dir, "pattern": [file_name_pattern] # 只处理指定的文件 } # 创建Pipeline并运行（pipeline.run() 没有返回值，结果会保存到destination） pipeline = create_pipeline_from_config(temp_config) pipeline.run() # 从destination配置的输出目录读取解析结果 output_dir = self.pipeline_config["destination"]["output_dir"] # 确保输出目录存在 os.makedirs(output_dir, exist_ok=True) # 获取文件名（不含路径和扩展名） file_name = os.path.splitext(os.path.basename(file_path))[0] result_file = os.path.join(output_dir, f"{file_name}.json") # 读取JSON文件 if not os.path.exists(result_file): raise FileNotFoundError( f"解析结果文件不存在: {result_file} 

“

 f"请检查输出目录: {output_dir} 

”

GPT plus 代充 只需 145 f"原始文件路径: {file_path}" ) with open(result_file, 'r', encoding='utf-8') as f: elements = json.load(f) return elements def aggregate_text_from_elements(self, elements: list) -> str: """ 聚合 elements 中的 text 字段，形成完整文档文本 Args: elements: elements 列表 Returns: str: 聚合后的完整文本 """ texts = [] for element in elements: if isinstance(element, dict): text = element.get('text', '') else: text = getattr(element, 'text', '') if text and text.strip(): texts.append(text.strip()) return " 

”.join(texts)

def load_document(self, file_path: str): """ 加载并解析文档，将文本内容存储到内存中 Args: file_path: 文档路径 """ print(f"正在解析文档: {file_path}") elements = self.parse_document(file_path) text = self.aggregate_text_from_elements(elements) self._document_texts[file_path] = text print(f"文档解析完成，文本长度: {len(text)} 字符") def get_document_text(self, file_path: str = None) -> str: """ 获取文档文本内容 Args: file_path: 文档路径（可选），如果不提供则返回第一个文档的文本 Returns: str: 文档文本内容 """ if not self._document_texts: return "" # 如果没有加载任何文档，返回空字符串 if file_path not in ("None", "none", None, "", "null"): return self._document_texts.get(file_path, "") # 如果没有指定文件，返回第一个文档的文本 return next(iter(self._document_texts.values()), "") def setup_agent(self): """配置Agent和Tools""" tools = [ Tool( name="extract_invoice_info", description="从发票中提取结构化信息，包括发票基本信息、销售方/购买方信息、商品明细、金额信息等。如果已加载文档，可以直接调用无需参数；否则需要提供文档路径。", func=self.extract_invoice_info ), Tool( name="extract_medical_bill_info", description="从医疗票据中提取结构化信息，包括患者信息、医疗机构信息、就诊信息、费用明细、费用汇总等。如果已加载文档，可以直接调用无需参数；否则需要提供文档路径。", func=self.extract_medical_bill_info ), Tool( name="extract_contract_info", description="从合同中提取结构化信息，包括合同基本信息、合同双方信息、合同标的、关键条款、金额信息等。如果已加载文档，可以直接调用无需参数；否则需要提供文档路径。", func=self.extract_contract_info ), Tool( name="extract_resume_info", description="从简历中提取结构化信息，包括个人信息、教育经历、工作经历、技能等。如果已加载文档，可以直接调用无需参数；否则需要提供文档路径。", func=self.extract_resume_info ), Tool( name="extract_product_specs", description="从产品文档中提取产品规格和技术参数，包括产品名称、型号、技术参数、功能特性、价格等。如果已加载文档，可以直接调用无需参数；否则需要提供文档路径。", func=self.extract_product_specs ), Tool( name="extract_api_info", description="从技术文档中提取API接口信息，包括API端点、请求方法、请求参数、响应格式等。如果已加载文档，可以直接调用无需参数；否则需要提供文档路径。", func=self.extract_api_info ), Tool( name="format_data", description="将提取的数据格式化为标准格式（JSON、CSV等）。如果已加载文档，可以直接调用无需参数；否则需要提供文档路径。", func=self.format_data ) ] self.agent = create_agent( model=self.llm, tools=tools, debug=True, system_prompt="""你是一个专业的信息提取助手。你的任务是帮助用户： 1. 从文档中提取结构化信息（发票、医疗票据、合同、简历、产品规格、API接口等） 2. 将提取的信息格式化为标准格式（JSON、CSV等） 3. 验证提取数据的完整性和准确性 重要提示： - 在使用提取工具之前，确保文档已经通过 load_document() 方法加载 - 如果工具返回"未找到文档内容"的错误，说明需要先加载文档 - 可以直接调用提取工具，无需提供文档路径（如果文档已加载） 在回答时，请： - 提供结构化的提取结果 - 使用JSON或表格格式展示数据 - 如果数据不完整，说明缺失的部分 - 使用工具获取准确的信息，不要猜测 - 对于财务类文档（发票、医疗票据），确保金额和税务信息的准确性 - 对于合同文档，重点关注关键条款和风险点 """ ) def extract_invoice_info(self, file_path: str = None) -> str: """提取发票信息（使用 qwen-max 大模型）""" context_text = self.get_document_text(file_path) # 如果没有文档文本，返回提示信息 if not context_text: return "错误：未找到文档内容。请先使用 load_document() 方法加载文档，或提供文档路径。" prompt = f"""请从以下发票文本中提取结构化信息，并以 JSON 格式返回。 要求提取的信息包括： 1. 发票基本信息：invoice_code（发票代码）、invoice_number（发票号码）、date（开票日期） 2. 销售方信息：name（名称）、tax_id（纳税人识别号）、address（地址电话）、bank_account（开户行及账号） 3. 购买方信息：name（名称）、tax_id（纳税人识别号）、address（地址电话）、bank_account（开户行及账号） 4. 商品明细（数组）：name、specification、unit、quantity、unit_price、amount、tax_rate、tax_amount 5. 金额信息：total_amount（合计金额）、tax_amount（合计税额）、total_with_tax（价税合计） 6. 其他信息：remark（备注）、payee（收款人）、reviewer（复核人）、drawer（开票人） 请严格按照以下 JSON 格式返回，如果某个字段不存在，请使用空字符串 "" 或空对象 {{}} 或空数组 []： {{ "invoice_info": {{"invoice_code": "", "invoice_number": "", "date": ""}}, "seller": {{"name": "", "tax_id": "", "address": "", "bank_account": ""}}, "buyer": {{"name": "", "tax_id": "", "address": "", "bank_account": ""}}, "items": [}], "amounts": {{"total_amount": "", "tax_amount": "", "total_with_tax": ""}}, "other_info": {{"remark": "", "payee": "", "reviewer": "", "drawer": ""}} }} 发票文本内容： {context_text} 请只返回 JSON 格式的数据，不要包含任何其他解释或说明文字。""" return self._extract_with_llm(prompt) def _extract_with_llm(self, prompt: str) -> str: """ 通用的大模型提取方法，处理JSON解析和错误处理 Args: prompt: 发送给大模型的prompt Returns: str: JSON格式的提取结果 """ try: from langchain_core.messages import HumanMessage response = self.llm.invoke([HumanMessage(content=prompt)]) response_text = response.content.strip() if response.content else "" # 提取 JSON（处理 markdown 代码块） if "json" in response_text: json_start = response_text.find("json") + 7 json_end = response_text.find("", json_start) if json_end != -1: response_text = response_text[json_start:json_end].strip() elif "" in response_text: json_start = response_text.find("") + 3 json_end = response_text.find("", json_start) if json_end != -1: response_text = response_text[json_start:json_end].strip() # 验证是否为有效JSON data = json.loads(response_text) return json.dumps(data, ensure_ascii=False, indent=2) except json.JSONDecodeError as e: return json.dumps(, ensure_ascii=False, indent=2) except Exception as e: return json.dumps(, ensure_ascii=False, indent=2) def extract_resume_info(self, file_path: str = None) -> str: """提取简历信息（使用 qwen-max 大模型）""" context_text = self.get_document_text(file_path) # 如果没有文档文本，返回提示信息 if not context_text: return "错误：未找到文档内容。请先使用 load_document() 方法加载文档，或提供文档路径。" prompt = f"""请从以下简历文本中提取结构化信息，包括个人信息、教育经历、工作经历、技能等，并以 JSON 格式返回。 简历文本内容： {context_text} 请只返回 JSON 格式的数据，不要包含任何其他解释或说明文字。""" return self._extract_with_llm(prompt) def extract_product_specs(self, file_path: str = None) -> str: """提取产品规格（使用 qwen-max 大模型）""" context_text = self.get_document_text(file_path) # 如果没有文档文本，返回提示信息 if not context_text: return "错误：未找到文档内容。请先使用 load_document() 方法加载文档，或提供文档路径。" prompt = f"""请从以下产品文档文本中提取产品规格和技术参数，包括产品名称、型号、技术参数、功能特性、价格等，并以 JSON 格式返回。 产品文档文本内容： {context_text} 请只返回 JSON 格式的数据，不要包含任何其他解释或说明文字。""" return self._extract_with_llm(prompt) def extract_api_info(self, file_path: str = None) -> str: """提取API信息（使用 qwen-max 大模型）""" context_text = self.get_document_text(file_path) # 如果没有文档文本，返回提示信息 if not context_text: return "错误：未找到文档内容。请先使用 load_document() 方法加载文档，或提供文档路径。" prompt = f"""请从以下技术文档文本中提取API接口信息，包括API端点、请求方法、请求参数、响应格式、认证方式等，并以 JSON 格式返回。 技术文档文本内容： {context_text} 请只返回 JSON 格式的数据，不要包含任何其他解释或说明文字。""" return self._extract_with_llm(prompt) def extract_medical_bill_info(self, file_path: str = None) -> str: """提取医疗票据信息（使用 qwen-max 大模型）""" context_text = self.get_document_text(file_path) # 如果没有文档文本，返回提示信息 if not context_text: return "错误：未找到文档内容。请先使用 load_document() 方法加载文档，或提供文档路径。" prompt = f"""请从以下医疗票据文本中提取结构化信息，包括患者信息、医疗机构信息、就诊信息、费用明细、费用汇总等，并以 JSON 格式返回。 医疗票据文本内容： {context_text} 请只返回 JSON 格式的数据，不要包含任何其他解释或说明文字。""" return self._extract_with_llm(prompt) def extract_contract_info(self, file_path: str = None) -> str: """提取合同信息（使用 qwen-max 大模型）""" context_text = self.get_document_text(file_path) # 如果没有文档文本，返回提示信息 if not context_text: return "错误：未找到文档内容。请先使用 load_document() 方法加载文档，或提供文档路径。" prompt = f"""请从以下合同文本中提取结构化信息，包括合同基本信息、合同双方信息、合同标的、关键条款、金额信息等，并以 JSON 格式返回。 合同文本内容： {context_text} 请只返回 JSON 格式的数据，不要包含任何其他解释或说明文字。""" return self._extract_with_llm(prompt) def format_data(self, file_path: str = None) -> str: """数据格式化""" import pandas as pd context_text = self.get_document_text(file_path) # 如果没有文档文本，返回提示信息 if not context_text: return "错误：未找到文档内容。请先使用 load_document() 方法加载文档，或提供文档路径。" prompt = f"""请从以下文本中提取所有键值对信息，并以 JSON 格式返回。 要求： 1. 提取所有形如"键：值"或"键:值"的键值对 2. 返回格式为数组，每个元素包含 key、value、source（来源文件名） 文本内容： {context_text} 请返回 JSON 格式的数组，格式如下： [ {{ "key": "键名", "value": "值", "source": "文件名" }} ] 请只返回 JSON 格式的数据，不要包含任何其他解释或说明文字。""" try: result_json = self._extract_with_llm(prompt) data_list = json.loads(result_json) if isinstance(data_list, dict) and "error" in data_list: return result_json if not data_list: return "未找到可格式化的数据" # 格式化为CSV try: df = pd.DataFrame(data_list) csv_output = df.to_csv(index=False) return f"JSON格式： 

{result_json}

CSV格式： {csv_output}”

GPT plus 代充 只需 145 except Exception as e: return f"JSON格式： 

{result_json}

CSV格式化失败：{str(e)}”

 except Exception as e: return f"数据格式化失败：{str(e)}" def query(self, question: str) -> str: """查询Agent""" from langchain_core.messages import HumanMessage response = self.agent.invoke({ "messages": [HumanMessage(content=question)] }) return response["messages"][-1].content 

def main():

GPT plus 代充 只需 145"""主函数""" agent = InformationExtractionAgent() # 1. 加载文档 document_path = "./extraction_documents/invoice.pdf" # 示例路径 if os.path.exists(document_path): agent.load_document(document_path) # 2. 查询示例 questions = [ "从发票中提取发票代码、发票号码、销售方和购买方信息、商品明细和金额", # "从医疗票据中提取患者信息、医院信息、诊断结果和费用明细", # "从合同中提取合同编号、合同双方信息、合同金额和关键条款", # "从简历中提取所有个人信息、教育经历和工作经历", # "从产品文档中提取产品规格和技术参数", # "从技术文档中提取所有API接口信息", "将提取的数据格式化为JSON格式" ] for question in questions: print(f" 

{‘=’*60}“)

 print(f"问题: {question}") print(f"{'='*60}") answer = agent.query(question) print(f" 

回答: {answer}”)

if name == “main”:

GPT plus 代充 只需 145main()