Merge pull request #991 from parea-ai/PAI-1365-langchain-add-session-…

…id-and-others

add additional info to langchain handler, fix float on parse, try cat…

cookbook/langchain/trace_langchain_simple.py

@@ -22,7 +22,7 @@
 def main():
     return chain.invoke(
         {"input": "Write a Hello World program in Python using FastAPI."},
-        config={"callbacks": [handler]},
+        config={"callbacks": [PareaAILangchainTracer(session_id="123", tags=["fastapi"], metadata={"key": "value"}, end_user_identifier="user123", deployment_id="456")]},
     )
 
 

cookbook/langchain/trace_langchain_with_deployed_prompt.py

@@ -12,7 +12,6 @@
 load_dotenv()
 
 p = Parea(api_key=os.getenv("PAREA_API_KEY"))
-handler = PareaAILangchainTracer()
 
 
 CONTEXT = """Company: Nike. 2023
@@ -63,11 +62,11 @@ def main(question):
             "context": CONTEXT,
             "question": question,
         },
-        config={"callbacks": [handler]},
+        config={"callbacks": [PareaAILangchainTracer(deployment_id="p-JTDYylldIrMbMisT70DJZ")]},
     )
     summary = summary_chain.invoke(
         {"content": answer},
-        config={"callbacks": [handler]},
+        config={"callbacks": [PareaAILangchainTracer(deployment_id="p-OGWAo6yvVKr1hUBY6bmHw")]},
     )
     return summary
 

parea/client.py

@@ -63,9 +63,9 @@
 
 @define
 class Parea:
-    api_key: str = field(init=True, default=os.getenv("PAREA_API_KEY"))
-    project_name: str = field(init=True, default="default")
-    cache: Cache = field(init=True, default=None)
+    api_key: str = field(default=os.getenv("PAREA_API_KEY"))
+    project_name: str = field(default="default")
+    cache: Cache = field(default=None)
     _project: ProjectSchema = field(init=False, default=None)
     _client: HTTPClient = field(init=False, default=HTTPClient())
 
@@ -81,7 +81,7 @@ def __attrs_post_init__(self):
                 parea_logger.set_project_uuid(self.project_uuid)
             except httpx.HTTPStatusError as e:
                 if e.response.status_code == 502:
-                    logger.error(f"Error creating Parea project please try again")
+                    logger.error("Error creating Parea project please try again")
                 else:
                     raise
             parea_logger.set_client(self._client)

parea/helpers.py

@@ -90,8 +90,18 @@ def structure_union_type(obj: Any, cl: type) -> Any:
         else:
             return None
 
+    def structure_float_or_none(obj: Any, cl: type) -> Optional[float]:
+        if obj is None:
+            return None
+        try:
+            return float(obj)
+        except (ValueError, TypeError):
+            return None
+
     converter = GenConverter()
     converter.register_structure_hook(Union[str, Dict[str, str], None], structure_union_type)
+    converter.register_structure_hook(float, structure_float_or_none)
+    converter.register_structure_hook(Optional[float], structure_float_or_none)
     return converter.structure(d, TraceLogTree)
 
 

parea/utils/trace_integrations/langchain.py

@@ -1,4 +1,4 @@
-from typing import Union
+from typing import Any, Dict, List, Optional, Union
 
 import logging
 from uuid import UUID
@@ -19,6 +19,29 @@ class PareaAILangchainTracer(BaseTracer):
     parent_trace_id: UUID
     _parea_root_trace_id: str = None
     _parea_parent_trace_id: str = None
+    _session_id: Optional[str] = None
+    _tags: List[str] = []
+    _metadata: Dict[str, Any] = {}
+    _end_user_identifier: Optional[str] = None
+    _deployment_id: Optional[str] = None
+
+    def __init__(
+        self,
+        session_id: Optional[str] = None,
+        tags: Optional[List[str]] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+        end_user_identifier: Optional[str] = None,
+        deployment_id: Optional[str] = None,
+        **kwargs: Any,
+    ):
+        super().__init__(**kwargs)
+        self._session_id = session_id
+        self._end_user_identifier = end_user_identifier
+        self._deployment_id = deployment_id
+        if tags:
+            self._tags = tags
+        if metadata:
+            self._metadata = metadata
 
     def _persist_run(self, run: Union[Run, LLMRun, ChainRun, ToolRun]) -> None:
         if is_logging_disabled():
@@ -27,6 +50,11 @@ def _persist_run(self, run: Union[Run, LLMRun, ChainRun, ToolRun]) -> None:
         # using .dict() since langchain Run class currently set to Pydantic v1
         data = run.dict()
         data["_parea_root_trace_id"] = self._parea_root_trace_id or None
+        data["_session_id"] = self._session_id
+        data["_tags"] = self._tags
+        data["_metadata"] = self._metadata
+        data["_end_user_identifier"] = self._end_user_identifier
+        data["_deployment_id"] = self._deployment_id
         # check if run has an attribute execution order
         if (hasattr(run, "execution_order") and run.execution_order == 1) or run.parent_run_id is None:
             data["_parea_parent_trace_id"] = self._parea_parent_trace_id or None

parea/utils/universal_encoder.py

@@ -128,4 +128,8 @@ def default(self, obj: Any):
 
 
 def json_dumps(obj, **kwargs) -> str:
-    return json.dumps(obj, cls=UniversalEncoder, **kwargs) if not isinstance(obj, str) else obj
+    try:
+        return json.dumps(obj, cls=UniversalEncoder, **kwargs) if not isinstance(obj, str) else obj
+    except TypeError as e:
+        logger.debug(f"Error serializing object: {obj} with error: {e}")
+        return str(obj)

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "parea-ai"
 packages = [{ include = "parea" }]
-version = "0.2.184"
+version = "0.2.185"
 description = "Parea python sdk"
 readme = "README.md"
 authors = ["joel-parea-ai <[email protected]>"]