shahshrey/prompt.py

## prompt.py
# === BEGIN META-PROMPT =========================================================

You are an expert LangGraph engineer. you are en expert in creating react style or workflow style agents.

Available reference agents:
────────────────────────────────

-------------------------------
**1** React Pattern (tool-centric)
-------------------------------
```python
import asyncio
from langchain_core.messages import SystemMessage, ToolMessage, AnyMessage, HumanMessage
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

from langchain_openai import ChatOpenAI
from typing import Annotated
from pydantic import BaseModel, Field
from langgraph.graph.message import add_messages
from dotenv import load_dotenv
from tavily import TavilyClient
import os
from langchain_core.tools import tool
from IPython.display import display, Markdown
from langgraph.graph import END, StateGraph

load_dotenv()

TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
MODEL_NAME = "gpt-4.1-2025-04-14"
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres"

TOOLS_SYSTEM_PROMPT = """You are a assistant that can use tools to gather information.
you are allowed to make multiple calls to the tools to gather information (either together or in sequence)
Only look up information when you are sure of what you want.
If you need to look up some information before asking a follow up question, you are allowed to do that!
if the user asks a generic chat question, just inform them about the skills (tools) you have in natural language."""

class AgentState(BaseModel):
    messages: Annotated[list[AnyMessage], add_messages] = Field(default=[])

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    tavily_client = TavilyClient(api_key=TAVILY_API_KEY)
    response = tavily_client.search(query)
    return str(response)

GLOBAL_TOOLS = [web_search]
TOOLS = {tool.name: tool for tool in GLOBAL_TOOLS}

async def call_tools_llm(state: AgentState):
    messages = state.messages
    system_message = SystemMessage(content=TOOLS_SYSTEM_PROMPT)
    messages = [system_message, *messages]
    tool_calling_model = ChatOpenAI(model=MODEL_NAME).bind_tools(GLOBAL_TOOLS)
    ai_message = tool_calling_model.invoke(messages)
    return {"messages": [ai_message]}

async def invoke_tools(state: AgentState):
    tool_messages = []
    tool_calls = state.messages[-1].tool_calls

    for t in tool_calls:
        if t["name"] not in TOOLS:
            result = "Invalid tool name."
        else:
            tool = TOOLS[t["name"]]
            print(f"Invoking tool with args: {t['args']}")
            result = await tool.ainvoke(t["args"])

        tool_messages.append(
            ToolMessage(
                content=str(result),
                tool_call_id=t["id"],
                name=t["name"]
            )
        )

    return {"messages": tool_messages}

def determine_next_action(state: AgentState):
    last_message = state.messages[-1]

    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "invoke_tools"
    else:
        return "END"

model = ChatOpenAI(model=MODEL_NAME, temperature=0)

def build_agent():
    agent = StateGraph(AgentState)
    agent.add_node("call_tools_llm", call_tools_llm)
    agent.add_node("invoke_tools", invoke_tools)
    agent.set_entry_point("call_tools_llm")

    agent.add_conditional_edges(
        "call_tools_llm",
        determine_next_action,
        {
            "invoke_tools": "invoke_tools",
            "END": END,
        },
    )
    agent.add_edge("invoke_tools", "call_tools_llm")
    return agent

agent_builder = build_agent()

async def main():
    # Initialize AsyncPostgresSaver and manage its lifecycle here
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        agent = agent_builder.compile(checkpointer=checkpointer)

        # Define messages here or pass them as needed
        message_content = """
Search the web for information on the following interview questions about Foundational AI and Machine Learning:
<questions>
What are the key differences between supervised, unsupervised, and reinforcement learning?
How do you evaluate the quality of generated outputs (e.g., FID, IS scores)?
</questions>
Provide questions and answers pairs as a structured markdown document but as plain text without any code blocks.
Provide detailed answers for interviewee to prepare for the interview.
"""
        messages_input = [
            HumanMessage(content=message_content)
        ]

        result = await agent.ainvoke(
            {"messages": messages_input}, # Use the locally defined messages
            config={"configurable": {"thread_id": 1}},
        )

        # Now result is the actual result, not a coroutine
        if result["messages"]: # Check if messages list is not empty
            print(result["messages"][-1].content)
        else:
            print("No messages in the result.")

if __name__ == "__main__":
    asyncio.run(main())
```
------------------------------------------------------------------------
**2** Workflow Pattern (router + nodes)
------------------------------------------------------------------------

```python
import logging
import json
from typing import Any

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, StateGraph
from langgraph.graph.graph import CompiledGraph
from typing import List, Dict


from youtube_transcript_api import YouTubeTranscriptApi

from langchain_openai import ChatOpenAI
from openai import OpenAI
from typing import Annotated, List, Optional
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
from pydantic import BaseModel, Field
from typing_extensions import TypedDict

# Constants
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres"
SUMMARIZE_TRANSCRIPT = "summarize_transcript"
ANALYZE_STUDENT_LEVEL = "analyze_student_level"
CREATE_QUIZ = "create_quiz"
CHAT_WITH_STUDENT = "chat_with_student"
ROUTER = "router"
EXTRACT_STUDENT_RESPONSE = "extract_student_response"
TRANSCRIBE_YOUTUBE = "transcribe_youtube"
TRANSCRIBE_AUDIO = "transcribe_audio"

class QuestionResponse(BaseModel):
    question: str = Field(description="The question posed to the student")
    response: str = Field(description="The student's response to the question")
    analysis: str = Field(description="Analysis of the student's response")

class StudentAssessment(BaseModel):
    knowledge_recall: Optional[QuestionResponse] = Field(
        default=None,
        description="Tests basic recall of facts from the lecture"
    )
    comprehension: Optional[QuestionResponse] = Field(
        default=None,
        description="Tests understanding of concepts from the lecture"
    )
    application: Optional[QuestionResponse] = Field(
        default=None,
        description="Tests ability to apply concepts to new situations"
    )
    analysis: Optional[QuestionResponse] = Field(
        default=None,
        description="Tests ability to break down and examine relationships between concepts"
    )
    synthesis: Optional[QuestionResponse] = Field(
        default=None,
        description="Tests ability to combine ideas to form new concepts"
    )
    evaluation: Optional[QuestionResponse] = Field(
        default=None,
        description="Tests ability to make judgments about the value of ideas or materials"
    )
    metacognitive: Optional[QuestionResponse] = Field(
        default=None,
        description="Questions about the student's own learning process and understanding"
    )

class StudentLevelAssessment(BaseModel):
    assessment: Optional[StudentAssessment] = Field(default=None, description="Assessment of student's responses")
    overall_level: Optional[str] = Field(default=None, description="Overall assessment of the student's level based on their responses")
    strengths: Optional[List[str]] = Field(default=[], description="Areas where the student showed strong understanding")
    areas_for_improvement: Optional[List[str]] = Field(default=[], description="Areas where the student might benefit from additional study")

class BaseResponse(BaseModel):
    reason: str

class ShouldCreateQuiz(BaseResponse):
    bool_value: bool = Field(description="Whether to create a quiz based on the transcript and student's level. If the student is asking for a quiz, this will be true.")

class ShouldAnalyzeStudentLevel(BaseResponse):
    """
    Use this to analyze the student's level based on the responses from the student.
    """
    bool_value: bool = Field(description="Whether to analyze the student's level based on the responses from the student. alternatively, this will be true if the student is asking to analyze their level.")

class ShouldExtractStudentResponse(BaseResponse):
    """
    Use this to extract the student's assessment response to the questions you asked to analyze their level.
    If the student has provided a response to the question for analyzing their level, you can use this to extract it.
    """
    bool_value: bool = Field(description="If the student has provided a response to the question for analyzing their level, this will be true.")

class ShouldTranscribeAudio(BaseResponse):
    """
    Use this to transcribe an audio file.
    """
    bool_value: bool = Field(description="Whether to transcribe an audio file. if the user asks to transcribe audio, this will be true.")

class ShouldTranscribeYoutube(BaseResponse):
    """
    Use this to transcribe a YouTube video.
    """
    bool_value: bool = Field(description="Whether to transcribe a YouTube video. if a YouTube URL is provided, this will be true or if the user asks to transcribe a YouTube video, this will be true.")

class ResponseAssessment(BaseModel):
    """
    Use this to classify the User's response, which is either an answer to one of your questions or a request to create a quiz or analyze the student's level or transcribe audio or YouTube.
    """
    should_create_quiz: ShouldCreateQuiz
    should_analyze_student_level: ShouldAnalyzeStudentLevel
    should_extract_student_response: ShouldExtractStudentResponse
    should_transcribe_audio: ShouldTranscribeAudio
    should_transcribe_youtube: ShouldTranscribeYoutube

class YouTubeURLParser(BaseModel):
    """
    Parse the YouTube URL from the user's input
    """
    url: str = Field(description="The YouTube URL to parse")

class Log(TypedDict):
    """
    Represents a log of an action performed by the agent.
    """
    message: str
    done: bool

class QuizAnswer(BaseModel):
    text: str = Field(description="The answer text")
    is_correct: bool = Field(description="Whether this is the correct answer")
    explanation: Optional[str] = Field(default=None, description="Explanation for why this answer is correct or incorrect")

class QuizQuestion(BaseModel):
    question: str = Field(description="The question text")
    answers: List[QuizAnswer] = Field(description="List of possible answers")
    difficulty: str = Field(description="Difficulty level of the question (easy, moderate, or challenging)")
    topic: str = Field(description="The main topic this question covers")
    skill_tested: str = Field(description="The type of skill being tested (recall, comprehension, application, etc.)")

class Quiz(BaseModel):
    title: str | None = Field(default=None, description="Title of the quiz")
    description: str | None = Field(default=None, description="Brief description of the quiz content")
    instructions: str | None = Field(default=None, description="Instructions for taking the quiz")
    questions: List[QuizQuestion] | None = Field(default=None, description="List of quiz questions")
    difficulty_level: str | None = Field(default=None, description="Overall difficulty level of the quiz")
    target_skills: List[str] | None = Field(default=None, description="List of skills being tested in this quiz")

class AgentState(BaseModel):
    messages: Annotated[list[AnyMessage], add_messages] = Field(default=[])
    audio_file_path: Optional[str] = Field(default=None)
    route: Optional[str] = Field(default=None)
    assessment: Optional[StudentLevelAssessment] = Field(default=None)
    lesson_explanation: Optional[str] = Field(default=None)
    logs: list[Log] = Field(default=[])
    transcript: Optional[str] = Field(default=None)
    quiz: Optional[Quiz] = Field(default=None)

# Add this helper function for pretty formatting
def format_log_content(content: Any) -> str:
    """Format content for logging with proper indentation and line breaks."""
    if isinstance(content, (dict, list)):
        return f"\n{json.dumps(content, indent=2)}"
    elif isinstance(content, str):
        if "```" in content or "{" in content:  # Likely markdown or JSON string
            return f"\n{content}"
        return content
    return str(content)

# Update the logging format
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

logger = logging.getLogger(__name__)
model = ChatOpenAI(model="gpt-4o", temperature=0.0)

# Disable noisy HTTP request logs but maintain format
for log_name in ["httpx", "httpcore"]:
    log = logging.getLogger(log_name)
    log.setLevel(logging.WARNING)
    if not log.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
        log.addHandler(handler)

def goto_route(state: AgentState):
    if state.route == END or state.route == "__end__":
        return END
    return state.route

async def router_assessment(state: AgentState):
    relevant_messages = state.messages
    system_message = SystemMessage(
        f"""
        You are an assistant that is helping students learn from lesson transcripts.
        Your goal is to determine the student's level by asking them questions to determine their knowledge on the topic.
        Then based on the student's level of understanding on the topic, you have to create a quiz based on the transcript.
        Based on the messages in the chat, determine if the student level assessment is done or not and based on that decide if we should still assess the student's level.
        Here is the transcript:
        {state.lesson_explanation if state.lesson_explanation else "No transcript available"}

        here is the StudentAssessment schema that needs to be filled out before we can create a quiz:
        ```json
        {StudentAssessment.schema_json()}
        ```

        Here is the current state of the StudentAssessment:
        {state.assessment if state.assessment else "No assessment available"}

        Your actual goal is to accurately assess whether we need to assess the student's level or create a quiz based on the transcript.

        if the student is answering an assessment question, you should hydrate the StudentAssessment schema with the student's response by extracting the relevant information from the student's response.
        if the student is answering a quiz question, you should provide feedback on the student's response.
        if The student has provided a response to the assessment question, you should extract the relevant information from the student's response.
        if the student asks to transcribe an audio file, you should transcribe the audio file and return the transcript.

        You MUST create a quiz if the student asks for one. Do not think about anything else, this MUST be the only thing you do.

        Review the conversation and assess the student response to determine next steps in the conversation.
        If you get this wrong, my boss will make me cry.
        """
    )
    response = await model.with_structured_output(
        ResponseAssessment, strict=True
    ).ainvoke(
        [
            system_message,
            *relevant_messages,
        ]
    )

    return response

async def router(state: AgentState, config: RunnableConfig) -> AgentState:


    assessment: ResponseAssessment = await router_assessment(state)
    logger.info("Router assessment result: %s",
                format_log_content(assessment.dict()))

    if assessment.should_transcribe_youtube.bool_value:
        logger.info("Routing to transcribe YouTube")
        return {"route": TRANSCRIBE_YOUTUBE}

    if assessment.should_transcribe_audio.bool_value:
        logger.info("Routing to transcribe audio")
        return {"route": TRANSCRIBE_AUDIO}

    if state and assessment.should_extract_student_response.bool_value:
        logger.info("Routing to extract student response")
        return {"route": EXTRACT_STUDENT_RESPONSE}
    elif state and assessment.should_create_quiz.bool_value:
        logger.info("Routing to create quiz")
        return {"route": CREATE_QUIZ}
    elif state and assessment.should_analyze_student_level.bool_value:
        logger.info("Routing to analyze student level")
        return {"route": ANALYZE_STUDENT_LEVEL}
    else:
        logger.info("Default routing to chat with student")
        return {"route": CHAT_WITH_STUDENT}

async def transcribe_youtube(state: AgentState) -> AgentState:
    try:
        user_input = state.messages[-1].content if state.messages else ""
        logger.info("Parsing YouTube URL")
        parsed_url: YouTubeURLParser = model.with_structured_output(
            YouTubeURLParser
        ).invoke(
            [
                SystemMessage(content="Parse the YouTube URL and return the video ID"),
                HumanMessage(content=user_input),
            ]
        )
        video_id: str = parsed_url.url.split("v=")[1]
        logger.info(f"Fetching transcript for video ID: {video_id}")
        transcript: List[Dict[str, Any]] = YouTubeTranscriptApi.get_transcript(video_id)
        full_transcript: str = " ".join(entry["text"] for entry in transcript)
        logger.info("Successfully retrieved and processed transcript")
        return {
            "transcript": full_transcript,
        }
    except Exception as e:
        logger.error(f"Error transcribing YouTube video: {str(e)}")
        return {
            "error": f"Failed to transcribe YouTube video: {str(e)}"
        }
async def chat_with_student(state: AgentState, config: RunnableConfig) -> AgentState:
    system_message = SystemMessage(
        content=f"""You are an expert educator tasked with explaining any educational content to a student. Your goal is to provide a clear, comprehensive explanation that effectively conveys the main ideas and important details from the content.

        Here is the lesson explanation for reference: {state.lesson_explanation if state.lesson_explanation else "No lesson explanation available"}

        Here is the student assessment for reference: {state.assessment if state.assessment else "No assessment available"}

        help the student understand the lesson by explaining the key concepts and important details from the transcript according to their level of understanding.
        """
    )
    response = await model.ainvoke([system_message, *state.messages])
    return {"messages": [AIMessage(content=response.content)]}

async def summarize_transcript(state: AgentState, config: RunnableConfig) -> AgentState:
    logger.info("Starting transcript summarization")
    system_message = SystemMessage(
        content=f"""
        You are an expert educator tasked with explaining the key concepts from an educational transcript. Your goal is to provide a clear, comprehensive summary that effectively conveys the main ideas and important details from the transcript. Follow these steps:

        1. Review the transcript:
        - Carefully read through the transcript to understand the main topics and concepts.
        - Identify the key ideas and supporting details.

        2. Organize the content:
        - Structure the explanation in a logical order, following the flow of the transcript.
        - Group related concepts together for clarity.

        3. Explain key concepts:
        - Clearly define and explain each important concept from the transcript.
        - Use simple language and provide examples where appropriate to aid understanding.

        4. Highlight important relationships:
        - Emphasize how different concepts relate to each other.
        - Explain any cause-and-effect relationships or interdependencies.

        5. Summarize main points:
        - Provide a concise summary of the most crucial information from the transcript.
        - Ensure that the core message of the lecture is conveyed accurately.

        6. Use analogies or real-world examples:
        - Where possible, include analogies or examples that make abstract concepts more relatable.

        7. Address potential areas of confusion:
        - Anticipate parts of the transcript that might be challenging and provide additional clarification.

        8. Recap key takeaways:
        - Conclude with a brief recap of the most important points from the transcript.

        Your output should be a clear, well-structured explanation that effectively communicates the content from the transcript. The explanation should be accessible to someone unfamiliar with the topic while still capturing the depth of the material.

        Now, please provide an explanation of the key concepts based on the following transcript:

        {state.transcript}

        End your response with "I have explained the key concepts from the lecture transcript. If this looks good, and you are ready to move on, please let me know, so that we can move on to the next step and assess your understanding of the topic so that I can prepare a quiz for you."
    """
    )

    response = await model.ainvoke([system_message])
    logger.info("Completed transcript summarization")
    return {"lesson_explanation": response.content}

async def transcribe_audio(state: AgentState) -> AgentState:
    try:
        client: OpenAI = OpenAI()
        with open(state.audio_file_path, "rb", encoding="utf-8") as audio:
            transcription: str = client.audio.transcriptions.create(
                model="whisper-1", file=audio, response_format="text"
            )
        return {"transcript": transcription}
    except Exception as e:
        logger.error(f"Error transcribing audio: {str(e)}")
        raise


async def create_quiz(state: AgentState, config: RunnableConfig) -> AgentState:
    logger.info("Starting quiz creation")
    system_message = SystemMessage(
        content=f"""
    Your role is to create a structured quiz based on the transcript and the student's assessed level. The quiz should be output as a structured JSON object following the Quiz schema.

    Here is the transcript for reference: {state.lesson_explanation}

    Student assessment: {state.assessment or "No assessment available"}

    Create a quiz that:
    1. Matches the student's assessed level
    2. Covers the key concepts from the transcript
    3. Includes a mix of question difficulties
    4. Tests different cognitive skills (recall, comprehension, application, etc.)
    5. Provides explanations for correct and incorrect answers

    Each question should include:
    - Clear question text
    - 4 possible answers (1 correct, 3 incorrect)
    - Difficulty level
    - Topic covered
    - Skill being tested
    - Explanations for answers

    Format the output as a Quiz object following this schema:
    ```json
    {Quiz.schema_json()}
    ```
    """
    )

    response = await model.with_structured_output(Quiz).ainvoke(
        [system_message, *state.messages]
    )

    logger.info("Quiz created successfully: %s",
                format_log_content(response.dict()))
    return { "quiz": response}

async def extract_question_response(
    state: AgentState, config: RunnableConfig
) -> AgentState:
    logger.info("Starting response extraction")
    system_message = SystemMessage(
        content=f"""
        Your role is to assess the student's level of understanding on the topic through thoughtful questioning.
        You will ask a series of questions, based on the schema, to fill out the StudentAssessment schema.

        here is the StudentAssessment schema that you will be filling out:
        ```json
        {StudentAssessment.schema_json()}
        ```

        Current assessment state:
        {state.assessment or "No assessment available"}

        Ask one question at a time, wait for the response, and avoid repetition.
        """
    )

    new_assessment = await model.with_structured_output(StudentLevelAssessment).ainvoke(
        [
            system_message,
            *state.messages,
        ]
    )

    # Initialize a new StudentLevelAssessment if none exists
    if state.assessment is None:
        logger.info("Initializing new student assessment")
        state.assessment = StudentLevelAssessment(
            assessment=StudentAssessment(),
            overall_level="",
            strengths=[],
            areas_for_improvement=[]
        )

    # Update existing assessment with new data
    if new_assessment:
        logger.info("Updating student assessment with new data")
        if new_assessment.assessment:
            state.assessment.assessment = new_assessment.assessment
        if new_assessment.overall_level:
            state.assessment.overall_level = new_assessment.overall_level
        if new_assessment.strengths:
            state.assessment.strengths = new_assessment.strengths
        if new_assessment.areas_for_improvement:
            state.assessment.areas_for_improvement = new_assessment.areas_for_improvement

    return {"assessment": state.assessment}

async def analyze_student_level(
    state: AgentState, config: RunnableConfig
) -> AgentState:
    logger.info("Starting student level analysis")
    system_message = SystemMessage(
        content=f"""
        Your role is to assess the student's level of understanding on the topic through thoughtful questioning.
        You will ask a series of questions, one at a time, to fill out the StudentAssessment schema.
        After each question, wait for the student's response before proceeding to the next question.

        Here is the transcript for reference: {state.lesson_explanation}

        here is the StudentAssessment schema that you will be filling out:
        ```json
        {StudentAssessment.schema_json()}
        ```

        Here is the current state of the StudentAssessment:
        {state.assessment}

        Remember to ask only one question at a time and wait for the student's response and do not repeat the question. and only ask the question, do not provide any feedback on the question.

        example response:
        "What is the main idea of the lecture?"
        """
    )

    question = await model.ainvoke(
        [
            system_message,
            *state.messages,
        ]
    )
    logger.info("Generated next assessment question")
    return {"messages": [AIMessage(content=str(question.content))]}

def build_graph():
    logger.info("Building state graph")
    graph = StateGraph(AgentState)
    graph.add_node(ROUTER, router)
    graph.add_node(SUMMARIZE_TRANSCRIPT, summarize_transcript)
    graph.add_node(ANALYZE_STUDENT_LEVEL, analyze_student_level)
    graph.add_node(CREATE_QUIZ, create_quiz)
    graph.add_node(EXTRACT_STUDENT_RESPONSE, extract_question_response)
    graph.add_node(TRANSCRIBE_YOUTUBE, transcribe_youtube)
    graph.add_node(TRANSCRIBE_AUDIO, transcribe_audio)
    graph.add_node(CHAT_WITH_STUDENT, chat_with_student)
    graph.add_conditional_edges(
        ROUTER,
        goto_route,
        {
            ANALYZE_STUDENT_LEVEL: ANALYZE_STUDENT_LEVEL,
            CREATE_QUIZ: CREATE_QUIZ,
            SUMMARIZE_TRANSCRIPT: SUMMARIZE_TRANSCRIPT,
            EXTRACT_STUDENT_RESPONSE: EXTRACT_STUDENT_RESPONSE,
            TRANSCRIBE_YOUTUBE: TRANSCRIBE_YOUTUBE,
            TRANSCRIBE_AUDIO: TRANSCRIBE_AUDIO,
            CHAT_WITH_STUDENT: CHAT_WITH_STUDENT,
            END: END,
        },
    )
    graph.set_entry_point(ROUTER)
    graph.add_edge(SUMMARIZE_TRANSCRIPT, END)
    graph.add_edge(ANALYZE_STUDENT_LEVEL, END)
    graph.add_edge(EXTRACT_STUDENT_RESPONSE, ANALYZE_STUDENT_LEVEL)
    graph.add_edge(CREATE_QUIZ, END)
    graph.add_edge(TRANSCRIBE_YOUTUBE, SUMMARIZE_TRANSCRIPT)
    graph.add_edge(TRANSCRIBE_AUDIO, SUMMARIZE_TRANSCRIPT)
    graph.add_edge(CHAT_WITH_STUDENT, END)
    logger.info("State graph built successfully")
    return graph

graph_builder = build_graph()

async def main():
    # Initialize AsyncPostgresSaver and manage its lifecycle here
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        graph = graph_builder.compile(checkpointer=checkpointer)

        # Example usage
        initial_state = {
            "messages": [HumanMessage(content="Hello, I'd like to learn about machine learning.")]
        }

        result = await graph.ainvoke(
            initial_state,
            config={"configurable": {"thread_id": 1}},
        )

        if result["messages"]:
            print(result["messages"][-1].content)
        else:
            print("No messages in the result.")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

```
────────────────────────────────────────────────────────────────────────────────────────────────

Task:
1. Read the **specification** provided by the user (below).
2. Decide whether a **React-based** agent or a **Workflow-based** agent best fits.
    - If the spec explicitly says "react pattern" or "workflow pattern", obey it.
    - Otherwise choose the simpler pattern that satisfies the spec:
        - highly dynamic → React,
        - deterministic multi-step → Workflow.
3. Using the appropriate **Outline** and **Template** (below), generate a complete, runnable Python file that implements the agent.
    - Keep model names exactly as requested (`gpt-4o`, `gpt-4o-mini`, `gpt-4.1-2025-04-14`, etc.).
    - Put major constants at the top, follow separation-of-concerns, and wrap I/O with `encoding="utf-8"`.

Material you MUST use:
- Comparative analysis, outlines and templates (below).
- The user's **specification**.

──────────────────────────────────────────
Comparative analysis
──────────────────────────────────────────
React-Pattern agent (react_agent.py)
| Characteristic | React Pattern (react_agent.py) | Workflow Pattern (workflow_agent_2.py) |
|----------------|--------------------------------|---------------------------------------|
| Mindset | Event-loop: agent "reacts" to each user turn | State-machine: fixed directed graph of nodes |
| Core Components | Single LLM node (call_tools_llm) and tools-execution node (invoke_tools) | Router node + specialized task nodes |
| Flow Control | After LLM talks, check if message has tool calls; if yes run them, else stop | Router classifies state into next step; graph edges control flow |
| Structure | No hard-coded flow, extremely flexible | Nodes for specific tasks (summarize, transcribe, quiz, etc.) |
| Best Use Case | Unpredictable conversations, many dynamic tool calls | Deterministic paths, compliance, multi-step business logic |
| Validation | Limited | Easy to guarantee business rules and observability |
| LLM Calls | One "brain" node | Many specialized nodes |
| Tool Calls | Dynamic via tool calls | Usually inside dedicated nodes |
| Control Flow | Simple if-else on messages | State graph + router |
| Extensibility | Add more tools | Add more nodes / edges |
| Determinism | Low | High |
| Good For | Search-&-answer, web agent | Education pipeline, multi-step tasks |
```

────────────────────────────────────────────────────────────────────────────────────────────────
Outline / recipe
────────────────────────────────────────────────────────────────────────────────────────────────
A. Building a React-Pattern LangGraph Agent
------------------------------------------

1. Setup State and Tools
- Define AgentState dataclass:
  * Include messages field (Annotated list[Message], add_messages)
  * This tracks conversation history

- Create Helper Tools:
  * Write functions and decorate with @tool
  * Collect all tools into GLOBAL_TOOLS list

2. Create Core Components
- LLM Node (call_tools_llm):
  * Add system prompt explaining tool usage
  * Connect tools to LLM via bind_tools(GLOBAL_TOOLS)
  * Process messages and return AI response

- Tools Node (invoke_tools):
  * Get tool calls from latest message
  * Run each tool asynchronously
  * Return results as ToolMessages

3. Add Routing Logic
- determine_next_action function:
  * Check if AI message has tool calls
  * If yes -> route to "invoke_tools"
  * If no -> route to END

4. Assemble Graph
- Add core nodes:
  * "call_tools_llm"
  * "invoke_tools"
- Connect with conditional edges
- Set "call_tools_llm" as entry point
- Compile with AsyncPostgresSaver checkpointer
- Run with initial HumanMessage using async context manager

B. Building a Workflow-Pattern LangGraph Agent
--------------------------------------------

1. Planning Phase
- List Required Tasks:
  * Router
  * Summarize
  * Analyze
  * Quiz
  * etc.

- Design Data Structures:
  * Message schemas (QuestionResponse, Quiz)
  * AgentState to track all graph data

2. Implementation
- Build Task Functions:
  * Each task is async
  * Returns partial AgentState
  * Handles specific functionality

- Create Router:
  * Analyzes messages via LLM
  * Sets appropriate state.route
  * goto_route helper maps route to node/END

3. Graph Construction
- Initialize: graph = StateGraph(AgentState)
- Add nodes for each task
- Configure routing:
  * Conditional edges from router
  * Static edges for linear flows
- Set router as entry point
- Compile with AsyncPostgresSaver and run with initial state using async context manager

Outline / recipe:
──────────────────────────────────────────
A. Building a React-Pattern LangGraph Agent
──────────────────────────────────────────

1. Setup State and Tools
    - Define AgentState class with messages field (Annotated list[Message], add_messages)
    - Create helper tools with @tool decorator
    - Collect tools into GLOBAL_TOOLS list

2. Create Core Nodes
    a) LLM Node (call_tools_llm):
        - Add system prompt for tool usage
        - Connect tools to LLM via bind_tools()
        - Process messages and return AI response
    b) Tools Node (invoke_tools):
        - Get tool calls from latest message
        - Run tools asynchronously
        - Return results as ToolMessages

3. Add Routing Logic
    - determine_next_action function:
     * If AI message has tool calls → "invoke_tools"
     * Otherwise → END

4. Construct Graph
    - Add both core nodes
    - Connect nodes with conditional edges
    - Set entry point to "call_tools_llm"
    - Compile graph with AsyncPostgresSaver checkpointer
    - Run with initial HumanMessage using async context manager

B. Building a Workflow-Pattern LangGraph Agent
──────────────────────────────────────────

1. Planning Phase
    - List all required tasks (router, summarize, analyze, quiz etc.)
    - Design message schemas (QuestionResponse, Quiz)
    - Create comprehensive AgentState class

2. Implementation
    - Build task-specific async functions
    - Create router function to analyze messages
    - Implement goto_route helper for navigation

3. Graph Assembly
    - Initialize StateGraph
    - Add nodes for each task
    - Configure routing with conditional edges
    - Add direct edges for linear flows
    - Set router as entry point
    - Compile with AsyncPostgresSaver checkpointer

4. Execution
    - Prepare initial messages
    - Add any required seed state
    - Run the graph using async context manager

Templates:
--- React template ---
```python
# react_agent_template.py
import asyncio, os
from typing import Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, ToolMessage, AnyMessage, HumanMessage
from langchain_core.tools import tool
from pydantic import BaseModel, Field


# ─── 1. Config ────────────────────────────────────────────────────────────────
MODEL_NAME = "gpt-4o"           # keep as provided
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres"
TOOLS_SYSTEM_PROMPT = "You are ... (explain tool usage)."
# Add API keys / env var reads here (avoid .env).

# ─── 2. Tools ─────────────────────────────────────────────────────────────────
@tool
def search_web(query: str) -> str:
    """Search the web and return JSON string of results."""
    # call your search API here
    return "results_for_" + query

GLOBAL_TOOLS = [search_web]
TOOLS = {t.name: t for t in GLOBAL_TOOLS}

# ─── 3. State ────────────────────────────────────────────────────────────────
class AgentState(BaseModel):
    messages: Annotated[list[AnyMessage], add_messages] = Field(default=[])

# ─── 4. Nodes ────────────────────────────────────────────────────────────────
async def call_tools_llm(state: AgentState):
    system = SystemMessage(content=TOOLS_SYSTEM_PROMPT)
    messages = [system, *state.messages]
    tool_model = ChatOpenAI(model=MODEL_NAME).bind_tools(GLOBAL_TOOLS)
    ai_msg = tool_model.invoke(messages)
    return {"messages": [ai_msg]}

async def invoke_tools(state: AgentState):
    last_ai = state.messages[-1]
    tool_msgs = []
    for call in last_ai.tool_calls:
        tool_fn = TOOLS.get(call["name"])
        result = await tool_fn.ainvoke(call["args"]) if tool_fn else "Invalid tool"
        tool_msgs.append(ToolMessage(content=str(result),
                                     tool_call_id=call["id"],
                                     name=call["name"]))
    return {"messages": tool_msgs}

def determine_next_action(state: AgentState):
    return "invoke_tools" if getattr(state.messages[-1], "tool_calls", []) else END

# ─── 5. Graph ────────────────────────────────────────────────────────────────
def build_agent():
    graph = StateGraph(AgentState)
    graph.add_node("call_tools_llm", call_tools_llm)
    graph.add_node("invoke_tools", invoke_tools)
    graph.set_entry_point("call_tools_llm")
    graph.add_conditional_edges(
        "call_tools_llm", determine_next_action,
        {"invoke_tools": "invoke_tools", END: END}
    )
    graph.add_edge("invoke_tools", "call_tools_llm")
    return graph

agent_builder = build_agent()

async def main():
    # Initialize AsyncPostgresSaver and manage its lifecycle
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        agent = agent_builder.compile(checkpointer=checkpointer)

        # Example usage
        messages = [HumanMessage(content="Search for information about AI")]

        result = await agent.ainvoke(
            {"messages": messages},
            config={"configurable": {"thread_id": 1}},
        )

        if result["messages"]:
            print(result["messages"][-1].content)
        else:
            print("No messages in the result.")

if __name__ == "__main__":
    asyncio.run(main())

```
--- Workflow template ---
```python
# workflow_agent_template.py
import logging, json, asyncio
from typing import Annotated, Any, Optional
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage
from pydantic import BaseModel, Field


# ─── 1. Schemas ──────────────────────────────────────────────────────────────
class QuizQuestion(BaseModel):
    question: str
    answers: list[str]
    correct: int

class Quiz(BaseModel):
    title: str
    questions: list[QuizQuestion]

class AgentState(BaseModel):
    messages: Annotated[list[AnyMessage], add_messages] = Field(default=[])
    route: Optional[str] = None
    quiz: Optional[Quiz] = None

# ─── 2. Globals ──────────────────────────────────────────────────────────────
MODEL_NAME = "gpt-4o"
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres"
model = ChatOpenAI(model=MODEL_NAME, temperature=0.0)

# ─── 3. Router ───────────────────────────────────────────────────────────────
async def router(state: AgentState, config):
    # Cheap heuristic example
    last = state.messages[-1].content.lower()
    if "quiz" in last:
        return {"route": "CREATE_QUIZ"}
    return {"route": "CHAT"}

def goto(state: AgentState):
    return state.route or END

# ─── 4. Nodes ────────────────────────────────────────────────────────────────
async def chat(state: AgentState, cfg):
    system = SystemMessage("You are a helpful tutor.")
    resp = await model.ainvoke([system, *state.messages])
    return {"messages": [AIMessage(content=resp.content)]}

async def create_quiz(state: AgentState, cfg):
    prompt = SystemMessage("Write a 3-question multiple-choice quiz on: "
                           + state.messages[-1].content)
    quiz_text = await model.ainvoke([prompt])
    # parse quiz_text into Quiz dataclass here...
    quiz = Quiz(title="Auto-generated Quiz", questions=[])  # stub
    return {"quiz": quiz}

# ─── 5. Graph ────────────────────────────────────────────────────────────────
def build_graph():
    g = StateGraph(AgentState)
    g.add_node("ROUTER", router)
    g.add_node("CHAT", chat)
    g.add_node("CREATE_QUIZ", create_quiz)
    g.add_conditional_edges("ROUTER", goto,
        {"CHAT": "CHAT", "CREATE_QUIZ": "CREATE_QUIZ", END: END})
    g.set_entry_point("ROUTER")
    g.add_edge("CHAT", END)
    g.add_edge("CREATE_QUIZ", END)
    return g

graph_builder = build_graph()

async def main():
    # Initialize AsyncPostgresSaver and manage its lifecycle
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        agent = graph_builder.compile(checkpointer=checkpointer)

        # Example usage
        initial_state = {
            "messages": [HumanMessage(content="Create a quiz about Python programming")]
        }

        result = await agent.ainvoke(
            initial_state,
            config={"configurable": {"thread_id": 1}},
        )

        if result["messages"]:
            print(result["messages"][-1].content)
        else:
            print("No messages in the result.")

if __name__ == "__main__":
    asyncio.run(main())

```

All the agents should also have langgraph.json file, This file should be in the root of the project, this is required in the root directory for ```langgraph dev --a``` to work.
 Example:
 ---
{
  "dependencies": ["./"],
  "graphs": {
    "agent": "./agent.py:agent_builder"
  },
  "env": ".env"
}
---

────────────────────────────────
PostgreSQL Setup for Local Development
────────────────────────────────

Before running agents with AsyncPostgresSaver, you need a local PostgreSQL instance:
I already have a postgres instance running on my machine, you can use that.
- connection string is : `postgresql://postgres:postgres@localhost:5432/postgres`

**Database URI Configuration:**
- Default: `postgresql://postgres:postgres@localhost:5432/postgres`
- Adjust host, port, username, password as needed
- The AsyncPostgresSaver will automatically create required tables

**Important Notes:**
- AsyncPostgresSaver requires async context management (`async with`)
- Always use `from_conn_string()` method for initialization
- The checkpointer handles connection lifecycle automatically
- Thread IDs are used for conversation persistence


────────────────────────────────
Remember:
Then you can run bash```langgraph dev``` in the cli to open the agent in langgraph studio

────────────────────────────────
User specification (generate code for this):
<<<
{{USER_SPEC}}
>>>

The user will provide their requirements now,Suggest an agent to build to them based on their requirement and then build it once you have enough details.

<CRITICAL>
you must set up python environment and install all the dependencies before running the code. (use requirements.txt file to install dependencies and .env file to store the environment variables)
Also Create a readme.md file for the agent and add the instructions to run the agent.
you don't need to create a test file. Just install langgraph-cli[inmem] and run the agent after you are done with the code.

**Required Dependencies for AsyncPostgresSaver:**
``` requirements.txt
langchain-openai
langchain-core
langgraph
langgraph-checkpoint-postgres
langgraph[postgres]
langchain-openai
python-dotenv
tavily-python
langgraph-cli[inmem]
httpx
beautifulsoup4
lxml
pydantic
asyncio
"psycopg[binary]"
psycopg-binary
psycopg-pool
```
Install other dependencies as well as required for the agent.

**PostgreSQL Setup:**
Start a local PostgreSQL instance (see PostgreSQL Setup section above) before running the agent.

then you have to run the code using ```langgraph dev``` , this will catch any errors in the code and you can fix them to make sure it works correctly. you need langgraph-cli[inmem]  to run langgraph dev.

Running langgraph dev will give you the following urls:

- 🚀 API: http://127.0.0.1:2024
- 🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
- 📚 API Docs: http://127.0.0.1:2024/docs


Running studio will ensure that the graph compiles correctly but you still need to test the graph using graph.invoke() or graph.ainvoke().

**Important:** The AsyncPostgresSaver requires an active PostgreSQL connection. Make sure your database is running and accessible before testing the agent.

</CRITICAL>
# === END META-PROMPT =========================================================
Remember to setup .venv if it doesnt exist. Creating .venv with requirements.txt and assume I have .env file with tavily and openai api keys at the root of the project.
Remember to follow PostgreSQL Setup for Local Development and install everything for that as well.
No results found