From e5d701a045bb63e74d837e039b8d42e4487463fb Mon Sep 17 00:00:00 2001
From: Eric Wang <wangx184@chinatelecom.cn>
Date: Sat, 25 Jul 2026 01:59:24 +0800
Subject: [PATCH] Initial commit: add LangGraph examples

---
 .gitignore               |  23 ++++
 basic_agent.py           | 115 +++++++++++++++++++
 chatopenai_structured.py | 231 +++++++++++++++++++++++++++++++++++++++
 langgraph_chatopenai.py  | 198 +++++++++++++++++++++++++++++++++
 4 files changed, 567 insertions(+)
 create mode 100644 .gitignore
 create mode 100644 basic_agent.py
 create mode 100644 chatopenai_structured.py
 create mode 100644 langgraph_chatopenai.py

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..e0c7f39
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,23 @@
+# Environment variables and secrets
+.env
+.env.*
+!.env.example
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# IDE and OS files
+.vscode/
+.idea/
+.DS_Store
+Thumbs.db
diff --git a/basic_agent.py b/basic_agent.py
new file mode 100644
index 0000000..2fb62dc
--- /dev/null
+++ b/basic_agent.py
@@ -0,0 +1,115 @@
+"""
+Basic LangGraph Example - A simple conversational agent with state management.
+
+This demonstrates core LangGraph concepts:
+- State schemas
+- Nodes (functions that process state)
+- Edges (connections between nodes)
+- Conditional routing
+"""
+
+from typing import TypedDict, Literal, Annotated
+from langgraph.graph import StateGraph, START, END
+from langgraph.graph.message import add_messages
+
+
+# Define the state schema
+class State(TypedDict):
+    """State represents what flows through our graph."""
+    messages: Annotated[list, add_messages]
+    counter: int
+
+
+# Define node functions
+def greeting_node(state: State) -> dict:
+    """Process greeting messages."""
+    print("Processing greeting...")
+    return {"counter": state["counter"] + 1}
+
+
+def math_node(state: State) -> dict:
+    """Process math-related messages."""
+    print("Processing math request...")
+    return {"counter": state["counter"] + 1}
+
+
+def default_node(state: State) -> dict:
+    """Handle unknown/unrecognized messages."""
+    print("Handling general message...")
+    return {"counter": state["counter"] + 1}
+
+
+# Define the router function
+def route_message(state: State) -> Literal["greeting", "math", "default"]:
+    """Route to appropriate node based on message content."""
+    messages = state["messages"]
+    if not messages:
+        return "default"
+
+    last_message = messages[-1].lower() if isinstance(messages[-1], str) else str(messages[-1]).lower()
+
+    if any(word in last_message for word in ["hello", "hi", "hey", "greetings"]):
+        return "greeting"
+    elif any(word in last_message for word in ["add", "subtract", "multiply", "divide", "calculate"]):
+        return "math"
+    else:
+        return "default"
+
+
+# Build the graph
+def create_graph():
+    """Create and compile the LangGraph workflow."""
+
+    # Initialize the graph builder
+    builder = StateGraph(State)
+
+    # Add nodes
+    builder.add_node("greeting", greeting_node)
+    builder.add_node("math", math_node)
+    builder.add_node("default", default_node)
+
+    # Add edges with conditional routing
+    builder.add_conditional_edges(
+        START,
+        route_message,
+        {
+            "greeting": "greeting",
+            "math": "math",
+            "default": "default"
+        }
+    )
+
+    # All nodes lead to END (or could chain to other nodes)
+    builder.add_edge("greeting", END)
+    builder.add_edge("math", END)
+    builder.add_edge("default", END)
+
+    # Compile the graph
+    return builder.compile()
+
+
+def main():
+    """Run the basic LangGraph agent."""
+    graph = create_graph()
+
+    print("=== Basic LangGraph Agent ===")
+    print("Try messages like: 'hello', 'add 5 and 3', or anything else")
+    print()
+
+    # Test cases
+    test_messages = [
+        "Hello there!",
+        "Can you add 5 and 3?",
+        "What's the weather like?"
+    ]
+
+    for message in test_messages:
+        print(f"Input: {message}")
+        initial_state = {"messages": [message], "counter": 0}
+        result = graph.invoke(initial_state)
+        print(f"Output: Processed with counter = {result['counter']}")
+        print()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/chatopenai_structured.py b/chatopenai_structured.py
new file mode 100644
index 0000000..e4ff65f
--- /dev/null
+++ b/chatopenai_structured.py
@@ -0,0 +1,231 @@
+"""
+LangGraph with ChatOpenAI and Structured Output Example.
+
+This demonstrates using LangGraph with ChatOpenAI's with_structured_output feature
+for type-safe structured JSON responses.
+"""
+
+from typing import TypedDict, Literal, List
+from langgraph.graph import StateGraph, START, END
+from langchain_openai import ChatOpenAI
+from pydantic import BaseModel, Field
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+# ============ Helper to create model ============
+
+def create_chat_model(model: str = "glm-4.5-air", temperature: float = 0.7):
+    """Create a ChatOpenAI model instance."""
+    return ChatOpenAI(
+        model=model,
+        temperature=temperature,
+        max_retries=2,
+        api_key=os.getenv("OPENAI_API_KEY"),
+        base_url=os.getenv("OPENAI_BASE_URL", "https://open.bigmodel.cn/api/paas/v4"),
+    )
+
+
+# ============ Structured Output Schemas ============
+
+class SentimentAnalysis(BaseModel):
+    """Schema for sentiment analysis output."""
+    sentiment: Literal["positive", "negative", "neutral"] = Field(
+        description="Overall sentiment of the text"
+    )
+    confidence: float = Field(
+        description="Confidence score between 0 and 1",
+        ge=0,
+        le=1
+    )
+    emotions: List[str] = Field(
+        description="List of detected emotions"
+    )
+    keywords: List[str] = Field(
+        description="Key words or phrases extracted"
+    )
+    analysis: str = Field(
+        description="Detailed analysis explanation"
+    )
+
+
+class EntityExtraction(BaseModel):
+    """Schema for entity extraction output."""
+    persons: List[str] = Field(default=[], description="Person names mentioned")
+    locations: List[str] = Field(default=[], description="Locations mentioned")
+    organizations: List[str] = Field(default=[], description="Organizations mentioned")
+    dates: List[str] = Field(default=[], description="Dates or time references")
+
+
+class CombinedAnalysis(BaseModel):
+    """Combined analysis result."""
+    sentiment: SentimentAnalysis
+    entities: EntityExtraction
+    summary: str = Field(description="Brief summary of the input text")
+
+
+# ============ State Schema ============
+
+class State(TypedDict):
+    """State flows through the graph."""
+    input_text: str
+    sentiment_result: SentimentAnalysis | None
+    entity_result: EntityExtraction | None
+    final_summary: str
+
+
+# ============ Node Functions ============
+
+def analyze_sentiment(state: State) -> dict:
+    """Analyze sentiment using with_structured_output."""
+    print(f"Analyzing sentiment for: '{state['input_text']}'...")
+
+    # Create model with structured output
+    model = create_chat_model().with_structured_output(SentimentAnalysis)
+
+    prompt = f"""
+    你是一个情感分析专家。请分析用户输入的情感。
+
+    输入文本：{state['input_text']}
+
+    请按照指定的 JSON Schema 返回分析结果，包括 sentiment, confidence, emotions, keywords 和 analysis。
+    """
+
+    result: SentimentAnalysis = model.invoke(prompt)
+
+    print(f"  Sentiment: {result.sentiment}")
+    print(f"  Confidence: {result.confidence:.0%}")
+    print(f"  Emotions: {', '.join(result.emotions)}")
+
+    return {"sentiment_result": result}
+
+
+def extract_entities(state: State) -> dict:
+    """Extract entities using with_structured_output."""
+    print("Extracting entities...")
+
+    # Create model with structured output
+    model = create_chat_model().with_structured_output(EntityExtraction)
+
+    prompt = f"""
+    你是一个命名实体识别专家。请从文本中提取实体。
+
+    输入文本：{state['input_text']}
+
+    请提取 persons, locations, organizations, dates 等实体信息。
+    """
+
+    result: EntityExtraction = model.invoke(prompt)
+
+    print(f"  Persons: {result.persons}")
+    print(f"  Locations: {result.locations}")
+    print(f"  Organizations: {result.organizations}")
+
+    return {"entity_result": result}
+
+
+def generate_summary(state: State) -> dict:
+    """Generate a combined summary using with_structured_output."""
+    print("Generating final summary...")
+
+    sentiment = state.get("sentiment_result")
+    entities = state.get("entity_result")
+
+    # Create model with structured output
+    model = create_chat_model().with_structured_output(CombinedAnalysis)
+
+    prompt = f"""
+    请对以下分析结果生成综合摘要：
+
+    输入文本：{state['input_text']}
+
+    情感分析结果：
+    - 情感：{sentiment.sentiment if sentiment else 'N/A'}
+    - 置信度：{sentiment.confidence if sentiment else 'N/A'}
+    - 情绪：{sentiment.emotions if sentiment else 'N/A'}
+    - 关键词：{sentiment.keywords if sentiment else 'N/A'}
+
+    实体提取结果：
+    - 人名：{entities.persons if entities else 'N/A'}
+    - 地点：{entities.locations if entities else 'N/A'}
+    - 组织：{entities.organizations if entities else 'N/A'}
+    - 日期：{entities.dates if entities else 'N/A'}
+
+    请生成一个简短的摘要。
+    """
+
+    result: CombinedAnalysis = model.invoke(prompt)
+
+    return {"final_summary": result.summary}
+
+
+# ============ Build Graph ============
+
+def create_analysis_graph():
+    """Create and compile the analysis workflow graph."""
+
+    builder = StateGraph(State)
+
+    # Add nodes
+    builder.add_node("sentiment", analyze_sentiment)
+    builder.add_node("entities", extract_entities)
+    builder.add_node("summarize", generate_summary)
+
+    # Add edges
+    builder.add_edge(START, "sentiment")
+    builder.add_edge("sentiment", "entities")
+    builder.add_edge("entities", "summarize")
+    builder.add_edge("summarize", END)
+
+    return builder.compile()
+
+
+# ============ Main ============
+
+def main():
+    """Run the structured analysis agent."""
+    graph = create_analysis_graph()
+
+    print("=== LangGraph with ChatOpenAI Structured Output ===\n")
+
+    # Test with a sample text
+    test_text = "今天天气真好，我和朋友们一起去公园野餐，大家都很开心！"
+
+    print(f"Input text: {test_text}\n")
+    print("-" * 50)
+
+    # Run the graph
+    initial_state = {
+        "input_text": test_text,
+        "sentiment_result": None,
+        "entity_result": None,
+        "final_summary": ""
+    }
+
+    result = graph.invoke(initial_state)
+
+    print("-" * 50)
+    print("\n=== Final Summary ===")
+    print(result["final_summary"])
+
+    # Access structured pydantic models
+    print("\n=== Structured Sentiment Data ===")
+    sentiment: SentimentAnalysis = result["sentiment_result"]
+    print(f"  sentiment: {sentiment.sentiment}")
+    print(f"  confidence: {sentiment.confidence}")
+    print(f"  emotions: {sentiment.emotions}")
+    print(f"  keywords: {sentiment.keywords}")
+    print(f"  analysis: {sentiment.analysis}")
+
+    print("\n=== Structured Entity Data ===")
+    entities: EntityExtraction = result["entity_result"]
+    print(f"  persons: {entities.persons}")
+    print(f"  locations: {entities.locations}")
+    print(f"  organizations: {entities.organizations}")
+    print(f"  dates: {entities.dates}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/langgraph_chatopenai.py b/langgraph_chatopenai.py
new file mode 100644
index 0000000..5cdce07
--- /dev/null
+++ b/langgraph_chatopenai.py
@@ -0,0 +1,198 @@
+"""
+LangGraph Example with ChatOpenAI.
+
+A multi-turn conversational agent with memory and tool routing.
+"""
+
+from typing import TypedDict, List, Literal
+from langgraph.graph import StateGraph, START, END
+from langchain_openai import ChatOpenAI
+from pydantic import BaseModel, Field
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+# ============ Model Helper ============
+
+def create_chat_model(model: str = "openai/gpt-5.4", temperature: float = 0.7):
+    """Create a ChatOpenAI model instance."""
+    return ChatOpenAI(
+        model=model,
+        temperature=temperature,
+        max_retries=2,
+        api_key=os.getenv("OPENAI_API_KEY"),
+        base_url=os.getenv("OPENAI_BASE_URL", "https://api.qnaigc.com/v1"),
+    )
+
+
+# ============ State Schema ============
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+
+class State(TypedDict):
+    messages: List[Message]
+    category: str
+
+
+# ============ Router Function ============
+
+def categorize_message(state: State) -> Literal["greeting", "question", "fallback"]:
+    """Use LLM to categorize the user's message."""
+
+    model = create_chat_model(temperature=0.1)
+
+    last_message = state["messages"][-1].content if state["messages"] else ""
+
+    prompt = f"""
+    Categorize this message into ONE category: greeting, question, or fallback.
+    Return ONLY the category name.
+
+    Message: {last_message}
+    """
+
+    response = model.invoke(prompt).content.strip().lower()
+
+    if "greeting" in response:
+        return "greeting"
+    elif "question" in response:
+        return "question"
+    else:
+        return "fallback"
+
+
+# ============ Node Functions ============
+
+def handle_greeting(state: State) -> dict:
+    """Handle greeting messages."""
+    print("  [Node: Greeting]")
+
+    model = create_chat_model()
+
+    last_message = state["messages"][-1].content
+
+    prompt = f"""
+    Respond to this greeting in a friendly way. Keep it brief.
+
+    User: {last_message}
+    Assistant:
+    """
+
+    response = model.invoke(prompt).content
+
+    return {
+        "messages": state["messages"] + [Message(role="assistant", content=response)]
+    }
+
+
+def handle_question(state: State) -> dict:
+    """Handle question messages."""
+    print("  [Node: Question]")
+
+    model = create_chat_model()
+
+    last_message = state["messages"][-1].content
+
+    prompt = f"""
+    Answer this question helpfully and concisely.
+
+    User: {last_message}
+    Assistant:
+    """
+
+    response = model.invoke(prompt).content
+
+    return {
+        "messages": state["messages"] + [Message(role="assistant", content=response)]
+    }
+
+
+def handle_fallback(state: State) -> dict:
+    """Handle unrecognized messages."""
+    print("  [Node: Fallback]")
+
+    model = create_chat_model()
+
+    last_message = state["messages"][-1].content
+
+    prompt = f"""
+    Respond to this message in a helpful way.
+
+    User: {last_message}
+    Assistant:
+    """
+
+    response = model.invoke(prompt).content
+
+    return {
+        "messages": state["messages"] + [Message(role="assistant", content=response)]
+    }
+
+
+# ============ Build Graph ============
+
+def create_chat_graph():
+    """Create and compile the chat graph."""
+
+    builder = StateGraph(State)
+
+    # Add nodes
+    builder.add_node("greeting", handle_greeting)
+    builder.add_node("question", handle_question)
+    builder.add_node("fallback", handle_fallback)
+
+    # Add conditional routing
+    builder.add_conditional_edges(
+        START,
+        categorize_message,
+        {
+            "greeting": "greeting",
+            "question": "question",
+            "fallback": "fallback"
+        }
+    )
+
+    # All paths lead to END
+    builder.add_edge("greeting", END)
+    builder.add_edge("question", END)
+    builder.add_edge("fallback", END)
+
+    return builder.compile()
+
+
+# ============ Main ============
+
+def main():
+    """Run the chat agent."""
+    graph = create_chat_graph()
+
+    print("=== LangGraph Chat Agent with ChatOpenAI ===\n")
+
+    # Test messages
+    test_inputs = [
+        "你好，很高兴见到你！",
+        "Python 中如何读取 JSON 文件？",
+        "随便聊聊吧"
+    ]
+
+    for user_input in test_inputs:
+        print(f"User: {user_input}")
+
+        initial_state = {
+            "messages": [Message(role="user", content=user_input)],
+            "category": ""
+        }
+
+        result = graph.invoke(initial_state)
+
+        assistant_message = result["messages"][-1].content
+        print(f"Assistant: {assistant_message}\n")
+        print("-" * 50 + "\n")
+
+
+if __name__ == "__main__":
+    main()