LLM Agents and Getting Started with Them

pip install langchain langchain-groq langgraph langchain-core pydantic

#config
import os
from dotenv import load_dotenv

load_dotenv()

# Set your API key
GROQ_API_KEY="YOUR_API_KEY"
os.environ["GROQ_API_KEY"] = GROQ_API_KEY
model = 'openai/gpt-oss-safeguard-20b'

######### llm brain #############

import os
from langchain_groq import ChatGroq

# Setup Groq LLM
llm = ChatGroq(
    temperature=0,
    model_name=model,
    api_key=os.environ.get("GROQ_API_KEY")
)

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. Perception -> Action (Direct Chain)
reflex_prompt = ChatPromptTemplate.from_template(
    "You are a chef. Given a request: {input}, provide a single recipe immediately."
)

reflex_agent = reflex_prompt | llm | StrOutputParser()

#invoke the brain

print(reflex_agent.invoke({"input": "I want a spicy pasta."}))

from langchain.tools import tool
from langchain.chat_models import init_chat_model
import operator
from langgraph.prebuilt import ToolNode, InjectedState
import operator
from typing import Annotated, List, Literal

llm = ChatGroq(
    temperature=0,
    model_name=model,
    api_key=os.environ.get("GROQ_API_KEY")
)



@tool
def get_inventory(state:Annotated[dict, InjectedState]):
  "Get cuurent user inventory"
  return state["inventory"]

@tool
def get_dietary_prefs(state:Annotated[dict, InjectedState]):
  "Get user dietary preferences"
  return state["dietary_prefs"]

@tool
def get_history(state:Annotated[dict, InjectedState]):
  "Get user history"
  return state["history"]


# Augment the LLM with tools
tools = [get_inventory, get_dietary_prefs, get_history]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = llm.bind_tools(tools)

# Step 2: Define state

from langchain.messages import AnyMessage
from typing_extensions import TypedDict
import operator

class RecipeState(TypedDict):
    inventory: List[str]      # What's in the fridge
    dietary_prefs: List[str]  # e.g., "Vegetarian"
    suggestion: str           # The output
    messages: Annotated[List[AnyMessage], operator.add]


# Step 3: Define model node
from langchain.messages import SystemMessage


def llm_call(state: dict):
    """LLM decides whether to call a tool or not"""

    # Combine system message with user messages
    messages_to_send = [
        SystemMessage(
            content="""You are a helpful chef agent. When user asks for a recipe, look at user's current inventory, dietary prefrences to suggest the recipe.
            You can use the available tools whenever needed."""
        )
    ] + state["messages"]

    response = model_with_tools.invoke(messages_to_send)

    return {
        "messages": [response],
        "inventory": state.get("inventory", []),
        "dietary_prefs": state.get("dietary_prefs", [])
    }


# Step 4: Define tool node

from langchain.messages import ToolMessage

tool_node = ToolNode(tools)

# Step 5: Define logic to determine whether to end

from langgraph.graph import StateGraph, START, END


# Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call
def should_continue(state: RecipeState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""

    messages = state["messages"]
    last_message = messages[-1]

    # If the LLM makes a tool call, then perform an action
    if last_message.tool_calls:
        return "tool_node"

    # Otherwise, we stop (reply to the user)
    return END

# Step 6: Build agent

# Build workflow
agent_builder = StateGraph(RecipeState)

# Add nodes
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# Add edges to connect nodes
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

# Compile the agent
agent = agent_builder.compile()


from IPython.display import Image, display
# Show the agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# Invoke
from langchain.messages import HumanMessage
messages = [HumanMessage(content="Suggest a recipe for pasta")]
current_inventory = ["pasta", "water", "tomatoes", "salt","parmesan"]
current_dietary_prefs = ["vegetarian"]
messages = agent.invoke({"inventory":current_inventory,
                         "dietary_prefs":current_dietary_prefs,
                         "messages": messages})
for m in messages["messages"]:
    m.pretty_print()

# Step 1: Define tools and model

from langchain.tools import tool
from langchain.chat_models import init_chat_model
import operator
from langgraph.prebuilt import ToolNode, InjectedState

llm = ChatGroq(
    temperature=0,
    model_name=model,
    api_key=os.environ.get("GROQ_API_KEY")
)



@tool
def get_inventory(state:Annotated[dict, InjectedState]):
  "Get cuurent user inventory"
  return state["inventory"]

@tool
def get_dietary_prefs(state:Annotated[dict, InjectedState]):
  "Ger user dietary prefrences"
  return state["dietary_prefs"]

@tool
def get_remaining_calories_range(state:Annotated[dict, InjectedState]):
  "Get range of remaining calories"
  return (state["total_calories"]- state["consumed_calories"]+state["error_margin_calories"],
          state["total_calories"]- state["consumed_calories"]-state["error_margin_calories"])


# Augment the LLM with tools
tools = [get_inventory, get_dietary_prefs, get_history]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = llm.bind_tools(tools)

# Step 2: Define state

from langchain.messages import AnyMessage,HumanMessage, SystemMessage
from typing_extensions import TypedDict, Annotated
import operator
from typing import List

class RecipeState(TypedDict):
    inventory: List[str]      # What's in the fridge
    dietary_prefs: List[str]  # e.g., "Vegetarian"
    suggestion: str           # The output
    total_calories: int       # The total number of calories to consume
    consumed_calories: int     # The number of calories already_consumed
    error_margin_calories: int  # The number of calories that can be added or deleted from total calories
    num_tries: int            # The number of tries the agent has made
    messages: Annotated[List[AnyMessage], operator.add]


# Step 3: Define Planner Node
def planner_node(state: RecipeState):
    """Suggests a recipe and estimates calories."""
    messages_to_send = [
        SystemMessage(content=(
            "You are a chef. Suggest a recipe based on inventory and dietary prefs. "
            "IMPORTANT: You MUST provide an estimated calorie count for the meal."
            "You can use the available tools whenever needed."
        ))
    ] + state["messages"]

    response = model_with_tools.invoke(messages_to_send)
    return {"messages": [response],
            "num_tries":state.get("num_tries",0)}

def verify_search_node(state: RecipeState):
    """Checks if the suggested meal fits the calorie constraints."""
    last_message = state["messages"][-1].content

    # Simple logic: Ask LLM to extract or verify,
    # or use regex/logic if you want to be strict.
    prompt = f"""
    The user's goal is {state['total_calories']} calories (margin: +/- {state['error_margin_calories']}).
    Current consumed: {state['consumed_calories']}.
    The chef suggested: {last_message}

    Does this recipe fit the remaining calorie budget?
    If yes, reply 'VALID'. If no, explain why.
    You can use the avilable tools whenever needed.
    """

    verification_response = llm.invoke([HumanMessage(content=prompt)])

    # We can add this to messages to keep track of the critique
    return {"messages": [verification_response]}

  
def should_verify(state: RecipeState) -> Literal["verify_node", "tool_node", END]:
    last_message = state["messages"][-1]

    if last_message.tool_calls:
        return "tool_node"

    # If no tool calls, it means the LLM has made a suggestion. Now verify it.
    return "verify_node"

def is_it_valid(state: RecipeState) -> Literal["planner_node", END]:
    last_message = state["messages"][-1].content
    if "VALID" in last_message.upper():
        return END
    # If not valid, loop back to the planner to try again
    return "planner_node"
def num_tries_exceeded(state: RecipeState) -> Literal["planner_node",END]:
    if state["num_tries"] > 5:
        return END
    return "planner_node"


# Build workflow
agent_builder = StateGraph(RecipeState)

agent_builder.add_node("planner_node", planner_node)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_node("verify_node", verify_search_node)

agent_builder.add_edge(START, "planner_node")

# Route from Planner
agent_builder.add_conditional_edges(
    "planner_node",
    should_verify,
    {"tool_node": "tool_node",
     "verify_node": "verify_node",
     END: END}
)

# Route from Tool back to Planner
agent_builder.add_edge("tool_node", "planner_node")

# Route from Verifier back to Planner OR End
agent_builder.add_conditional_edges(
    "verify_node",
    is_it_valid,
    {"planner_node":
     "planner_node",
     END: END}
)

agent = agent_builder.compile()

from IPython.display import Image, display
# Show the agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# Invoke
from langchain.messages import HumanMessage
messages = [HumanMessage(content="Suggest a recipe for pasta")]
current_inventory = ["pasta", "water", "tomatoes", "salt","parmesan"]
current_dietary_prefs = ["vegetarian"]
messages = agent.invoke({"inventory":current_inventory,
                         "dietary_prefs":current_dietary_prefs,
                         "total_calories": 1000,
                         "consumed_calories": 800,
                         "error_margin_calories": 100,
                         "messages": messages})
for m in messages["messages"]:
    m.pretty_print()