Merge branch 'main' into dev/chenyin/msi_eval

examples/flows/evaluation/eval-chat-math/flow.dag.yaml

@@ -1,3 +1,4 @@
+$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
 inputs:
   groundtruth:
     type: string
@@ -31,4 +32,5 @@ nodes:
   aggregation: true
   use_variants: false
 node_variants: {}
-$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json
+environment:
+  python_requirements_txt: requirements.txt

scripts/docs/conf.py

@@ -59,6 +59,8 @@
 
 # Options for the linkcheck builder
 linkcheck_ignore = [
+    # openai related sites blocks the IP of the CI server.
+    r"https://openai\.com/",
     r"https://platform\.openai\.com/",
     r"https://help\.openai\.com/",
     # These are used in card links, for example 'xx.html', .md can't be resolved.

src/promptflow-devkit/promptflow/_sdk/_orchestrator/run_submitter.py

@@ -23,7 +23,6 @@
 from promptflow.tracing._operation_context import OperationContext
 from promptflow.tracing._start_trace import is_collection_writeable, start_trace
 
-from .._configuration import Configuration
 from .._load_functions import load_flow
 from ..entities._flows import FlexFlow
 from .utils import SubmitterHelper, variant_overwrite_context
@@ -36,7 +35,7 @@ class RunSubmitter:
 
     def __init__(self, client):
         self._client = client
-        self._config = Configuration(overrides=self._client._config)
+        self._config = self._client._config
         self.run_operations = self._client.runs
 
     def submit(self, run: Run, stream=False, **kwargs):
@@ -96,13 +95,13 @@ def _run_bulk(self, run: Run, stream=False, **kwargs):
             # pass with internal parameter `_collection`
             start_trace(
                 attributes=attributes,
-                run=run.name,
+                run=run,
                 _collection=collection_for_run,
                 path=flow_path,
             )
         else:
             logger.debug("trace collection is protected, will honor existing collection.")
-            start_trace(attributes=attributes, run=run.name, path=flow_path)
+            start_trace(attributes=attributes, run=run, path=flow_path)
 
         self._validate_inputs(run=run)
 
@@ -276,7 +275,7 @@ def _upload_run_to_cloud(cls, run: Run):
             from promptflow._sdk._tracing import _get_ws_triad_from_pf_config
             from promptflow.azure._cli._utils import _get_azure_pf_client
 
-            ws_triad = _get_ws_triad_from_pf_config(path=run._get_flow_dir().resolve())
+            ws_triad = _get_ws_triad_from_pf_config(path=run._get_flow_dir().resolve(), config=run._config)
             pf = _get_azure_pf_client(
                 subscription_id=ws_triad.subscription_id,
                 resource_group=ws_triad.resource_group_name,

src/promptflow-devkit/promptflow/_sdk/_pf_client.py

@@ -44,7 +44,7 @@ def __init__(self, **kwargs):
         # when this is set, telemetry from this client will use this user agent and ignore the one from OperationContext
         self._user_agent_override = kwargs.pop(USER_AGENT_OVERRIDE_KEY, None)
         self._connection_provider = kwargs.pop("connection_provider", None)
-        self._config = kwargs.get("config", None) or {}
+        self._config = Configuration(overrides=kwargs.get("config", None) or {})
         # The credential is used as an option to override
         # DefaultAzureCredential when using workspace connection provider
         self._credential = kwargs.get("credential", None)
@@ -212,7 +212,7 @@ def _run(
                 connections=connections,
                 environment_variables=environment_variables,
                 properties=properties,
-                config=Configuration(overrides=self._config),
+                config=self._config,
                 init=init,
                 dynamic_callable=dynamic_callable,
             )
@@ -372,7 +372,7 @@ def tools(self) -> ToolOperations:
     def _ensure_connection_provider(self) -> str:
         if not self._connection_provider:
             # Get a copy with config override instead of the config instance
-            self._connection_provider = Configuration(overrides=self._config).get_connection_provider()
+            self._connection_provider = self._config.get_connection_provider()
             logger.debug("PFClient connection provider: %s, setting to env.", self._connection_provider)
             from promptflow.core._connection_provider._connection_provider import ConnectionProvider
 

src/promptflow-devkit/promptflow/_sdk/_tracing.py

@@ -56,6 +56,7 @@
     extract_workspace_triad_from_trace_provider,
 )
 from promptflow._sdk._utilities.tracing_utils import get_workspace_kind, parse_kv_from_pb_attribute, parse_protobuf_span
+from promptflow._sdk.entities import Run
 from promptflow._utils.logger_utils import get_cli_sdk_logger
 from promptflow._utils.thread_utils import ThreadWithContextVars
 from promptflow.tracing._integrations._openai_injector import inject_openai_api
@@ -195,14 +196,15 @@ def _invoke_pf_svc() -> str:
     return port
 
 
-def _get_ws_triad_from_pf_config(path: typing.Optional[Path]) -> typing.Optional[AzureMLWorkspaceTriad]:
+def _get_ws_triad_from_pf_config(path: typing.Optional[Path], config=None) -> typing.Optional[AzureMLWorkspaceTriad]:
     from promptflow._sdk._configuration import Configuration
 
-    config = Configuration.get_instance().get_trace_destination(path=path)
-    _logger.info("resolved tracing.trace.destination: %s", config)
-    if not TraceDestinationConfig.need_to_export_to_azure(config):
+    config = config or Configuration.get_instance()
+    trace_destination = config.get_trace_destination(path=path)
+    _logger.info("resolved tracing.trace.destination: %s", trace_destination)
+    if not TraceDestinationConfig.need_to_export_to_azure(trace_destination):
         return None
-    return extract_workspace_triad_from_trace_provider(config)
+    return extract_workspace_triad_from_trace_provider(trace_destination)
 
 
 # priority: run > experiment > collection
@@ -368,6 +370,11 @@ def start_trace_with_devkit(collection: str, **kwargs: typing.Any) -> None:
     _logger.debug("kwargs: %s", kwargs)
     attrs = kwargs.get("attributes", None)
     run = kwargs.get("run", None)
+    if isinstance(run, Run):
+        run_config = run._config
+        run = run.name
+    else:
+        run_config = None
     path = kwargs.get("path", None)
 
     # honor and set attributes if user has specified
@@ -395,7 +402,7 @@ def start_trace_with_devkit(collection: str, **kwargs: typing.Any) -> None:
 
     # local to cloud feature
     _logger.debug("start_trace_with_devkit.path(from kwargs): %s", path)
-    ws_triad = _get_ws_triad_from_pf_config(path=path)
+    ws_triad = _get_ws_triad_from_pf_config(path=path, config=run_config)
     is_azure_ext_installed = _is_azure_ext_installed()
     if ws_triad is not None and not is_azure_ext_installed:
         warning_msg = (
@@ -602,17 +609,11 @@ def process_otlp_trace_request(
                 else:
                     all_spans.append(span)
 
-    if cloud_trace_only:
-        # If we only trace to cloud, we should make sure the data writing is success before return.
-        _try_write_trace_to_cosmosdb(
-            all_spans, get_created_by_info_with_cache, logger, get_credential, is_cloud_trace=True
-        )
-    else:
-        # Create a new thread to write trace to cosmosdb to avoid blocking the main thread
-        ThreadWithContextVars(
-            target=_try_write_trace_to_cosmosdb,
-            args=(all_spans, get_created_by_info_with_cache, logger, get_credential, False),
-        ).start()
+    # Create a new thread to write trace to cosmosdb to avoid blocking the main thread
+    ThreadWithContextVars(
+        target=_try_write_trace_to_cosmosdb,
+        args=(all_spans, get_created_by_info_with_cache, logger, get_credential, cloud_trace_only),
+    ).start()
 
     return
 

src/promptflow-devkit/promptflow/_sdk/entities/_run.py

@@ -172,15 +172,14 @@ def __init__(
         # default run name: flow directory name + timestamp
         self.name = name or self._generate_run_name()
         experiment_name = kwargs.get("experiment_name", None)
+        self._config: Configuration = kwargs.get("config", Configuration.get_instance())
         if self._run_source == RunInfoSources.LOCAL and not self._use_remote_flow:
             self.flow = Path(str(flow)).resolve().absolute()
             flow_dir = self._get_flow_dir()
             # sanitize flow_dir to avoid invalid experiment name
             self._experiment_name = _sanitize_python_variable_name(flow_dir.name)
             self._lineage_id = get_flow_lineage_id(flow_dir=flow_dir)
-            self._output_path = Path(
-                kwargs.get("output_path", self._generate_output_path(config=kwargs.get("config", None)))
-            )
+            self._output_path = Path(kwargs.get("output_path", self._generate_output_path(config=self._config)))
             if is_prompty_flow(self.flow):
                 self._flow_name = Path(self.flow).stem
             else:
@@ -786,8 +785,7 @@ def _validate_for_run_create_operation(self):
         if not self.run and not self.data:
             raise UserErrorException("at least one of data or run must be provided")
 
-    def _generate_output_path(self, config: Optional[Configuration]) -> Path:
-        config = config or Configuration.get_instance()
+    def _generate_output_path(self, config: Configuration) -> Path:
         path = config.get_run_output_path()
         if path is None:
             path = HOME_PROMPT_FLOW_DIR / ".runs"