LangGraph
import os
from typing import Annotated
from typing_extensions import TypedDict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
# Load environment variables
load_dotenv()
# --- Configuration ---
NEXOS_BASE_URL = os.getenv("NEXOS_BASE_URL")
NEXOS_API_KEY = os.getenv("NEXOS_API_KEY")
if not NEXOS_BASE_URL or not NEXOS_API_KEY:
raise ValueError("Please set NEXOS_BASE_URL and NEXOS_API_KEY in your .env file")
# --- 1. Define the State ---
class State(TypedDict):
# The 'add_messages' reducer appends new messages to the existing list
messages: Annotated[list[BaseMessage], add_messages]
# --- 2. Initialize the LLM ---
llm = ChatOpenAI(
model="gemini-2.5-flash", # or any other OpenAI-compatible model ID available to you
base_url=NEXOS_BASE_URL, # e.g. "https://api.nexos.ai/v1"
api_key=NEXOS_API_KEY,
temperature=0.7,
)
# --- 3. Define Nodes ---
def chatbot_node(state: State):
"""
Invokes the LLM with the current history of messages.
Returns a dictionary with the new message to be added to the state.
"""
response = llm.invoke(state["messages"])
return {"messages": [response]}
# --- 4. Build the Graph ---
builder = StateGraph(State)
# Add nodes
builder.add_node("chatbot", chatbot_node)
# Add edges (Simple linear flow: Start -> Chatbot -> End)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
# Compile the graph
graph = builder.compile()
def main():
print("--- Starting LangGraph Execution ---")
# Initial input to the graph
initial_input = {"messages": [HumanMessage(content="Hello! Explain the concept of a 'graph' in one sentence.")]}
# Stream the execution
# The stream yields events as the graph progresses
for event in graph.stream(initial_input):
for node_name, value in event.items():
print(f"\n--- Output from node '{node_name}' ---")
last_message = value["messages"][-1]
print(last_message.content)
print("--------------------------------------")
if __name__ == "__main__":
main()Last updated

