_collect_output() — langchain Function Reference

Architecture documentation for the _collect_output() function in shell_tool.py from the langchain codebase.

Function python LangChainCore Runnables calls 3 called by 1

Entity Profile

LangChainCore→ Runnables→ _collect_output() — langchain Function Reference

Dependency Diagram

graph TD
  89d586a2_0b74_ac5c_97da_a6a18a02f59d["_collect_output()"]
  cf4eeaa4_4bcc_69d9_a0f3_664b1eecb4e8["ShellSession"]
  89d586a2_0b74_ac5c_97da_a6a18a02f59d -->|defined in| cf4eeaa4_4bcc_69d9_a0f3_664b1eecb4e8
  da18dc22_e4fc_2b4a_2829_9217c9f5d8f5["execute()"]
  da18dc22_e4fc_2b4a_2829_9217c9f5d8f5 -->|calls| 89d586a2_0b74_ac5c_97da_a6a18a02f59d
  6744f48e_023a_08e7_32fe_b5a699793a47["_safe_int()"]
  89d586a2_0b74_ac5c_97da_a6a18a02f59d -->|calls| 6744f48e_023a_08e7_32fe_b5a699793a47
  2c92ef14_d880_f149_5132_20d7d028b5b7["_drain_remaining_stderr()"]
  89d586a2_0b74_ac5c_97da_a6a18a02f59d -->|calls| 2c92ef14_d880_f149_5132_20d7d028b5b7
  376fc508_1bee_23a7_7de6_267e13821247["restart()"]
  89d586a2_0b74_ac5c_97da_a6a18a02f59d -->|calls| 376fc508_1bee_23a7_7de6_267e13821247
  style 89d586a2_0b74_ac5c_97da_a6a18a02f59d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/shell_tool.py lines 240–325

    def _collect_output(
        self,
        marker: str,
        deadline: float,
        timeout: float,
    ) -> CommandExecutionResult:
        collected: list[str] = []
        total_lines = 0
        total_bytes = 0
        truncated_by_lines = False
        truncated_by_bytes = False
        exit_code: int | None = None
        timed_out = False

        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                timed_out = True
                break
            try:
                source, data = self._queue.get(timeout=remaining)
            except queue.Empty:
                timed_out = True
                break

            if data is None:
                continue

            if source == "stdout" and data.startswith(marker):
                _, _, status = data.partition(" ")
                exit_code = self._safe_int(status.strip())
                # Drain any remaining stderr that may have arrived concurrently.
                # The stderr reader thread runs independently, so output might
                # still be in flight when the stdout marker arrives.
                self._drain_remaining_stderr(collected, deadline)
                break

            total_lines += 1
            encoded = data.encode("utf-8", "replace")
            total_bytes += len(encoded)

            if total_lines > self._policy.max_output_lines:
                truncated_by_lines = True
                continue

            if (
                self._policy.max_output_bytes is not None
                and total_bytes > self._policy.max_output_bytes
            ):
                truncated_by_bytes = True
                continue

            if source == "stderr":
                stripped = data.rstrip("\n")
                collected.append(f"[stderr] {stripped}")
                if data.endswith("\n"):
                    collected.append("\n")
            else:
                collected.append(data)

        if timed_out:
            LOGGER.warning(
                "Command timed out after %.2f seconds; restarting shell session.",
                timeout,
            )
            self.restart()
            return CommandExecutionResult(
                output="",
                exit_code=None,
                timed_out=True,
                truncated_by_lines=truncated_by_lines,
                truncated_by_bytes=truncated_by_bytes,
                total_lines=total_lines,
                total_bytes=total_bytes,
            )

        output = "".join(collected)
        return CommandExecutionResult(
            output=output,
            exit_code=exit_code,
            timed_out=False,