跳到主要内容

PageIndex 实战案例

真实应用场景和示例代码

案例 1：财务报告分析​

场景​

分析上市公司年报，提取关键财务数据。

代码示例​

from pageindex import PageIndexClient
import json

# 初始化
client = PageIndexClient(workspace="./financial_workspace")

# 索引年报 PDF
doc_id = client.index("2025_annual_report.pdf")

# 获取文档结构
structure = json.loads(client.get_document_structure(doc_id))

# 找到"财务数据"章节
def find_section(structure, keyword):
    """递归查找包含关键词的章节"""
    for node in structure:
        if keyword in node.get('title', '') or keyword in node.get('summary', ''):
            return node
        if node.get('nodes'):
            result = find_section(node['nodes'], keyword)
            if result:
                return result
    return None

# 查找财务章节
financial_section = find_section(structure, "Financial")
if financial_section:
    print(f"找到财务章节：{financial_section['title']}")
    print(f"页码范围：{financial_section['start_index']}-{financial_section['end_index']}")
    
    # 获取该章节内容
    pages = f"{financial_section['start_index']}-{financial_section['end_index']}"
    content = json.loads(client.get_page_content(doc_id, pages))
    
    for page in content:
        print(f"\n=== 第 {page['page']} 页 ===")
        print(page['content'][:500])

输出示例​

找到财务章节：Financial Results
页码范围：25-42

=== 第 25 页 ===
Revenue increased 15% year-over-year to $2.5 billion...
Net income was $450 million, representing an 18% margin...

=== 第 26 页 ===
Operating expenses were $800 million, primarily driven by...

案例 2：法律合同审查​

场景​

审查法律合同，找出关键条款和风险点。

代码示例​

from pageindex import PageIndexClient
import json

client = PageIndexClient()
doc_id = client.index("service_agreement.pdf")

# 获取完整树结构
structure = json.loads(client.get_document_structure(doc_id))

# 打印所有顶级章节
def print_tree(nodes, indent=0):
    for node in nodes:
        print("  " * indent + f"• {node['title']} (页 {node['start_index']}-{node['end_index']})")
        if node.get('nodes'):
            print_tree(node['nodes'], indent + 1)

print("合同结构:")
print_tree(structure['structure'])

# 查找特定条款
def search_clause(structure, keyword):
    """搜索包含关键词的条款"""
    results = []
    for node in structure:
        if keyword.lower() in node.get('title', '').lower():
            results.append({
                'title': node['title'],
                'pages': f"{node['start_index']}-{node['end_index']}",
                'summary': node.get('summary', '')
            })
        if node.get('nodes'):
            results.extend(search_clause(node['nodes'], keyword))
    return results

# 查找责任条款
liability_clauses = search_clause(structure, "liability")
print("\n责任相关条款:")
for clause in liability_clauses:
    print(f"  - {clause['title']} (页 {clause['pages']})")
    print(f"    摘要：{clause['summary'][:100]}...")

输出示例​

合同结构:
• 1. Definitions (页 1-3)
• 2. Services (页 4-8)
  • 2.1 Scope of Services (页 4-5)
  • 2.2 Service Levels (页 6-8)
• 3. Payment Terms (页 9-12)
• 4. Liability and Indemnification (页 13-18)
  • 4.1 Limitation of Liability (页 13-15)
  • 4.2 Indemnification (页 16-18)

责任相关条款:
  - 4. Liability and Indemnification (页 13-18)
    摘要：This section outlines the liability limitations and indemnification obligations...
  - 4.1 Limitation of Liability (页 13-15)
    摘要：In no event shall either party's total liability exceed the total amount paid...

案例 3：技术文档问答​

场景​

为技术文档构建智能问答系统。

代码示例​

from pageindex import PageIndexClient
from agents import Agent, Runner, function_tool
import json

# 初始化
client = PageIndexClient(workspace="./tech_docs")

# 索引多个文档
api_doc_id = client.index("api_reference.pdf")
guide_doc_id = client.index("user_guide.pdf")

# 文档映射
DOCUMENTS = {
    "api": {"id": api_doc_id, "name": "API 参考"},
    "guide": {"id": guide_doc_id, "name": "用户指南"}
}

@function_tool
def search_document(query: str, doc_type: str = "api") -> str:
    """搜索文档内容"""
    doc_id = DOCUMENTS[doc_type]["id"]
    
    # 获取结构
    structure = json.loads(client.get_document_structure(doc_id))
    
    # 简单搜索：查找相关章节
    def find_relevant(node, query, depth=0):
        if depth > 3:
            return None
        title = node.get('title', '').lower()
        summary = node.get('summary', '').lower()
        if query.lower() in title or query.lower() in summary:
            return {
                'title': node['title'],
                'pages': f"{node['start_index']}-{node['end_index']}"
            }
        for child in node.get('nodes', []):
            result = find_relevant(child, query, depth + 1)
            if result:
                return result
        return None
    
    result = find_relevant(structure['structure'], query)
    if result:
        # 获取内容
        content = json.loads(client.get_page_content(doc_id, result['pages']))
        return f"找到相关章节：{result['title']}\n内容:\n" + "\n".join([p['content'] for p in content])
    return "未找到相关内容"

@function_tool
def list_documents() -> str:
    """列出所有可用文档"""
    return "\n".join([f"- {name}: {doc['id'][:8]}..." for doc in DOCUMENTS.values()])

# 创建 Agent
agent = Agent(
    name="技术文档助手",
    instructions="你是一个技术文档助手。帮助用户查找文档内容。",
    tools=[search_document, list_documents],
)

# 运行
question = "如何配置 API 密钥？"
result = Runner.run(agent, question)
print(result.final_output)

输出示例​

Question: 如何配置 API 密钥？

[tool call]: search_document(query="API 密钥", doc_type="api")
找到相关章节：Authentication
内容:
To configure your API key, add it to your environment:
export API_KEY="sk-your-key-here"
Or add it to your config file:
{
  "api_key": "sk-your-key-here"
}

案例 4：学术研究辅助​

场景​

辅助研究人员快速定位论文中的关键信息。

代码示例​

from pageindex import PageIndexClient
import json

client = PageIndexClient()

# 索引多篇论文
papers = {
    "attention": client.index("attention-is-all-you-need.pdf"),
    "transformer": client.index("transformer-xl.pdf"),
    "bert": client.index("bert.pdf")
}

def compare_across_papers(topic, papers):
    """跨论文比较某个主题"""
    results = {}
    for name, doc_id in papers.items():
        structure = json.loads(client.get_document_structure(doc_id))
        
        # 搜索相关章节
        def find_topic(node, topic):
            if topic.lower() in node.get('title', '').lower():
                return node
            for child in node.get('nodes', []):
                result = find_topic(child, topic)
                if result:
                    return result
            return None
        
        section = find_topic(structure['structure'], topic)
        if section:
            content = json.loads(client.get_page_content(doc_id, 
                f"{section['start_index']}-{section['end_index']}"))
            results[name] = {
                'section': section['title'],
                'pages': f"{section['start_index']}-{section['end_index']}",
                'summary': section.get('summary', '')
            }
    return results

# 比较各论文对"Attention"的处理
comparison = compare_across_papers("attention", papers)
for paper, info in comparison.items():
    print(f"\n{paper.upper()}:")
    print(f"  章节：{info['section']}")
    print(f"  页码：{info['pages']}")
    print(f"  摘要：{info['summary']}")

案例 5：批量文档处理​

场景​

批量处理多个文档，构建企业知识库。

代码示例​

from pageindex import PageIndexClient
import os
import json

# 创建工作空间
client = PageIndexClient(workspace="./enterprise_knowledge_base")

# 批量索引文档
def batch_index(directory):
    """批量索引目录下的所有 PDF"""
    results = []
    for filename in os.listdir(directory):
        if filename.endswith('.pdf'):
            filepath = os.path.join(directory, filename)
            print(f"索引：{filename}")
            try:
                doc_id = client.index(filepath)
                results.append({
                    'filename': filename,
                    'doc_id': doc_id,
                    'status': 'success'
                })
            except Exception as e:
                results.append({
                    'filename': filename,
                    'status': 'error',
                    'error': str(e)
                })
    return results

# 执行批量索引
docs_dir = "./documents"
index_results = batch_index(docs_dir)

# 保存索引报告
report = {
    'total': len(index_results),
    'success': sum(1 for r in index_results if r['status'] == 'success'),
    'error': sum(1 for r in index_results if r['status'] == 'error'),
    'details': index_results
}

with open('./indexing_report.json', 'w') as f:
    json.dump(report, f, indent=2, ensure_ascii=False)

print(f"\n索引完成：{report['success']}/{report['total']} 成功")

案例 6：智能检索增强​

场景​

结合传统搜索和 PageIndex，实现智能检索。

代码示例​

from pageindex import PageIndexClient
import json
import re

class SmartRetriever:
    def __init__(self, workspace):
        self.client = PageIndexClient(workspace=workspace)
    
    def search(self, query, doc_id, max_results=3):
        """智能检索：结合关键词匹配和语义理解"""
        structure = json.loads(self.client.get_document_structure(doc_id))
        
        # 1. 关键词匹配
        keywords = self.extract_keywords(query)
        matches = self.find_by_keywords(structure, keywords)
        
        # 2. 排序并获取内容
        matches = matches[:max_results]
        results = []
        for match in matches:
            content = json.loads(self.client.get_page_content(
                doc_id, f"{match['start']}-{match['end']}"))
            results.append({
                'title': match['title'],
                'pages': f"{match['start']}-{match['end']}",
                'content': '\n'.join([c['content'] for c in content]),
                'score': match['score']
            })
        return results
    
    def extract_keywords(self, query):
        """提取关键词"""
        # 简单实现：分词并过滤停用词
        stop_words = {'的', '了', '在', '是', '我', '有', '和', '就'}
        words = re.findall(r'\w+|\S', query)
        return [w for w in words if w not in stop_words and len(w) > 1]
    
    def find_by_keywords(self, structure, keywords, depth=0):
        """递归查找关键词匹配的章节"""
        matches = []
        for node in structure:
            title = node.get('title', '')
            summary = node.get('summary', '')
            text = node.get('text', '')
            
            # 计算匹配分数
            score = 0
            for keyword in keywords:
                if keyword in title:
                    score += 3
                if keyword in summary:
                    score += 2
                if keyword in text:
                    score += 1
            
            if score > 0:
                matches.append({
                    'title': title,
                    'start': node['start_index'],
                    'end': node['end_index'],
                    'score': score
                })
            
            if node.get('nodes'):
                matches.extend(self.find_by_keywords(node['nodes'], keywords, depth + 1))
        
        return sorted(matches, key=lambda x: x['score'], reverse=True)

# 使用
retriever = SmartRetriever(workspace="./docs")
doc_id = retriever.client.index("manual.pdf")

results = retriever.search("如何配置环境变量", doc_id)
for r in results:
    print(f"\n=== {r['title']} (页 {r['pages']}, 分数：{r['score']}) ===")
    print(r['content'][:300])

案例 7：可视化树结构​

场景​

将 PageIndex 树结构可视化，便于浏览文档结构。

代码示例​

import json
from pageindex import PageIndexClient

client = PageIndexClient()
doc_id = client.index("document.pdf")
structure = json.loads(client.get_document_structure(doc_id))

def visualize_tree(nodes, indent=0, max_depth=3):
    """可视化树结构"""
    if indent >= max_depth:
        return
    
    for node in nodes:
        prefix = "  " * indent
        title = node.get('title', 'Untitled')
        pages = f"({node['start_index']}-{node['end_index']})"
        summary = node.get('summary', '')[:50]
        
        print(f"{prefix}├─ {title} {pages}")
        if summary:
            print(f"{prefix}│  {summary}...")
        
        if node.get('nodes'):
            visualize_tree(node['nodes'], indent + 1, max_depth)

print("文档结构:")
visualize_tree(structure['structure'])

输出示例​

文档结构:
├─ Introduction (1-3)
│  This paper presents a novel approach to...
├─ Method (4-10)
│  We propose a new architecture that...
├─ Experiments (11-20)
│  Our experiments show significant improvements...
│  ├─ Setup (11-13)
│  │  We used the following configuration...
│  └─ Results (14-20)
│     Table 1 shows the main results...
└─ Conclusion (21-22)
   We have presented a new method...

完整示例代码​

所有示例代码可在以下位置找到：

示例文件路径
简单客户端examples/simple_client_demo.py
Agentic RAGexamples/agentic_vectorless_rag_demo.py
批量处理examples/batch_indexing.py
智能检索examples/smart_retriever.py

下一步​

相关文档​

更多示例持续更新中...