Langchain: How do I define an output parser like a dictionary?

README.md

LangChain Market Analysis Tool (Example)

This Python tool leverages the LangChain framework to suggest products that could be successful in various industries and performs market analysis through parallel execution chains. It uses OpenAI's GPT-3.5 model for generating insights based on the given industry sector.

Features

• Model Definitions: Defines Pydantic models for handling company, subsidiary, and ticker data, ensuring type safety and easy data manipulation.
• Sequential Chains: Constructs a sequential chain for generating product suggestions within a specific industry.
• Parallel Chains: Implements parallel execution chains to analyze multiple industries simultaneously, showcasing the power of RunnableParallel for concurrent data processing.
• Dynamic Environment Configuration: Uses environment variables for OpenAI API credentials, promoting security and ease of configuration.

Setup

1. Install Dependencies: Ensure you have Python 3.6 or later installed. Install the required packages (e.g., langchain_core, langchain_openai, pydantic) via pip:

   pip install langchain-core langchain-openai pydantic

2. Environment Variables: The script requires your OpenAI API Key. It will prompt for this key at runtime. Optionally, you can set the OPENAI_API_KEY environment variable manually before running the script.

3. Configuration: The default model is set to "gpt-3.5-turbo-0125". Adjust this in the code if necessary.

Running the Tool

To run the tool, simply execute the script in your terminal:

python market_analysis_tool.py

You will be prompted to enter your OpenAI API Key. After entering the key, the tool will perform both sequential and parallel market analyses, printing the results to the console.

Sequential Chain

The sequential chain suggests a product for a specified industry sector. It dynamically constructs a prompt, sends it to OpenAI's model, and parses the output into a Pydantic model for easy access and manipulation.

Parallel Chains

The parallel chain analyzes multiple industries simultaneously, using RunnableParallel to execute multiple chains in parallel. This feature is particularly useful for broad market analyses across various sectors.

Security Note

Your OpenAI API Key is sensitive information. The script prompts for this key at runtime to avoid hard-coding it into the source code. Always ensure your API keys are kept secure and are not exposed in shared or public code repositories.

---

output_parralel_chain.json

{
    "key": "company_one",
    "value": {
        "industry": "Fundraising and Nonprofit Management",
        "name": "Fundraiser Pro",
        "product": "Donation Management Software",
        "subsidiaries": [
            {
                "name": "DonorConnect",
                "ticker_data": {
                    "close": 55.75,
                    "high": 60.0,
                    "low": 45.5,
                    "open": 50.25,
                    "ticker": "DCON",
                    "volume": 100000.0,
                },
            },
            {
                "name": "FundraiserTech",
                "ticker_data": {
                    "close": 35.25,
                    "high": 40.75,
                    "low": 28.0,
                    "open": 30.5,
                    "ticker": "FTECH",
                    "volume": 75000.0,
                },
            },
        ],
    },
},
{
    "key": "company_two",
    "value": {
        "industry": "Fundraising and Nonprofit Management",
        "name": "FundraisePro",
        "product": "Fundraising Software",
        "subsidiaries": [
            {
                "name": "DonorConnect",
                "ticker_data": {
                    "close": 55.75,
                    "high": 60.0,
                    "low": 45.5,
                    "open": 50.25,
                    "ticker": "DONR",
                    "volume": 10000.0,
                },
            },
            {
                "name": "EventRaise",
                "ticker_data": {
                    "close": 35.25,
                    "high": 40.75,
                    "low": 28.0,
                    "open": 30.5,
                    "ticker": "EVNT",
                    "volume": 7500.0,
                },
            },
        ],
    },
},
{
    "key": "company_three",
    "value": {
        "industry": "Fundraising and Nonprofit Management",
        "name": "FundraisingTech",
        "product": "Donation Management Software",
        "subsidiaries": [
            {
                "name": "DonorConnect",
                "ticker_data": {
                    "close": 51.75,
                    "high": 52.0,
                    "low": 50.0,
                    "open": 50.25,
                    "ticker": "DCON",
                    "volume": 10000.0,
                },
            },
            {
                "name": "FundraiserPro",
                "ticker_data": {
                    "close": 77.25,
                    "high": 78.0,
                    "low": 75.0,
                    "open": 75.5,
                    "ticker": "FPRO",
                    "volume": 15000.0,
                },
            },
        ],
    },
}

output_sequential_chain.json

{
  "industry": "Adult Toys",
  "name": "Pleasure Palace",
  "product": "Ultimate Pleasure Wand",
  "subsidiaries": [
    {
      "name": "Pleasure Palace Europe",
      "ticker_data": {
        "close": 52.75,
        "high": 55.0,
        "low": 49.75,
        "open": 50.25,
        "ticker": "PPE",
        "volume": 100000.0
      }
    },
    {
      "name": "Pleasure Palace Asia",
      "ticker_data": {
        "close": 58.25,
        "high": 62.0,
        "low": 57.75,
        "open": 60.5,
        "ticker": "PPA",
        "volume": 75000.0
      }
    }
  ]
}

pydantic_output_parser.py

from langchain_core.runnables import (
    RunnableParallel,
)
from pprint import pprint
import os
from typing import List, Dict, Any

from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

from getpass import getpass

os.environ["OPENAI_API_KEY"] = getpass("Enter your OpenAI API Key: ")
os.environ["OPENAI_MODEL"] = "gpt-3.5-turbo-0125"
llm = ChatOpenAI()

"""
Model
"""


class TickerData(BaseModel):
    ticker: str = Field(description="stock ticker")
    volume: float
    open: float
    close: float
    high: float
    low: float


class Subsidiary(BaseModel):
    name: str = Field(description="subsidiary name")
    ticker_data: TickerData


class Company(BaseModel):
    name: str = Field(description="company name")
    product: str = Field(description="product name")
    industry: str = Field(description="industry name")
    subsidiaries: List[Subsidiary] = Field(description="subsidiaries")


"""
Sequential Chains
"""

parser = PydanticOutputParser(pydantic_object=Company)

prompt = PromptTemplate(
    template="Suggest a product that would sell like hotcakes in the {industry} sector\n{format_instructions}\n",
    input_variables=["industry"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

chain = prompt | ChatOpenAI() | parser

"""
Parallel Chains
"""


class Market(BaseModel):
    industry_one: str = Field(description="industry one")
    industry_two: str = Field(description="industry two")
    industry_three: str = Field(description="industry three")


market_parser = PydanticOutputParser(pydantic_object=Market)

format_instructions = market_parser.get_format_instructions()
market_prompt_template = PromptTemplate(
    template="What are three industries in the {market}\n{instructions}\n",
    input_variables=["market"],
    partial_variables={"instructions": format_instructions},
)
market_runnable = market_prompt_template | ChatOpenAI() | market_parser

chain_inputs = RunnableParallel(
    company_one={"industry": lambda x: x.industry_one} | chain,
    company_two={"industry": lambda x: x.industry_two} | chain,
    company_three={"industry": lambda x: x.industry_three} | chain,
)

market_chain = market_runnable | chain_inputs


if __name__ == "__main__":
    print("Running Sequential Chain...")
    company = chain.invoke({"industry": "Adult Toys"})
    pprint(company.dict(), indent=4)
    print("Running Parallel Chain...")
    response = market_chain.invoke({"market": "Televangelism"})
    for key, value in response.items():
        pprint({"key": key, "value": value.dict()}, indent=4)