Skip to main content
Open In ColabOpen on GitHub

valthera

Overview​

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.

In this guide, you'll learn:

  • Core Concepts: Overview of the components (Data Aggregator, Scorer, Reasoning Engine, and Trigger Generator).
  • System Architecture: How data flows through the system and how decisions are made.
  • Customization: How to extend connectors, scoring metrics, and decision rules to fit your needs.

Let's dive in!

Setup​

This section covers installation of dependencies and setting up custom data connectors for Valthera.

pip install openai langchain langchain_openai valthera langchain_valthera langgraph
from valthera.connectors.base import BaseConnector

class MockHubSpotConnector(BaseConnector):
    """
    Simulates data retrieval from HubSpot. Provides information such as lead score, lifecycle stage, and marketing metrics.
    """
    def get_user_data(self, user_id: str):
        return {
            "hubspot_contact_id": "999-ZZZ",
            "lifecycle_stage": "opportunity",
            "lead_status": "engaged",
            "lead_score": 100,  # 100 implies full lead score factor
            "company_name": "MaxMotivation Corp.",
            "last_contacted_date": "2023-09-20",
            "marketing_emails_opened": 20,
            "marketing_emails_clicked": 10            
        }


class MockPostHogConnector(BaseConnector):
    """
    Simulates data retrieval from PostHog. Provides session data and engagement events.
    """
    def get_user_data(self, user_id: str):
        return {
            "distinct_ids": [user_id, f"email_{user_id}"],
            "last_event_timestamp": "2023-09-20T12:34:56Z",
            "feature_flags": ["beta_dashboard", "early_access"],
            "session_count": 30,
            "avg_session_duration_sec": 400,
            "recent_event_types": ["pageview", "button_click", "premium_feature_used"],
            "events_count_past_30days": 80,
            "onboarding_steps_completed": 5
        }


class MockSnowflakeConnector(BaseConnector):
    """
    Simulates retrieval of additional user profile data from Snowflake.
    """
    def get_user_data(self, user_id: str):
        return {
            "user_id": user_id,
            "email": f"{user_id}@example.com",
            "subscription_status": "paid",
            "plan_tier": "premium",
            "account_creation_date": "2023-01-01",
            "preferred_language": "en",
            "last_login_datetime": "2023-09-20T12:00:00Z"
        }

Instantiation​

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.

from valthera.aggregator import DataAggregator

data_aggregator = DataAggregator(
    connectors={
        "hubspot": MockHubSpotConnector(),
        "posthog": MockPostHogConnector(),
        "snowflake": MockSnowflakeConnector()
    }
)

# You can now fetch unified user data by calling data_aggregator.get_user_context(user_id)
from valthera.scorer import ValtheraScorer

# Scoring configuration for user motivation
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)}
]

# Scoring configuration for user ability
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)}
]

# Instantiate the scorer
scorer = ValtheraScorer(motivation_config, ability_config)

Invocation​

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.

import os
from valthera.reasoning_engine import ReasoningEngine
from langchain_openai import ChatOpenAI

decision_rules = [
    {"condition": "motivation >= 0.75 and ability >= 0.75", "action": "trigger", "description": "Both scores are high enough."},
    {"condition": "motivation < 0.75", "action": "improve_motivation", "description": "User motivation is low."},
    {"condition": "ability < 0.75", "action": "improve_ability", "description": "User ability is low."},
    {"condition": "otherwise", "action": "defer", "description": "No action needed at this time."}
]

reasoning_engine = ReasoningEngine(
    llm=ChatOpenAI(
        model_name="gpt-4-turbo",
        temperature=0.0,
        openai_api_key=os.environ.get("OPENAI_API_KEY")
    ),
    decision_rules=decision_rules
)
API Reference:ChatOpenAI
from valthera.trigger_generator import TriggerGenerator

trigger_generator = TriggerGenerator(
    llm=ChatOpenAI(
        model_name="gpt-4-turbo",
        temperature=0.7,
        openai_api_key=os.environ.get("OPENAI_API_KEY")
    )
)
from langchain_valthera.tools import ValtheraTool
from langgraph.prebuilt import create_react_agent

valthera_tool = ValtheraTool(
    data_aggregator=data_aggregator,
    motivation_config=motivation_config,
    ability_config=ability_config,
    reasoning_engine=reasoning_engine,
    trigger_generator=trigger_generator
)

llm = ChatOpenAI(model_name="gpt-4-turbo", temperature=0.0, openai_api_key=os.environ.get("OPENAI_API_KEY"))
tools = [valthera_tool]
graph = create_react_agent(llm, tools=tools)

inputs = {
    "messages": [("user", "Evaluate behavior for user_12345: Finish Onboarding")]
}

for response in graph.stream(inputs, stream_mode="values"):
    print(response)
API Reference:create_react_agent

Chaining​

This integration does not currently support chaining operations. Future releases may include chaining support.

API reference​

Below is an overview of the key APIs provided by the Valthera integration:

  • Data Aggregator: Use data_aggregator.get_user_context(user_id) to fetch aggregated user data.
  • Scorer: The ValtheraScorer computes motivation and ability scores based on the provided configurations.
  • Reasoning Engine: The ReasoningEngine evaluates decision rules to determine the appropriate action (trigger, improve motivation, improve ability, or defer).
  • Trigger Generator: Generates personalized trigger messages using the LLM.
  • Valthera Tool: Integrates all the components to process inputs and execute the agent workflow.

For detailed usage, refer to the inline documentation in the source code.

Was this page helpful?