feat(client): add native environments (#1122)

Author: hassiebp

langfuse/api/reference.md

@@ -1120,7 +1120,15 @@ client.health.health()
 <dl>
 <dd>
 
-Batched ingestion for Langfuse Tracing. If you want to use tracing via the API, such as to build your own Langfuse client implementation, this is the only API route you need to implement.
+Batched ingestion for Langfuse Tracing.
+If you want to use tracing via the API, such as to build your own Langfuse client implementation, this is the only API route you need to implement.
+
+Within each batch, there can be multiple events.
+Each event has a type, an id, a timestamp, metadata and a body.
+Internally, we refer to this as the "event envelope" as it tells us something about the event but not the trace.
+We use the event id within this envelope to deduplicate messages to avoid processing the same event twice, i.e. the event id should be unique per request.
+The event.body.id is the ID of the actual trace and will be used for updates and will be visible within the Langfuse App.
+I.e. if you want to update a trace, you'd use the same body id, but separate event IDs.
 
 Notes:
 
@@ -1141,9 +1149,7 @@ Notes:
 <dd>
 
 ```python
-import datetime
-
-from langfuse import IngestionEvent_TraceCreate, TraceBody
+from langfuse import IngestionEvent_ScoreCreate, ScoreBody
 from langfuse.client import FernLangfuse
 
 client = FernLangfuse(
@@ -1156,29 +1162,17 @@ client = FernLangfuse(
 )
 client.ingestion.batch(
     batch=[
-        IngestionEvent_TraceCreate(
-            body=TraceBody(
-                id="string",
-                timestamp=datetime.datetime.fromisoformat(
-                    "2024-01-15 09:30:00+00:00",
-                ),
-                name="string",
-                user_id="string",
-                input={"key": "value"},
-                output={"key": "value"},
-                session_id="string",
-                release="string",
-                version="string",
-                metadata={"key": "value"},
-                tags=["string"],
-                public=True,
+        IngestionEvent_ScoreCreate(
+            id="abcdef-1234-5678-90ab",
+            timestamp="2022-01-01T00:00:00.000Z",
+            body=ScoreBody(
+                id="abcdef-1234-5678-90ab",
+                trace_id="1234-5678-90ab-cdef",
+                name="My Score",
+                value=0.9,
             ),
-            id="string",
-            timestamp="string",
-            metadata={"key": "value"},
         )
     ],
-    metadata={"key": "value"},
 )
 
 ```

langfuse/api/resources/commons/types/base_score.py

@@ -33,6 +33,11 @@ class BaseScore(pydantic_v1.BaseModel):
     Reference an annotation queue on a score. Populated if the score was initially created in an annotation queue.
     """
 
+    environment: typing.Optional[str] = pydantic_v1.Field(default=None)
+    """
+    The environment from which this score originated. Can be any lowercase alphanumeric string with hyphens and underscores that does not start with 'langfuse'.
+    """
+
     def json(self, **kwargs: typing.Any) -> str:
         kwargs_with_defaults: typing.Any = {
             "by_alias": True,

langfuse/api/resources/commons/types/observation.py

@@ -125,6 +125,11 @@ class Observation(pydantic_v1.BaseModel):
     The cost details of the observation. Key is the name of the cost metric, value is the cost in USD. The total key is the sum of all (non-total) cost metrics or the total value ingested.
     """
 
+    environment: typing.Optional[str] = pydantic_v1.Field(default=None)
+    """
+    The environment from which this observation originated. Can be any lowercase alphanumeric string with hyphens and underscores that does not start with 'langfuse'.
+    """
+
     def json(self, **kwargs: typing.Any) -> str:
         kwargs_with_defaults: typing.Any = {
             "by_alias": True,

langfuse/api/resources/commons/types/score.py

@@ -28,6 +28,7 @@ class Score_Numeric(pydantic_v1.BaseModel):
     comment: typing.Optional[str] = None
     config_id: typing.Optional[str] = pydantic_v1.Field(alias="configId", default=None)
     queue_id: typing.Optional[str] = pydantic_v1.Field(alias="queueId", default=None)
+    environment: typing.Optional[str] = None
     data_type: typing.Literal["NUMERIC"] = pydantic_v1.Field(
         alias="dataType", default="NUMERIC"
     )
@@ -85,6 +86,7 @@ class Score_Categorical(pydantic_v1.BaseModel):
     comment: typing.Optional[str] = None
     config_id: typing.Optional[str] = pydantic_v1.Field(alias="configId", default=None)
     queue_id: typing.Optional[str] = pydantic_v1.Field(alias="queueId", default=None)
+    environment: typing.Optional[str] = None
     data_type: typing.Literal["CATEGORICAL"] = pydantic_v1.Field(
         alias="dataType", default="CATEGORICAL"
     )
@@ -142,6 +144,7 @@ class Score_Boolean(pydantic_v1.BaseModel):
     comment: typing.Optional[str] = None
     config_id: typing.Optional[str] = pydantic_v1.Field(alias="configId", default=None)
     queue_id: typing.Optional[str] = pydantic_v1.Field(alias="queueId", default=None)
+    environment: typing.Optional[str] = None
     data_type: typing.Literal["BOOLEAN"] = pydantic_v1.Field(
         alias="dataType", default="BOOLEAN"
     )

langfuse/api/resources/commons/types/trace.py

@@ -70,6 +70,11 @@ class Trace(pydantic_v1.BaseModel):
     Public traces are accessible via url without login
     """
 
+    environment: typing.Optional[str] = pydantic_v1.Field(default=None)
+    """
+    The environment from which this trace originated. Can be any lowercase alphanumeric string with hyphens and underscores that does not start with 'langfuse'.
+    """
+
     def json(self, **kwargs: typing.Any) -> str:
         kwargs_with_defaults: typing.Any = {
             "by_alias": True,

langfuse/api/resources/ingestion/client.py

@@ -31,7 +31,15 @@ def batch(
         request_options: typing.Optional[RequestOptions] = None,
     ) -> IngestionResponse:
         """
-        Batched ingestion for Langfuse Tracing. If you want to use tracing via the API, such as to build your own Langfuse client implementation, this is the only API route you need to implement.
+        Batched ingestion for Langfuse Tracing.
+        If you want to use tracing via the API, such as to build your own Langfuse client implementation, this is the only API route you need to implement.
+
+        Within each batch, there can be multiple events.
+        Each event has a type, an id, a timestamp, metadata and a body.
+        Internally, we refer to this as the "event envelope" as it tells us something about the event but not the trace.
+        We use the event id within this envelope to deduplicate messages to avoid processing the same event twice, i.e. the event id should be unique per request.
+        The event.body.id is the ID of the actual trace and will be used for updates and will be visible within the Langfuse App.
+        I.e. if you want to update a trace, you'd use the same body id, but separate event IDs.
 
         Notes:
 
@@ -72,28 +80,26 @@ def batch(
         client.ingestion.batch(
             batch=[
                 IngestionEvent_TraceCreate(
+                    id="abcdef-1234-5678-90ab",
+                    timestamp="2022-01-01T00:00:00.000Z",
                     body=TraceBody(
-                        id="string",
+                        id="abcdef-1234-5678-90ab",
                         timestamp=datetime.datetime.fromisoformat(
-                            "2024-01-15 09:30:00+00:00",
+                            "2022-01-01 00:00:00+00:00",
                         ),
-                        name="string",
-                        user_id="string",
-                        input={"key": "value"},
-                        output={"key": "value"},
-                        session_id="string",
-                        release="string",
-                        version="string",
-                        metadata={"key": "value"},
-                        tags=["string"],
+                        name="My Trace",
+                        user_id="1234-5678-90ab-cdef",
+                        input="My input",
+                        output="My output",
+                        session_id="1234-5678-90ab-cdef",
+                        release="1.0.0",
+                        version="1.0.0",
+                        metadata="My metadata",
+                        tags=["tag1", "tag2"],
                         public=True,
                     ),
-                    id="string",
-                    timestamp="string",
-                    metadata={"key": "value"},
                 )
             ],
-            metadata={"key": "value"},
         )
         """
         _response = self._client_wrapper.httpx_client.request(
@@ -142,7 +148,15 @@ async def batch(
         request_options: typing.Optional[RequestOptions] = None,
     ) -> IngestionResponse:
         """
-        Batched ingestion for Langfuse Tracing. If you want to use tracing via the API, such as to build your own Langfuse client implementation, this is the only API route you need to implement.
+        Batched ingestion for Langfuse Tracing.
+        If you want to use tracing via the API, such as to build your own Langfuse client implementation, this is the only API route you need to implement.
+
+        Within each batch, there can be multiple events.
+        Each event has a type, an id, a timestamp, metadata and a body.
+        Internally, we refer to this as the "event envelope" as it tells us something about the event but not the trace.
+        We use the event id within this envelope to deduplicate messages to avoid processing the same event twice, i.e. the event id should be unique per request.
+        The event.body.id is the ID of the actual trace and will be used for updates and will be visible within the Langfuse App.
+        I.e. if you want to update a trace, you'd use the same body id, but separate event IDs.
 
         Notes:
 
@@ -187,28 +201,26 @@ async def main() -> None:
             await client.ingestion.batch(
                 batch=[
                     IngestionEvent_TraceCreate(
+                        id="abcdef-1234-5678-90ab",
+                        timestamp="2022-01-01T00:00:00.000Z",
                         body=TraceBody(
-                            id="string",
+                            id="abcdef-1234-5678-90ab",
                             timestamp=datetime.datetime.fromisoformat(
-                                "2024-01-15 09:30:00+00:00",
+                                "2022-01-01 00:00:00+00:00",
                             ),
-                            name="string",
-                            user_id="string",
-                            input={"key": "value"},
-                            output={"key": "value"},
-                            session_id="string",
-                            release="string",
-                            version="string",
-                            metadata={"key": "value"},
-                            tags=["string"],
+                            name="My Trace",
+                            user_id="1234-5678-90ab-cdef",
+                            input="My input",
+                            output="My output",
+                            session_id="1234-5678-90ab-cdef",
+                            release="1.0.0",
+                            version="1.0.0",
+                            metadata="My metadata",
+                            tags=["tag1", "tag2"],
                             public=True,
                         ),
-                        id="string",
-                        timestamp="string",
-                        metadata={"key": "value"},
                     )
                 ],
-                metadata={"key": "value"},
             )
 
 

langfuse/api/resources/ingestion/types/observation_body.py

@@ -41,6 +41,7 @@ class ObservationBody(pydantic_v1.BaseModel):
     parent_observation_id: typing.Optional[str] = pydantic_v1.Field(
         alias="parentObservationId", default=None
     )
+    environment: typing.Optional[str] = None
 
     def json(self, **kwargs: typing.Any) -> str:
         kwargs_with_defaults: typing.Any = {

langfuse/api/resources/ingestion/types/optional_observation_body.py

@@ -25,6 +25,7 @@ class OptionalObservationBody(pydantic_v1.BaseModel):
         alias="parentObservationId", default=None
     )
     version: typing.Optional[str] = None
+    environment: typing.Optional[str] = None
 
     def json(self, **kwargs: typing.Any) -> str:
         kwargs_with_defaults: typing.Any = {

langfuse/api/resources/ingestion/types/score_body.py

@@ -25,6 +25,7 @@ class ScoreBody(pydantic_v1.BaseModel):
     id: typing.Optional[str] = None
     trace_id: str = pydantic_v1.Field(alias="traceId")
     name: str
+    environment: typing.Optional[str] = None
     value: CreateScoreValue = pydantic_v1.Field()
     """
     The value of the score. Must be passed as string for categorical scores, and numeric for boolean and numeric scores. Boolean score values must equal either 1 or 0 (true or false)

langfuse/api/resources/ingestion/types/trace_body.py

@@ -21,6 +21,7 @@ class TraceBody(pydantic_v1.BaseModel):
     version: typing.Optional[str] = None
     metadata: typing.Optional[typing.Any] = None
     tags: typing.Optional[typing.List[str]] = None
+    environment: typing.Optional[str] = None
     public: typing.Optional[bool] = pydantic_v1.Field(default=None)
     """
     Make trace publicly accessible via url

langfuse/api/resources/score/types/get_scores_response_data.py

@@ -30,6 +30,7 @@ class GetScoresResponseData_Numeric(pydantic_v1.BaseModel):
     comment: typing.Optional[str] = None
     config_id: typing.Optional[str] = pydantic_v1.Field(alias="configId", default=None)
     queue_id: typing.Optional[str] = pydantic_v1.Field(alias="queueId", default=None)
+    environment: typing.Optional[str] = None
     data_type: typing.Literal["NUMERIC"] = pydantic_v1.Field(
         alias="dataType", default="NUMERIC"
     )
@@ -88,6 +89,7 @@ class GetScoresResponseData_Categorical(pydantic_v1.BaseModel):
     comment: typing.Optional[str] = None
     config_id: typing.Optional[str] = pydantic_v1.Field(alias="configId", default=None)
     queue_id: typing.Optional[str] = pydantic_v1.Field(alias="queueId", default=None)
+    environment: typing.Optional[str] = None
     data_type: typing.Literal["CATEGORICAL"] = pydantic_v1.Field(
         alias="dataType", default="CATEGORICAL"
     )
@@ -146,6 +148,7 @@ class GetScoresResponseData_Boolean(pydantic_v1.BaseModel):
     comment: typing.Optional[str] = None
     config_id: typing.Optional[str] = pydantic_v1.Field(alias="configId", default=None)
     queue_id: typing.Optional[str] = pydantic_v1.Field(alias="queueId", default=None)
+    environment: typing.Optional[str] = None
     data_type: typing.Literal["BOOLEAN"] = pydantic_v1.Field(
         alias="dataType", default="BOOLEAN"
     )

langfuse/callback/langchain.py

@@ -87,6 +87,7 @@ def __init__(
         sdk_integration: Optional[str] = None,
         sample_rate: Optional[float] = None,
         mask: Optional[MaskFunction] = None,
+        environment: Optional[str] = None,
     ) -> None:
         LangfuseBaseCallbackHandler.__init__(
             self,
@@ -113,6 +114,7 @@ def __init__(
             sdk_integration=sdk_integration or "langchain",
             sample_rate=sample_rate,
             mask=mask,
+            environment=environment,
         )
 
         self.runs = {}