github/langchain
1
0
Fork 0
mirror of https://github.com/hwchase17/langchain.git synced 2025-06-27 08:58:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

community: add Valthera integration (#30105)

Browse Source 
```markdown
**Description:**  
This PR integrates Valthera into LangChain, introducing an framework designed to send highly personalized nudges by an LLM agent. This is modeled after Dr. BJ Fogg's Behavior Model. This integration includes:

- Custom data connectors for HubSpot, PostHog, and Snowflake.
- A unified data aggregator that consolidates user data.
- Scoring configurations to compute motivation and ability scores.
- A reasoning engine that determines the appropriate user action.
- A trigger generator to create personalized messages for user engagement.

**Issue:**  
N/A

**Dependencies:**  
N/A

**Twitter handle:**  
- `@vselvarajijay`

**Tests and Docs:**  
- `docs/docs/integrations/tools/valthera` 
- `https://github.com/valthera/langchain-valthera/tree/main/tests`

```

---------

Co-authored-by: Chester Curme <chester.curme@gmail.com>
This commit is contained in:
Vijay Selvaraj 2025-03-09 14:19:08 -07:00 committed by GitHub
parent 3823daa0b9
commit df459d0d5e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 506 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
docs/docs/integrations/providers/valthera.mdx Normal file
View File

@ -0,0 +1,63 @@
# Valthera
> [Valthera](https://github.com/valthera/valthera) is an open-source framework that empowers LLM Agents to drive meaningful, context-aware user engagement. It evaluates user motivation and ability in real time, ensuring that notifications and actions are triggered only when users are most receptive.
>
> **langchain-valthera** integrates Valthera with LangChain, enabling developers to build smarter, behavior-driven engagement systems that deliver personalized interactions.
## Installation and Setup
### Install langchain-valthera
Install the LangChain Valthera package via pip:
```bash
pip install -U langchain-valthera
```
Import the ValtheraTool:
```python
from langchain_valthera.tools import ValtheraTool
```
### Example: Initializing the ValtheraTool for LangChain
This example shows how to initialize the ValtheraTool using a `DataAggregator` and configuration for motivation and ability scoring.
```python
import os
from langchain_openai import ChatOpenAI
from valthera.aggregator import DataAggregator
from mocks import hubspot, posthog, snowflake # Replace these with your actual connector implementations
from langchain_valthera.tools import ValtheraTool
# Initialize the DataAggregator with your data connectors
data_aggregator = DataAggregator(
connectors={
"hubspot": hubspot(),
"posthog": posthog(),
"app_db": snowflake()
}
)
# Initialize the ValtheraTool with your scoring configurations
valthera_tool = ValtheraTool(
data_aggregator=data_aggregator,
motivation_config=[
{"key": "hubspot_lead_score", "weight": 0.30, "transform": lambda x: min(x, 100) / 100.0},
{"key": "posthog_events_count_past_30days", "weight": 0.30, "transform": lambda x: min(x, 50) / 50.0},
{"key": "hubspot_marketing_emails_opened", "weight": 0.20, "transform": lambda x: min(x / 10.0, 1.0)},
{"key": "posthog_session_count", "weight": 0.20, "transform": lambda x: min(x / 5.0, 1.0)}
],
ability_config=[
{"key": "posthog_onboarding_steps_completed", "weight": 0.30, "transform": lambda x: min(x / 5.0, 1.0)},
{"key": "posthog_session_count", "weight": 0.30, "transform": lambda x: min(x / 10.0, 1.0)},
{"key": "behavior_complexity", "weight": 0.40, "transform": lambda x: 1 - (min(x, 5) / 5.0)}
]
)
print("✅ ValtheraTool successfully initialized for LangChain integration!")
```
The langchain-valthera integration allows you to assess user behavior and decide on the best course of action for engagement, ensuring that interactions are both timely and relevant within your LangChain applications.

439
docs/docs/integrations/tools/valthera.ipynb Normal file
View File

@ -0,0 +1,439 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"sidebar_label: Valthera\n",
"---"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Valthera\n",
"\n",
"Enable AI agents to engage users when they're most likely to respond.\n",
"\n",
"## Overview\n",
"\n",
"Valthera is an open-source framework that enables LLM Agents to engage users in a more meaningful way. It is built on BJ Fogg's Behavior Model (B=MAT) and leverages data from multiple sources (such as HubSpot, PostHog, and Snowflake) to assess a user's **motivation** and **ability** before triggering an action.\n",
"\n",
"In this guide, you'll learn:\n",
"\n",
"- **Core Concepts:** Overview of the components (Data Aggregator, Scorer, Reasoning Engine, and Trigger Generator).\n",
"- **System Architecture:** How data flows through the system and how decisions are made.\n",
"- **Customization:** How to extend connectors, scoring metrics, and decision rules to fit your needs.\n",
"\n",
"Let's dive in!\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Setup\n",
"\n",
"This section covers installation of dependencies and setting up custom data connectors for Valthera."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"vscode": {
"languageId": "shellscript"
}
},
"outputs": [],
"source": [
"pip install openai langchain langchain_openai valthera langchain_valthera langgraph"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from typing import Any, Dict, List\n",
"\n",
"from valthera.connectors.base import BaseConnector\n",
"\n",
"\n",
"class MockHubSpotConnector(BaseConnector):\n",
" \"\"\"\n",
" Simulates data retrieval from HubSpot. Provides information such as lead score,\n",
" lifecycle stage, and marketing metrics.\n",
" \"\"\"\n",
"\n",
" def get_user_data(self, user_id: str) -> Dict[str, Any]:\n",
" \"\"\"\n",
" Retrieve mock HubSpot data for a given user.\n",
"\n",
" Args:\n",
" user_id: The unique identifier for the user\n",
"\n",
" Returns:\n",
" A dictionary containing HubSpot user data\n",
" \"\"\"\n",
" return {\n",
" \"hubspot_contact_id\": \"999-ZZZ\",\n",
" \"lifecycle_stage\": \"opportunity\",\n",
" \"lead_status\": \"engaged\",\n",
" \"hubspot_lead_score\": 100,\n",
" \"company_name\": \"MaxMotivation Corp.\",\n",
" \"last_contacted_date\": \"2023-09-20\",\n",
" \"hubspot_marketing_emails_opened\": 20,\n",
" \"marketing_emails_clicked\": 10,\n",
" }\n",
"\n",
"\n",
"class MockPostHogConnector(BaseConnector):\n",
" \"\"\"\n",
" Simulates data retrieval from PostHog. Provides session data and engagement events.\n",
" \"\"\"\n",
"\n",
" def get_user_data(self, user_id: str) -> Dict[str, Any]:\n",
" \"\"\"\n",
" Retrieve mock PostHog data for a given user.\n",
"\n",
" Args:\n",
" user_id: The unique identifier for the user\n",
"\n",
" Returns:\n",
" A dictionary containing PostHog user data\n",
" \"\"\"\n",
" return {\n",
" \"distinct_ids\": [user_id, f\"email_{user_id}\"],\n",
" \"last_event_timestamp\": \"2023-09-20T12:34:56Z\",\n",
" \"feature_flags\": [\"beta_dashboard\", \"early_access\"],\n",
" \"posthog_session_count\": 30,\n",
" \"avg_session_duration_sec\": 400,\n",
" \"recent_event_types\": [\"pageview\", \"button_click\", \"premium_feature_used\"],\n",
" \"posthog_events_count_past_30days\": 80,\n",
" \"posthog_onboarding_steps_completed\": 5,\n",
" }\n",
"\n",
"\n",
"class MockSnowflakeConnector(BaseConnector):\n",
" \"\"\"\n",
" Simulates retrieval of additional user profile data from Snowflake.\n",
" \"\"\"\n",
"\n",
" def get_user_data(self, user_id: str) -> Dict[str, Any]:\n",
" \"\"\"\n",
" Retrieve mock Snowflake data for a given user.\n",
"\n",
" Args:\n",
" user_id: The unique identifier for the user\n",
"\n",
" Returns:\n",
" A dictionary containing Snowflake user data\n",
" \"\"\"\n",
" return {\n",
" \"user_id\": user_id,\n",
" \"email\": f\"{user_id}@example.com\",\n",
" \"subscription_status\": \"paid\",\n",
" \"plan_tier\": \"premium\",\n",
" \"account_creation_date\": \"2023-01-01\",\n",
" \"preferred_language\": \"en\",\n",
" \"last_login_datetime\": \"2023-09-20T12:00:00Z\",\n",
" \"behavior_complexity\": 3,\n",
" }"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Instantiation\n",
"\n",
"In this section, we instantiate the core components. First, we create a Data Aggregator to combine data from the custom connectors. Then, we configure the scoring metrics for motivation and ability."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from valthera.aggregator import DataAggregator\n",
"\n",
"# Constants for configuration\n",
"LEAD_SCORE_MAX = 100\n",
"EVENTS_COUNT_MAX = 50\n",
"EMAILS_OPENED_FACTOR = 10.0\n",
"SESSION_COUNT_FACTOR_1 = 5.0\n",
"ONBOARDING_STEPS_FACTOR = 5.0\n",
"SESSION_COUNT_FACTOR_2 = 10.0\n",
"BEHAVIOR_COMPLEXITY_MAX = 5.0\n",
"\n",
"# Initialize data aggregator\n",
"data_aggregator = DataAggregator(\n",
" connectors={\n",
" \"hubspot\": MockHubSpotConnector(),\n",
" \"posthog\": MockPostHogConnector(),\n",
" \"snowflake\": MockSnowflakeConnector(),\n",
" }\n",
")\n",
"\n",
"# You can now fetch unified user data by calling data_aggregator.get_user_context(user_id)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from typing import Callable, Union\n",
"\n",
"from valthera.scorer import ValtheraScorer\n",
"\n",
"\n",
"# Define transform functions with proper type annotations\n",
"def transform_lead_score(x: Union[int, float]) -> float:\n",
" \"\"\"Transform lead score to a value between 0 and 1.\"\"\"\n",
" return min(x, LEAD_SCORE_MAX) / LEAD_SCORE_MAX\n",
"\n",
"\n",
"def transform_events_count(x: Union[int, float]) -> float:\n",
" \"\"\"Transform events count to a value between 0 and 1.\"\"\"\n",
" return min(x, EVENTS_COUNT_MAX) / EVENTS_COUNT_MAX\n",
"\n",
"\n",
"def transform_emails_opened(x: Union[int, float]) -> float:\n",
" \"\"\"Transform emails opened to a value between 0 and 1.\"\"\"\n",
" return min(x / EMAILS_OPENED_FACTOR, 1.0)\n",
"\n",
"\n",
"def transform_session_count_1(x: Union[int, float]) -> float:\n",
" \"\"\"Transform session count for motivation to a value between 0 and 1.\"\"\"\n",
" return min(x / SESSION_COUNT_FACTOR_1, 1.0)\n",
"\n",
"\n",
"def transform_onboarding_steps(x: Union[int, float]) -> float:\n",
" \"\"\"Transform onboarding steps to a value between 0 and 1.\"\"\"\n",
" return min(x / ONBOARDING_STEPS_FACTOR, 1.0)\n",
"\n",
"\n",
"def transform_session_count_2(x: Union[int, float]) -> float:\n",
" \"\"\"Transform session count for ability to a value between 0 and 1.\"\"\"\n",
" return min(x / SESSION_COUNT_FACTOR_2, 1.0)\n",
"\n",
"\n",
"def transform_behavior_complexity(x: Union[int, float]) -> float:\n",
" \"\"\"Transform behavior complexity to a value between 0 and 1.\"\"\"\n",
" return 1 - (min(x, BEHAVIOR_COMPLEXITY_MAX) / BEHAVIOR_COMPLEXITY_MAX)\n",
"\n",
"\n",
"# Scoring configuration for user motivation\n",
"motivation_config = [\n",
" {\"key\": \"hubspot_lead_score\", \"weight\": 0.30, \"transform\": transform_lead_score},\n",
" {\n",
" \"key\": \"posthog_events_count_past_30days\",\n",
" \"weight\": 0.30,\n",
" \"transform\": transform_events_count,\n",
" },\n",
" {\n",
" \"key\": \"hubspot_marketing_emails_opened\",\n",
" \"weight\": 0.20,\n",
" \"transform\": transform_emails_opened,\n",
" },\n",
" {\n",
" \"key\": \"posthog_session_count\",\n",
" \"weight\": 0.20,\n",
" \"transform\": transform_session_count_1,\n",
" },\n",
"]\n",
"\n",
"# Scoring configuration for user ability\n",
"ability_config = [\n",
" {\n",
" \"key\": \"posthog_onboarding_steps_completed\",\n",
" \"weight\": 0.30,\n",
" \"transform\": transform_onboarding_steps,\n",
" },\n",
" {\n",
" \"key\": \"posthog_session_count\",\n",
" \"weight\": 0.30,\n",
" \"transform\": transform_session_count_2,\n",
" },\n",
" {\n",
" \"key\": \"behavior_complexity\",\n",
" \"weight\": 0.40,\n",
" \"transform\": transform_behavior_complexity,\n",
" },\n",
"]\n",
"\n",
"# Instantiate the scorer\n",
"scorer = ValtheraScorer(motivation_config, ability_config)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Invocation\n",
"\n",
"Next, we set up the Reasoning Engine and Trigger Generator, then bring all components together by instantiating the Valthera Tool. Finally, we execute the agent workflow to process an input message."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"\n",
"from langchain_openai import ChatOpenAI\n",
"from valthera.reasoning_engine import ReasoningEngine\n",
"\n",
"# Define threshold as constant\n",
"SCORE_THRESHOLD = 0.75\n",
"\n",
"\n",
"# Function to safely get API key\n",
"def get_openai_api_key() -> str:\n",
" \"\"\"Get OpenAI API key with error handling.\"\"\"\n",
" api_key = os.environ.get(\"OPENAI_API_KEY\")\n",
" if not api_key:\n",
" raise ValueError(\"OPENAI_API_KEY not found in environment variables\")\n",
" return api_key\n",
"\n",
"\n",
"# Decision rules using constant\n",
"decision_rules = [\n",
" {\n",
" \"condition\": f\"motivation >= {SCORE_THRESHOLD} and ability >= {SCORE_THRESHOLD}\",\n",
" \"action\": \"trigger\",\n",
" \"description\": \"Both scores are high enough.\",\n",
" },\n",
" {\n",
" \"condition\": f\"motivation < {SCORE_THRESHOLD}\",\n",
" \"action\": \"improve_motivation\",\n",
" \"description\": \"User motivation is low.\",\n",
" },\n",
" {\n",
" \"condition\": f\"ability < {SCORE_THRESHOLD}\",\n",
" \"action\": \"improve_ability\",\n",
" \"description\": \"User ability is low.\",\n",
" },\n",
" {\n",
" \"condition\": \"otherwise\",\n",
" \"action\": \"defer\",\n",
" \"description\": \"No action needed at this time.\",\n",
" },\n",
"]\n",
"\n",
"try:\n",
" api_key = get_openai_api_key()\n",
" reasoning_engine = ReasoningEngine(\n",
" llm=ChatOpenAI(\n",
" model_name=\"gpt-4-turbo\", temperature=0.0, openai_api_key=api_key\n",
" ),\n",
" decision_rules=decision_rules,\n",
" )\n",
"except ValueError as e:\n",
" print(f\"Error initializing reasoning engine: {e}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from valthera.trigger_generator import TriggerGenerator\n",
"\n",
"try:\n",
" api_key = get_openai_api_key() # Reuse the function for consistency\n",
" trigger_generator = TriggerGenerator(\n",
" llm=ChatOpenAI(\n",
" model_name=\"gpt-4-turbo\", temperature=0.7, openai_api_key=api_key\n",
" )\n",
" )\n",
"except ValueError as e:\n",
" print(f\"Error initializing trigger generator: {e}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_valthera.tools import ValtheraTool\n",
"from langgraph.prebuilt import create_react_agent\n",
"\n",
"try:\n",
" api_key = get_openai_api_key()\n",
"\n",
" # Initialize Valthera tool\n",
" valthera_tool = ValtheraTool(\n",
" data_aggregator=data_aggregator,\n",
" motivation_config=motivation_config,\n",
" ability_config=ability_config,\n",
" reasoning_engine=reasoning_engine,\n",
" trigger_generator=trigger_generator,\n",
" )\n",
"\n",
" # Create agent with LLM\n",
" llm = ChatOpenAI(model_name=\"gpt-4-turbo\", temperature=0.0, openai_api_key=api_key)\n",
" tools = [valthera_tool]\n",
" graph = create_react_agent(llm, tools=tools)\n",
"\n",
" # Define input message for testing\n",
" inputs = {\n",
" \"messages\": [(\"user\", \"Evaluate behavior for user_12345: Finish Onboarding\")]\n",
" }\n",
"\n",
" # Process the input and display responses\n",
" print(\"Running Valthera agent workflow...\")\n",
" for response in graph.stream(inputs, stream_mode=\"values\"):\n",
" print(response)\n",
"\n",
"except Exception as e:\n",
" print(f\"Error running Valthera workflow: {e}\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Chaining\n",
"\n",
"This integration does not currently support chaining operations. Future releases may include chaining support."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## API reference\n",
"\n",
"Below is an overview of the key APIs provided by the Valthera integration:\n",
"\n",
"- **Data Aggregator:** Use `data_aggregator.get_user_context(user_id)` to fetch aggregated user data.\n",
"- **Scorer:** The `ValtheraScorer` computes motivation and ability scores based on the provided configurations.\n",
"- **Reasoning Engine:** The `ReasoningEngine` evaluates decision rules to determine the appropriate action (trigger, improve motivation, improve ability, or defer).\n",
"- **Trigger Generator:** Generates personalized trigger messages using the LLM.\n",
"- **Valthera Tool:** Integrates all the components to process inputs and execute the agent workflow.\n",
"\n",
"For detailed usage, refer to the inline documentation in the source code."
]
}
],
"metadata": {
"language_info": {
"name": "python"
},
"title": "Valthera Developer Guide"
},
"nbformat": 4,
"nbformat_minor": 2
}

4
libs/packages.yml
View File

@ -506,3 +506,7 @@ packages:
repo: ContextualAI//langchain-contextual
downloads: 432
downloads_updated_at: '2025-03-09T01:40:49.430540+00:00'
- name: langchain-valthera
name_title: Valthera
path: .
repo: valthera/langchain-valthera