Skip to content

DocumentToolkit

Document toolkit for parsing documents and support Q&A.

Support backends:

DocumentToolkit

Bases: AsyncBaseToolkit

Source code in utu/tools/document_toolkit.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class DocumentToolkit(AsyncBaseToolkit):
    def __init__(self, config: ToolkitConfig = None) -> None:
        """Initialize the DocumentToolkit, with configed parser and llm."""
        super().__init__(config)
        if self.config.config.get("parser") == "chunkr":
            from .documents.chunkr_parser import ChunkrParser

            self.parser = ChunkrParser(self.config.config)
        elif self.config.config.get("parser") == "pymupdf":
            from .documents.pdf_parser import PDFParser

            self.parser = PDFParser(self.config.config)
        else:
            raise ValueError(f"Unsupported parser: {self.config.config.get('parser')}")
        self.text_limit = self.config.config.get("text_limit", 100_000)
        self.llm = SimplifiedAsyncOpenAI(**self.config.config_llm.model_provider.model_dump())
        self.md5_to_path = {}

    async def parse_document(self, md5: str) -> str:
        logger.info(f"[tool] parse_document: {self.md5_to_path[md5]}")
        return await self.parser.parse(self.md5_to_path[md5])

    def handle_path(self, path_or_url: str) -> str:
        md5 = FileUtils.get_file_md5(path_or_url)
        logger.info(f"md5 for {path_or_url}: {md5}")
        if FileUtils.is_web_url(path_or_url):
            # download document to data/_document, with md5
            fn = CACHE_DIR / "documents" / f"{md5}{FileUtils.get_file_ext(path_or_url)}"
            fn.parent.mkdir(parents=True, exist_ok=True)
            if not fn.exists():
                logger.info(f"Downloaded document file to {path_or_url}")
                FileUtils.download_file(url=path_or_url, save_path=fn)
            self.md5_to_path[md5] = fn  # record md5 to map
        else:
            self.md5_to_path[md5] = path_or_url
        return md5

    @register_tool
    async def document_qa(self, document_path: str, question: str | None = None) -> str:
        """Get file content summary or answer questions about attached document.

        Supported file types: pdf, docx, pptx, xlsx, xls, ppt, doc

        Args:
            document_path (str): Local path or URL to a document.
            question (str, optional): The question to answer. If not provided, return a summary of the document.
        """
        md5 = self.handle_path(document_path)
        document_markdown = await self.parse_document(md5)
        if len(document_markdown) > self.text_limit:
            document_markdown = document_markdown[: self.text_limit] + "\n..."
        messages = [
            {"role": "system", "content": TOOL_PROMPTS["document_sp"]},
            {"role": "user", "content": document_markdown},
        ]
        if question:
            messages.append({"role": "user", "content": TOOL_PROMPTS["document_qa"].format(question=question)})
        else:
            messages.append({"role": "user", "content": TOOL_PROMPTS["document_summary"]})
        output = await self.llm.query_one(messages=messages, **self.config.config_llm.model_params.model_dump())
        if not question:
            output = (
                f"You did not provide a particular question, so here is a detailed caption for the document: {output}"
            )
        return output

tools_map property 

tools_map: dict[str, Callable]

Lazy loading of tools map. - collect tools registered by @register_tool

__init__ 

__init__(config: ToolkitConfig = None) -> None

Initialize the DocumentToolkit, with configed parser and llm.

Source code in utu/tools/document_toolkit.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def __init__(self, config: ToolkitConfig = None) -> None:
    """Initialize the DocumentToolkit, with configed parser and llm."""
    super().__init__(config)
    if self.config.config.get("parser") == "chunkr":
        from .documents.chunkr_parser import ChunkrParser

        self.parser = ChunkrParser(self.config.config)
    elif self.config.config.get("parser") == "pymupdf":
        from .documents.pdf_parser import PDFParser

        self.parser = PDFParser(self.config.config)
    else:
        raise ValueError(f"Unsupported parser: {self.config.config.get('parser')}")
    self.text_limit = self.config.config.get("text_limit", 100_000)
    self.llm = SimplifiedAsyncOpenAI(**self.config.config_llm.model_provider.model_dump())
    self.md5_to_path = {}

document_qa async 

document_qa(
    document_path: str, question: str | None = None
) -> str

Get file content summary or answer questions about attached document.

Supported file types: pdf, docx, pptx, xlsx, xls, ppt, doc

Parameters:

Name Type Description Default
document_path str

Local path or URL to a document.

 required
question str

The question to answer. If not provided, return a summary of the document.

 None
Source code in utu/tools/document_toolkit.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@register_tool
async def document_qa(self, document_path: str, question: str | None = None) -> str:
    """Get file content summary or answer questions about attached document.

    Supported file types: pdf, docx, pptx, xlsx, xls, ppt, doc

    Args:
        document_path (str): Local path or URL to a document.
        question (str, optional): The question to answer. If not provided, return a summary of the document.
    """
    md5 = self.handle_path(document_path)
    document_markdown = await self.parse_document(md5)
    if len(document_markdown) > self.text_limit:
        document_markdown = document_markdown[: self.text_limit] + "\n..."
    messages = [
        {"role": "system", "content": TOOL_PROMPTS["document_sp"]},
        {"role": "user", "content": document_markdown},
    ]
    if question:
        messages.append({"role": "user", "content": TOOL_PROMPTS["document_qa"].format(question=question)})
    else:
        messages.append({"role": "user", "content": TOOL_PROMPTS["document_summary"]})
    output = await self.llm.query_one(messages=messages, **self.config.config_llm.model_params.model_dump())
    if not question:
        output = (
            f"You did not provide a particular question, so here is a detailed caption for the document: {output}"
        )
    return output

get_tools_map_func 

get_tools_map_func() -> dict[str, Callable]

Get tools map. It will filter tools by config.activated_tools if it is not None.

Source code in utu/tools/base.py 
36
37
38
39
40
41
42
43
44
45
def get_tools_map_func(self) -> dict[str, Callable]:
    """Get tools map. It will filter tools by config.activated_tools if it is not None."""
    if self.config.activated_tools:
        assert all(tool_name in self.tools_map for tool_name in self.config.activated_tools), (
            f"Error config activated tools: {self.config.activated_tools}! available tools: {self.tools_map.keys()}"
        )
        tools_map = {tool_name: self.tools_map[tool_name] for tool_name in self.config.activated_tools}
    else:
        tools_map = self.tools_map
    return tools_map

get_tools_in_agents 

get_tools_in_agents() -> list[FunctionTool]

Get tools in openai-agents format.

Source code in utu/tools/base.py 
47
48
49
50
51
52
53
54
55
56
57
58
def get_tools_in_agents(self) -> list[FunctionTool]:
    """Get tools in openai-agents format."""
    tools_map = self.get_tools_map_func()
    tools = []
    for _, tool in tools_map.items():
        tools.append(
            function_tool(
                tool,
                strict_mode=False,  # turn off strict mode
            )
        )
    return tools

get_tools_in_openai 

get_tools_in_openai() -> list[dict]

Get tools in OpenAI format.

Source code in utu/tools/base.py 
60
61
62
63
def get_tools_in_openai(self) -> list[dict]:
    """Get tools in OpenAI format."""
    tools = self.get_tools_in_agents()
    return [ChatCompletionConverter.tool_to_openai(tool) for tool in tools]

get_tools_in_mcp 

get_tools_in_mcp() -> list[Tool]

Get tools in MCP format.

Source code in utu/tools/base.py 
65
66
67
68
def get_tools_in_mcp(self) -> list[types.Tool]:
    """Get tools in MCP format."""
    tools = self.get_tools_in_agents()
    return [MCPConverter.function_tool_to_mcp(tool) for tool in tools]

call_tool async 

call_tool(name: str, arguments: dict) -> str

Call a tool by its name.

Source code in utu/tools/base.py 
70
71
72
73
74
75
76
async def call_tool(self, name: str, arguments: dict) -> str:
    """Call a tool by its name."""
    tools_map = self.get_tools_map_func()
    if name not in tools_map:
        raise ValueError(f"Tool {name} not found")
    tool = tools_map[name]
    return await tool(**arguments)