def _convert_to_v1_run()

in HowTo/gRPC/Linux/OpenAI/LangChain/PyServer/venv/Lib/langchain/callbacks/tracers/langchain_v1.py [0:0]

• 72 lines of code
• 31 McCabe index (conditional complexity)

    def _convert_to_v1_run(self, run: Run) -> Union[LLMRun, ChainRun, ToolRun]:
        session = self.session or self.load_default_session()
        if not isinstance(session, TracerSessionV1):
            raise ValueError(
                "LangChainTracerV1 is not compatible with"
                f" session of type {type(session)}"
            )

        if run.run_type == "llm":
            if "prompts" in run.inputs:
                prompts = run.inputs["prompts"]
            elif "messages" in run.inputs:
                prompts = [get_buffer_string(batch) for batch in run.inputs["messages"]]
            else:
                raise ValueError("No prompts found in LLM run inputs")
            return LLMRun(
                uuid=str(run.id) if run.id else None,
                parent_uuid=str(run.parent_run_id) if run.parent_run_id else None,
                start_time=run.start_time,
                end_time=run.end_time,
                extra=run.extra,
                execution_order=run.execution_order,
                child_execution_order=run.child_execution_order,
                serialized=run.serialized,
                session_id=session.id,
                error=run.error,
                prompts=prompts,
                response=run.outputs if run.outputs else None,
            )
        if run.run_type == "chain":
            child_runs = [self._convert_to_v1_run(run) for run in run.child_runs]
            return ChainRun(
                uuid=str(run.id) if run.id else None,
                parent_uuid=str(run.parent_run_id) if run.parent_run_id else None,
                start_time=run.start_time,
                end_time=run.end_time,
                execution_order=run.execution_order,
                child_execution_order=run.child_execution_order,
                serialized=run.serialized,
                session_id=session.id,
                inputs=run.inputs,
                outputs=run.outputs,
                error=run.error,
                extra=run.extra,
                child_llm_runs=[run for run in child_runs if isinstance(run, LLMRun)],
                child_chain_runs=[
                    run for run in child_runs if isinstance(run, ChainRun)
                ],
                child_tool_runs=[run for run in child_runs if isinstance(run, ToolRun)],
            )
        if run.run_type == "tool":
            child_runs = [self._convert_to_v1_run(run) for run in run.child_runs]
            return ToolRun(
                uuid=str(run.id) if run.id else None,
                parent_uuid=str(run.parent_run_id) if run.parent_run_id else None,
                start_time=run.start_time,
                end_time=run.end_time,
                execution_order=run.execution_order,
                child_execution_order=run.child_execution_order,
                serialized=run.serialized,
                session_id=session.id,
                action=str(run.serialized),
                tool_input=run.inputs.get("input", ""),
                output=None if run.outputs is None else run.outputs.get("output"),
                error=run.error,
                extra=run.extra,
                child_chain_runs=[
                    run for run in child_runs if isinstance(run, ChainRun)
                ],
                child_tool_runs=[run for run in child_runs if isinstance(run, ToolRun)],
                child_llm_runs=[run for run in child_runs if isinstance(run, LLMRun)],
            )
        raise ValueError(f"Unknown run type: {run.run_type}")