Deep Agents 进阶：LangChain 深度智能体 Skills 技能集成指南

## Skills System
...
**Available Skills:**
- **web-research**: Structured approach to conducting thorough web research
  → Read `/path/to/web-research/SKILL.md` for full instructions
...

uv add pdf2image pypdf reportlab pillow

import sys
import json
import tempfile
import os
from pathlib import Path
from typing import List, Dict, Optional, Union
from langchain_core.tools import tool

# Add scripts dir to sys.path to support imports within the scripts
# Assuming this file is at src/components/atomic/pdf/pdf_tools.py
# And scripts are at src/skills/pdf/scripts
SCRIPTS_DIR = Path(__file__).parent.parent.parent.parent / "skills/pdf/scripts"
sys.path.append(str(SCRIPTS_DIR))

# Import necessary modules from scripts
try:
   import extract_form_field_info
   import fill_fillable_fields
   import convert_pdf_to_images
   import check_bounding_boxes
   import fill_pdf_form_with_annotations
   import create_validation_image
   from pypdf import PdfReader, PdfWriter
except ImportError as e:
   # This might happen if dependencies are missing or paths are wrong
   # We will let the tools fail at runtime if imports failed
   print(f"Warning: Failed to import PDF scripts or dependencies in pdf_tools.py: {e}")

import io

@tool
def pdf_convert_to_images(pdf_path: str, output_dir: str, max_dim: int = 1000) -> str:
   """
   Converts pages of a PDF file into PNG images.
   Useful for visual inspection or when text extraction fails.
   
   Args:
       pdf_path: Absolute path to the source PDF file.
       output_dir: Directory where images will be saved. Will be created if it doesn't exist.
       max_dim: Maximum dimension (width or height) for the output images. Default is 1000.
       
   Returns:
       A success message with the number of images created.
   """
   if not os.path.exists(output_dir):
       os.makedirs(output_dir)
       
   try:
       # Capture stdout to count pages or get info if needed, though the function prints
       # We can just run it.
       convert_pdf_to_images.convert(pdf_path, output_dir, max_dim)
       return f"Successfully converted PDF pages to images in {output_dir}"
   except Exception as e:
       return f"Error converting PDF to images: {str(e)}"

@tool
def pdf_check_bounding_boxes(fields: Dict) -> List[str]:
   """
   Checks for overlapping bounding boxes in form field definitions.
   Use this to validate 'fields' data before attempting to fill a form with annotations.
   
   Args:
       fields: A dictionary containing 'form_fields' list and 'pages' list, 
               matching the structure required for form filling.
               
   Returns:
       A list of validation messages (Success or Failure details).
   """
   try:
       # The script expects a stream containing JSON
       fields_json = json.dumps(fields)
       stream = io.StringIO(fields_json)
       
       messages = check_bounding_boxes.get_bounding_box_messages(stream)
       return messages
   except Exception as e:
       return [f"Error checking bounding boxes: {str(e)}"]

@tool
def pdf_fill_with_annotations(input_pdf_path: str, fields: Dict, output_pdf_path: str) -> str:
   """
   Fills a PDF by adding text annotations (FreeText) at specific coordinates.
   This is different from filling standard form fields - it 'draws' text onto the page.
   Use this when the PDF is not a fillable form but you know the coordinates.
   
   Args:
       input_pdf_path: Absolute path to the source PDF.
       fields: A dictionary containing 'form_fields' and 'pages' definitions with coordinates.
       output_pdf_path: Absolute path where the result will be saved.
       
   Returns:
       A success message.
   """
   # Create a temporary JSON file for the fields
   with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
       json.dump(fields, tmp)
       tmp_path = tmp.name
       
   try:
       fill_pdf_form_with_annotations.fill_pdf_form(input_pdf_path, tmp_path, output_pdf_path)
       return f"Successfully filled PDF with annotations and saved to {output_pdf_path}"
   except Exception as e:
       return f"Error filling PDF with annotations: {str(e)}"
   finally:
       if os.path.exists(tmp_path):
           os.remove(tmp_path)

@tool
def pdf_create_validation_image(page_number: int, fields: Dict, input_image_path: str, output_image_path: str) -> str:
   """
   Draws bounding boxes on a page image to validate field coordinates.
   Useful for debugging or verifying 'fields' data before filling annotations.
   
   Args:
       page_number: The page number (1-based) to validate.
       fields: A dictionary containing 'form_fields' with 'label_bounding_box' and 'entry_bounding_box'.
       input_image_path: Path to the image of the PDF page (generated by pdf_convert_to_images).
       output_image_path: Path where the validation image will be saved.
       
   Returns:
       A success message.
   """
   # Create a temporary JSON file for the fields
   with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
       json.dump(fields, tmp)
       tmp_path = tmp.name
       
   try:
       create_validation_image.create_validation_image(page_number, tmp_path, input_image_path, output_image_path)
       return f"Successfully created validation image at {output_image_path}"
   except Exception as e:
       return f"Error creating validation image: {str(e)}"
   finally:
       if os.path.exists(tmp_path):
           os.remove(tmp_path)

@tool
def pdf_extract_form_fields(pdf_path: str) -> Union[List[Dict], str]:
   """
   Extracts information about fillable form fields from a PDF file.
   Use this to inspect a PDF form before filling it.
   
   Args:
       pdf_path: The absolute path to the PDF file.
       
   Returns:
       A list of dictionaries containing field information (field_id, page, type, rect, etc.),
       or an error message string.
   """
   try:
       reader = PdfReader(pdf_path)
       return extract_form_field_info.get_field_info(reader)
   except Exception as e:
       return f"Error extracting fields from {pdf_path}: {str(e)}"

@tool
def pdf_fill_form(input_pdf_path: str, fields: List[Dict], output_pdf_path: str) -> str:
   """
   Fills a PDF form with the provided field values.
   
   Args:
       input_pdf_path: Absolute path to the source PDF template.
       fields: A list of dictionaries, each MUST have 'field_id', 'page', and 'value'.
               Example: [{"field_id": "name", "page": 1, "value": "John Doe"}]
       output_pdf_path: Absolute path where the filled PDF will be saved.
       
   Returns:
       A success message or error description.
   """
   # Create a temporary JSON file for the fields because the script expects a file path
   with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp:
       json.dump(fields, tmp)
       tmp_path = tmp.name
       
   try:
       # Apply monkeypatch if available (fixes pypdf bug for choice fields)
       if hasattr(fill_fillable_fields, 'monkeypatch_pydpf_method'):
           fill_fillable_fields.monkeypatch_pydpf_method()
           
       # The script function doesn't return anything but might print to stdout or exit on error
       # We need to capture stdout/stderr or handle SystemExit if possible, 
       # but calling it directly is safer than subprocess if we trust it not to kill the process hard.
       # fill_pdf_fields calls sys.exit(1) on error, which raises SystemExit.
       
       fill_fillable_fields.fill_pdf_fields(input_pdf_path, tmp_path, output_pdf_path)
       return f"Successfully filled form and saved to {output_pdf_path}"
       
   except SystemExit as e:
       return f"Error: The PDF filling process exited with code {e}. This usually means a validation error occurred (e.g. invalid field ID or value). Please check the field definitions."
   except Exception as e:
       return f"Error filling form: {str(e)}"
   finally:
       if os.path.exists(tmp_path):
           os.remove(tmp_path)

@tool
def pdf_merge(pdf_paths: List[str], output_path: str) -> str:
   """
   Merges multiple PDF files into a single PDF.
   
   Args:
       pdf_paths: A list of absolute paths to the PDF files to merge, in order.
       output_path: The absolute path where the merged PDF will be saved.
       
   Returns:
       A success message.
   """
   try:
       writer = PdfWriter()
       for path in pdf_paths:
           reader = PdfReader(path)
           for page in reader.pages:
               writer.add_page(page)
       
       with open(output_path, "wb") as output:
           writer.write(output)
           
       return f"Successfully merged {len(pdf_paths)} files into {output_path}"
   except Exception as e:
       return f"Error merging PDFs: {str(e)}"

@tool
def pdf_extract_text(pdf_path: str) -> str:
   """
   Extracts text from a PDF file.
   
   Args:
       pdf_path: Absolute path to the PDF file.
       
   Returns:
       The extracted text content, separated by page markers.
   """
   try:
       reader = PdfReader(pdf_path)
       text = ""
       for i, page in enumerate(reader.pages):
           text += f"--- Page {i+1} ---\n"
           text += page.extract_text() + "\n"
       return text
   except Exception as e:
       return f"Error extracting text from {pdf_path}: {str(e)}"

from typing import List, Dict, Optional
import os
import re
from pathlib import Path
from dataclasses import dataclass

@dataclass
class SkillMetadata:
    name: str
    description: str
    path: str

class SkillRegistry:
    def __init__(self, skills_dir: str):
        self.skills_dir = Path(skills_dir).resolve()
        self.skills: List[SkillMetadata] = []
        self._load_skills()
        
    def _load_skills(self):
        if not self.skills_dir.exists():
            # Try finding it relative to project root if absolute path fails
            # Assuming project root is 3 levels up from here (src/middleware/..)
            # But better to just rely on what's passed or fail gracefully
            print(f"Warning: Skills directory {self.skills_dir} does not exist.")
            return

        for skill_dir in self.skills_dir.iterdir():
            if not skill_dir.is_dir():
                continue
                
            skill_md_path = skill_dir / "SKILL.md"
            if not skill_md_path.exists():
                continue
                
            metadata = self._parse_frontmatter(skill_md_path)
            if metadata:
                self.skills.append(metadata)

    def _parse_frontmatter(self, path: Path) -> Optional[SkillMetadata]:
        try:
            content = path.read_text(encoding="utf-8")
            match = re.match(r"^---\s*\n(.*?)\n---\s*\n", content, re.DOTALL)
            if not match:
                return None
                
            frontmatter = match.group(1)
            meta = {}
            for line in frontmatter.split("\n"):
                if ":" in line:
                    key, value = line.split(":", 1)
                    meta[key.strip()] = value.strip()
            
            if "name" in meta and "description" in meta:
                return SkillMetadata(
                    name=meta["name"],
                    description=meta["description"],
                    path=str(path)
                )
            return None
        except Exception:
            return None

    def get_system_prompt_addition(self) -> str:
        """Returns the formatted skills section for the system prompt."""
        if not self.skills:
            return ""
            
        lines = [
            "\n\n## Skills System",
            "You have access to specialized capabilities defined in Skills.",
            "\n**Available Skills:**"
        ]
        
        for skill in self.skills:
            lines.append(f"- **{skill.name}**: {skill.description}")
            lines.append(f"  -> Read `{skill.path}` for full instructions.")
            
        lines.append("\n**Progressive Disclosure:**")
        lines.append("1. Recognize when a task matches a skill's description.")
        lines.append("2. Use the `read_file` tool to read the SKILL.md content to get detailed instructions.")
        lines.append("3. Follow the specific workflows and rules defined in the SKILL.md.")
        
        return "\n".join(lines)

import os
from pathlib import Path
from utils.callbacks import LoggingStreamingCallbackHandler
from deepagents import create_deep_agent, CompiledSubAgent
from components.atomic.pdf.pdf_tools import (
   pdf_extract_form_fields,
   pdf_fill_form,
   pdf_merge,
   pdf_extract_text,
   pdf_convert_to_images,
   pdf_check_bounding_boxes,
   pdf_fill_with_annotations,
   pdf_create_validation_image
)
from langchain_community.agent_toolkits.file_management.toolkit import FileManagementToolkit
from langchain_experimental.tools import PythonREPLTool
from middleware.skill_registry import SkillRegistry
from llm.llm import get_llm_by_type, LLMType

# 1. Configure the model for the subagent
model = get_llm_by_type(LLMType.BASIC, new_instance=True)
model.streaming = True
model.callbacks = [LoggingStreamingCallbackHandler(agent_name="PDF Specialist", color="magenta")]

# 2. Define tools
# Calculate paths
current_file = Path(__file__)
src_dir = current_file.parent.parent # src/
skills_dir = src_dir / "skills"
project_root = src_dir.parent

# Use FileManagementToolkit to get read_file (safe and standard)
file_toolkit = FileManagementToolkit(
   root_dir=str(project_root),
   selected_tools=["read_file"]
)
file_management_tools = file_toolkit.get_tools()

# Python REPL for dynamic code execution
# ⚠️ SECURITY WARNING: 
# PythonREPLTool executes code locally. This is unsafe for production.
# For production usage, MUST replace this with a sandboxed environment like E2B or a Dockerized container.
# See: https://python.langchain.com/docs/integrations/tools/e2b_data_analysis
python_repl_tool = PythonREPLTool()

# Combine tools
pdf_tools = [
   pdf_extract_form_fields,
   pdf_fill_form,
   pdf_merge,
   pdf_extract_text,
   pdf_convert_to_images,
   pdf_check_bounding_boxes,
   pdf_fill_with_annotations,
   pdf_create_validation_image,
   python_repl_tool,
   *file_management_tools
]

# 3. Setup Skill Registry and System Prompt

# Initialize registry
registry = SkillRegistry(str(skills_dir))
skills_prompt_section = registry.get_system_prompt_addition()

# Define base prompt
base_system_prompt = """
You are a PDF Specialist Agent. Your primary responsibility is to handle PDF documents using the provided tools.

You have access to:
1.  **PDF Tools**: 
   - `pdf_extract_form_fields`: Extract info about fillable fields.
   - `pdf_fill_form`: Fill standard PDF form fields.
   - `pdf_merge`: Merge multiple PDFs.
   - `pdf_extract_text`: Extract text from PDF.
   - `pdf_convert_to_images`: Convert PDF pages to images (useful for vision).
   - `pdf_check_bounding_boxes`: Validate field coordinates.
   - `pdf_fill_with_annotations`: "Draw" text on PDF (for non-fillable forms).
   - `pdf_create_validation_image`: Draw validation boxes on page image.
2.  **File Tools**: `read_file` (essential for learning new skills).
3.  **Python Execution**: `python_repl` (Use this to run Python code for custom PDF creation, e.g. using `reportlab` or `pypdf` as described in SKILL.md).

**Core Philosophy:**
You are powered by a **Skills System**. You do not have all instructions pre-loaded. Instead:
1.  Check the **Available Skills** list below.
2.  If a task matches a skill, **READ the SKILL.md file** using `read_file`.
3.  Follow the instructions in that file exactly.
"""

# Combine prompts
final_system_prompt = f"{base_system_prompt}\n{skills_prompt_section}"

# 4. Create the independent agent graph
pdf_graph = create_deep_agent(
   model=model,
   tools=pdf_tools,
   system_prompt=final_system_prompt,
)

# 5. Wrap it as a CompiledSubAgent
pdf_subagent = CompiledSubAgent(
   name="pdf_specialist",
   description="A specialist agent for PDF operations. It uses a Skills System to dynamically load detailed instructions for tasks like form filling, merging, and extraction.",
   runnable=pdf_graph
)

import os
from dotenv import load_dotenv
from deepagents import create_deep_agent
from llm.llm import get_llm_by_type, LLMType
# 导入各个子代理模块
from subagent.subagent_search import search_subagent
from subagent.subagent_rag import rag_subagent
from subagent.subagent_pdf import pdf_subagent # 导入我们新创建的 PDF 专家

# 加载环境变量
load_dotenv()

def main():
    model = get_llm_by_type(LLMType.BASIC)

    # 创建主代理
    agent = create_deep_agent(
        model=model,
        # 将 pdf_subagent 加入到子代理列表中
        subagents=[search_subagent, rag_subagent, pdf_subagent],
        system_prompt=(
            # 在 System Prompt 中明确指导主代理如何进行任务分发
            "你是一个乐于助人的助手。请将研究任务委派给搜索专家（search_specialist）或 RAG 专家（rag_specialist）。"
            "遇到 PDF 处理任务（提取文本、填表、合并等），请委派给 PDF 专家（pdf_specialist）。"
            "请始终使用中文回答。"
            "\n\n对于复杂的任务，请使用 `write_todos` 工具来制定计划和跟踪进度。"
        )
    )
    
    # ... (后续为 Agent 运行和流式输出处理代码)

使用 PDF 工具处理如下内容
PDF 文件地址：
'/Users/XXX/Paper/Context Engineering/ContextEngineering2.0.pdf' 
1. 读取 PDF 的第一页内容，提取第一页标题和作者信息。
2. 创建一个新的 PDF，采用红色字体输出刚才读取的标题和作者信息，并添加水印 'ApFramework' 。
3. 保存文件到 '/Users/XXX/Paper/Context Engineering/' 文件夹，文件名为 'ContextEngineering2.0-WithWatermark.pdf'  

User: 使用 PDF 工具处理如下内容PDF 文件地址：'/Users/XXX/Paper/Context Engineering/ContextEngineering2.0.pdf' 1. 读取 PDF 的第一页内容，提取第一页标题和作者信息。2. 创建一个新的 PDF，采用红色字体输出刚才读取的标题和作者信息，并添加水印 'ApFramework' 。3. 保存文件到 '/Users/XXX/Paper/Context Engineering/' 文件夹，文件名为 'ContextEngineering2.0-WithWatermark.pdf'    
Agent response:

[Main Agent]
我将帮您处理这个PDF文件任务。这是一个涉及PDF读取、内容提取、创建新PDF并添加水印的多步骤任务。让我使用PDF专家来处理这个任务。

[PDF Specialist]
我将帮您处理这个PDF任务。让我按步骤来完成：

1. 首先读取PDF文件的第一页内容
2. 提取标题和作者信息
3. 创建新PDF并应用红色字体
4. 添加水印
5. 保存结果

让我开始第一步，提取PDF的文本内容：

[PDF Specialist]
很好！我可以看到第一页的内容。从提取的文本中，我可以识别出：

- **标题**: "Context Engineering 2.0: The Context of Context Engineering"
- **作者**: "Qishuo Hua1,3 Lyumanshan Ye1,3 Dayuan Fu2,3 Yang Xiao2,3 Xiaojie Cai1,3 Yunze Wu1,2,3 Jifan Lin1,3 Junfei Wang3 Pengfei Liu1,2,3,†"

现在我需要创建一个新的PDF文件，使用红色字体显示这些信息，并添加水印。我将使用Python的reportlab库来创建PDF：

Python REPL can execute arbitrary code. Use with caution.

[PDF Specialist]
完美！我已经成功完成了所有要求的任务：

✅ **任务完成情况**：

1. **读取PDF第一页内容** - 已成功提取文本内容
2. **提取标题和作者信息** - 
   - 标题: "Context Engineering 2.0: The Context of Context Engineering"
   - 作者: "Qishuo Hua¹,³ Lyumanshan Ye¹,³ Dayuan Fu²,³ Yang Xiao²,³ Xiaojie Cai¹,³ Yunze Wu¹,²,³ Jifan Lin¹,³ Junfei Wang³ Pengfei Liu¹,²,³,†"
3. **创建新PDF使用红色字体** - 标题和作者信息都使用红色字体显示
4. **添加水印 'ApFramework'** - 已添加半透明灰色水印，旋转45度
5. **保存结果** - 文件已保存到 `/Users/XXX/Paper/Context Engineering/ContextEngineering2.0-WithWatermark.pdf`

新创建的PDF具有以下特点：
- 标题使用红色字体，18号字，居中对齐
- 作者信息使用红色字体，12号字，居中对齐，并保留了原有的上标格式
- 页面上添加了半透明的"ApFramework"水印，以45度角旋转显示
- 整体布局简洁专业

文件已经成功生成并保存到指定位置。
完美！我已经成功完成了您要求的所有PDF处理任务。

## 任务完成情况：

✅ **1. 读取PDF第一页内容** - 已成功提取文本内容

✅ **2. 提取标题和作者信息** - 
- **标题**: "Context Engineering 2.0: The Context of Context Engineering"
- **作者**: "Qishuo Hua¹,³ Lyumanshan Ye¹,³ Dayuan Fu²,³ Yang Xiao²,³ Xiaojie Cai¹,³ Yunze Wu¹,²,³ Jifan Lin¹,³ Junfei Wang³ Pengfei Liu¹,²,³,†"

✅ **3. 创建新PDF使用红色字体** - 标题和作者信息都使用红色字体显示

✅ **4. 添加水印 'ApFramework'** - 已添加半透明灰色水印，旋转45度

✅ **5. 保存结果** - 文件已保存到 `/Users/XXX/Paper/Context Engineering/ContextEngineering2.0-WithWatermark.pdf`

## 新创建的PDF特点：
- **标题**: 红色字体，18号字，居中对齐
- **作者信息**: 红色字体，12号字，居中对齐，并保留了原有的上标格式
- **水印**: 半透明的"ApFramework"水印，以45度角旋转显示
- **整体布局**: 简洁专业

文件已经成功生成并保存到您指定的位置。您可以现在查看 `/Users/XXX/Paper/Context Engineering/ContextEngineering2.0-WithWatermark.pdf` 文件来确认效果。