del

examples/advanced_agents/llm_compiler/output_parser.py

@@ -1,154 +0,0 @@
-import re
-from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, Union
-import ast
-
-from langchain_core.exceptions import OutputParserException
-from langchain_core.messages import BaseMessage
-from langchain_core.output_parsers import BaseOutputParser
-from langchain_core.tools import BaseTool
-
-THOUGHT_PATTERN = r"Thought: ([^\n]*)"
-ACTION_PATTERN = r"\n*(\d+)\. (\w+)\((.*)\)(\s*#\w+\n)?"
-# $1 or ${1} -> 1
-ID_PATTERN = r"\$\{?(\d+)\}?"
-END_OF_PLAN = "<END_OF_PLAN>"
-
-
-### Helper functions
-
-
-def _ast_parse(arg: str) -> Any:
-    try:
-        return ast.literal_eval(arg)
-    except:  # noqa
-        return arg
-
-
-def _parse_llm_compiler_action_args(args: str, tool: Union[str, BaseTool]) -> list[Any]:
-    """Parse arguments from a string."""
-    if args == "":
-        return ()
-    if isinstance(tool, str):
-        return ()
-    extracted_args = {}
-    tool_key = None
-    prev_idx = None
-    for key in tool.args.keys():
-        # Split if present
-        if f"{key}=" in args:
-            idx = args.index(f"{key}=")
-            if prev_idx is not None:
-                # print(tool_key, flush=True)
-                # breakpoint()
-                extracted_args[tool_key] = _ast_parse(
-                    args[prev_idx:idx].strip().rstrip(",")
-                )
-            args = args.split(f"{key}=", 1)[1]
-            tool_key = key
-            prev_idx = 0
-    if prev_idx is not None:
-        extracted_args[tool_key] = _ast_parse(
-            args[prev_idx:].strip().rstrip(",").rstrip(")")
-        )
-    return extracted_args
-
-
-def default_dependency_rule(idx, args: str):
-    matches = re.findall(ID_PATTERN, args)
-    numbers = [int(match) for match in matches]
-    return idx in numbers
-
-
-def _get_dependencies_from_graph(
-    idx: int, tool_name: str, args: Dict[str, Any]
-) -> dict[str, list[str]]:
-    """Get dependencies from a graph."""
-    if tool_name == "join":
-        return list(range(1, idx))
-    return [i for i in range(1, idx) if default_dependency_rule(i, str(args))]
-
-
-def instantiate_task(
-    tools: Sequence[BaseTool],
-    idx: int,
-    tool_name: str,
-    args: Union[str, Any],
-    thought: Optional[str] = None,
-) -> dict:
-    if tool_name == "join":
-        tool = "join"
-    else:
-        try:
-            tool = tools[[tool.name for tool in tools].index(tool_name)]
-        except ValueError as e:
-            raise OutputParserException(f"Tool {tool_name} not found.") from e
-    tool_args = _parse_llm_compiler_action_args(args, tool)
-    dependencies = _get_dependencies_from_graph(idx, tool_name, tool_args)
-
-    return dict(
-        idx=idx,
-        tool=tool,
-        args=tool_args,
-        dependencies=dependencies,
-        thought=thought,
-    )
-
-
-class LLMCompilerPlanParser(BaseOutputParser[dict], extra="allow"):
-    """Planning output parser."""
-
-    tools: List[BaseTool]
-
-    def _transform(self, input: Iterator[Union[str, BaseMessage]]) -> Iterator[dict]:
-        texts = []
-        thought = None
-        for chunk in input:
-            # Assume input is str. TODO: support vision/other formats
-            text = chunk if isinstance(chunk, str) else str(chunk.content)
-            for task, thought in self.ingest_token(text, texts, thought):
-                yield {task["idx"]: task}
-        # Final possible task
-        if texts:
-            task, _ = self._parse_task("".join(texts), thought)
-            if task:
-                yield {task["idx"]: task}
-
-    def parse(self, text: str):
-        task_dict = {}
-        for task in self._transform([text]):
-            task_dict.update(task)
-        return task_dict
-
-    def ingest_token(
-        self, token: str, buffer: List[str], thought: Optional[str]
-    ) -> Iterator[Tuple[Optional[dict], str]]:
-        buffer.append(token)
-        if "\n" in token:
-            buffer_ = "".join(buffer).split("\n")
-            suffix = buffer_[-1]
-            for line in buffer_[:-1]:
-                task, thought = self._parse_task(line, thought)
-                if task:
-                    yield task, thought
-            buffer.clear()
-            buffer.append(suffix)
-
-    def _parse_task(self, line: str, thought: Optional[str] = None):
-        task = None
-        if match := re.match(THOUGHT_PATTERN, line):
-            # Optionally, action can be preceded by a thought
-            thought = match.group(1)
-        elif match := re.match(ACTION_PATTERN, line):
-            # if action is parsed, return the task, and clear the buffer
-            idx, tool_name, args, _ = match.groups()
-            idx = int(idx)
-            task = instantiate_task(
-                tools=self.tools,
-                idx=idx,
-                tool_name=tool_name,
-                args=args,
-                thought=thought,
-            )
-            thought = None
-        # Else it is just dropped
-        return task, thought