Deep Research con LangGraph: Crea un Asistente de IA para Investigar Automáticamente

Para realizar el asistente vamos a usar la siguiente arquitectura:

	

		
!uv init deepresearcher
	
	

		
			
		
	
	
Copied

	

		
error: Project is already initialized in `/Users/macm1/Documents/web/portafolio/posts/deepresearcher` (`pyproject.toml` file exists)
	

	

		
!cd deepresearcher && uv sync
	
	

		
			
		
	
	
Copied

	

		
Resolved 88 packages in 9ms
Audited 84 packages in 0.38ms
	

	

		
!cd deepresearcher && source .venv/bin/activate
	
	

		
			
		
	
	
Copied

	

		
!cd deepresearcher && source .venv/bin/activate && uv add "alive-progress>=3.3.0" "dotenv>=0.9.9" "grandalf>=0.8" "ipykernel>=6.30.1" "langchain>=0.3.27" "langchain-anthropic>=0.3.21" "langchain-community>=0.3.29" "langchain-mcp-adapters>=0.1.11" "langchain-openai>=0.3.33" "langgraph>=0.6.7" "rich>=14.1.0" "tavily-python>=0.7.12"
	
	

		
			
		
	
	
Copied

	

		
Resolved 88 packages in 0.59ms
Audited 84 packages in 0.01ms
	

	

		
from pathlib import Path
 
def get_current_dir() -> Path:
    """Get the current directory of the module.
 
    This function is compatible with Jupyter notebooks and regular Python scripts.
 
    Returns:
        Path object representing the current directory
    """
    try:
        return Path(__file__).resolve().parent
    except NameError: # __file__ is not defined
        return Path.cwd()
	
	

		
			
		
	
	
Copied

	

		
from datetime import datetime
 
def get_today_str() -> str:
    """Get current date in a human-readable format."""
    return datetime.now().strftime("%a %b %-d, %Y")
	
	

		
			
		
	
	
Copied

	

		
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
import json
import ast
 
console = Console()
 
def format_tool_outputs(tool_outputs_list: list) -> str:
    """Format a list of ToolMessage objects into a readable string"""
    formatted_parts = []
    
    for i, output in enumerate(tool_outputs_list, 1):
        # Handle ToolMessage objects that have been converted to strings
        if isinstance(output, dict):
            formatted_parts.append(f"
  Tool Output {i}:")
            formatted_parts.append(f"    Name: {output.get('name', 'N/A')}")
            formatted_parts.append(f"    Tool Call ID: {output.get('tool_call_id', 'N/A')}")
            
            # Format the content (which contains the actual output)
            content = output.get('content', 'N/A')
            if isinstance(content, str):
                # Truncate very long outputs
                if len(content) > 500:
                    content_preview = content[:500] + f"... ({len(content)} chars total)"
                else:
                    content_preview = content
                formatted_parts.append(f"    Content:")
                # Indent each line of content
                for line in content_preview.split('
'):
                    formatted_parts.append(f"      {line}")
            else:
                formatted_parts.append(f"    Content: {content}")
        else:
            formatted_parts.append(f"
  Tool Output {i}: {output}")
    
    return "
".join(formatted_parts)
 
def format_string_with_structures(content: str) -> str:
    """
    Detect and format structured data (lists, dicts) within a string.
    Handles cases like 'Tool calls: [...]' and 'Tool outputs: [...]' by parsing and formatting the structure.
    """
    # Check if the string contains 'Tool outputs:' followed by a list
    if 'Tool outputs:' in content:
        try:
            # Find the position where the tool outputs list starts
            tool_outputs_index = content.index('Tool outputs:')
            prefix = content[:tool_outputs_index + len('Tool outputs:')]
            
            # Extract the part that should be a list
            potential_list = content[tool_outputs_index + len('Tool outputs:'):].strip()
            
            # Try to parse it as a Python literal (list/dict)
            parsed_data = ast.literal_eval(potential_list)
            
            # If it's a list of tool outputs, format them nicely
            if isinstance(parsed_data, list):
                formatted_parts = [prefix + "
"]
                formatted_parts.append(format_tool_outputs(parsed_data))
                return "
".join(formatted_parts)
        except (ValueError, SyntaxError):
            # If parsing fails, continue to next check
            pass
    
    # Check if the string contains 'Tool calls:' followed by a list
    if 'Tool calls:' in content:
        try:
            # Find the position where the tool calls list starts
            tool_calls_index = content.index('Tool calls:')
            prefix = content[:tool_calls_index + len('Tool calls:')]
            
            # Extract the part that should be a list
            potential_list = content[tool_calls_index + len('Tool calls:'):].strip()
            
            # Try to parse it as a Python literal (list/dict)
            parsed_data = ast.literal_eval(potential_list)
            
            # If it's a list of tool calls, format them nicely
            if isinstance(parsed_data, list):
                formatted_parts = [prefix + "
"]
                for i, tool_call in enumerate(parsed_data, 1):
                    if isinstance(tool_call, dict):
                        formatted_parts.append(f"
  Tool Call {i}:")
                        formatted_parts.append(f"    Name: {tool_call.get('name', 'N/A')}")
                        formatted_parts.append(f"    ID: {tool_call.get('id', 'N/A')}")
                        formatted_parts.append(f"    Type: {tool_call.get('type', 'N/A')}")
                        if 'args' in tool_call:
                            formatted_parts.append(f"    Args:")
                            formatted_parts.append(f"       {json.dumps(tool_call['args'], indent=7, ensure_ascii=False)}")
                return "
".join(formatted_parts)
        except (ValueError, SyntaxError):
            # If parsing fails, return the original content
            pass
    
    # Try to detect if the entire content is a JSON/dict/list
    try:
        # Try JSON first
        parsed = json.loads(content)
        return json.dumps(parsed, indent=2, ensure_ascii=False)
    except (json.JSONDecodeError, TypeError):
        try:
            # Try Python literal (ast.literal_eval)
            parsed = ast.literal_eval(content)
            return json.dumps(parsed, indent=2, ensure_ascii=False)
        except (ValueError, SyntaxError):
            # If all parsing attempts fail, return original content
            return content
 
def format_message_content(message):
    """Convert message content to displayable string"""
    parts = []
    tool_calls_processed = False
 
    # Handle ToolCall objects (special case - they don't have 'content')
    if isinstance(message, dict) and 'args' in message and 'name' in message:
        # This is a ToolCall dictionary
        parts.append(f"Tool Call Name: {message.get('name', 'N/A')}")
        parts.append(f"ID: {message.get('id', 'N/A')}")
        
        # Format the args (which should be the tool calls list)
        args = message.get('args', [])
        if isinstance(args, list):
            for i, tool_call in enumerate(args, 1):
                if isinstance(tool_call, dict):
                    parts.append(f"
  Tool Call {i}:")
                    parts.append(f"    Name: {tool_call.get('name', 'N/A')}")
                    parts.append(f"    ID: {tool_call.get('id', 'N/A')}")
                    parts.append(f"    Type: {tool_call.get('type', 'N/A')}")
                    if 'args' in tool_call:
                        parts.append(f"    Args:")
                        parts.append(f"       {json.dumps(tool_call['args'], indent=7, ensure_ascii=False)}")
        return "
".join(parts)
    
    # Check if message has 'content' attribute
    if not hasattr(message, 'content'):
        # If it's a dict-like object, try to access it as a dict
        if hasattr(message, '__getitem__'):
            try:
                content = message['content'] if 'content' in message else str(message)
                parts.append(content)
                return "
".join(parts)
            except (KeyError, TypeError):
                parts.append(str(message))
                return "
".join(parts)
        else:
            parts.append(str(message))
            return "
".join(parts)
    
    # Handle main content (normal messages with 'content' attribute)
    if isinstance(message.content, str):
        # Try to detect and format structured data within the string
        formatted_content = format_string_with_structures(message.content)
        parts.append(formatted_content)
    elif isinstance(message.content, list):
        # Handle complex content like tool calls (Anthropic format)
        for item in message.content:
            if item.get('type') == 'text':
                parts.append(item['text'])
            elif item.get('type') == 'tool_use':
                parts.append(f"
🔧 Tool Call: {item['name']}")
                parts.append(f"   Args: {str(json.dumps(item['input'], indent=7, ensure_ascii=False))}")
                parts.append(f"   ID: {item.get('id', 'N/A')}")
                tool_calls_processed = True
    else:
        parts.append(str(message.content))
    
    # Handle tool calls attached to the message (OpenAI format) - only if not already processed
    if not tool_calls_processed and hasattr(message, 'tool_calls') and message.tool_calls:
        for tool_call in message.tool_calls:
            parts.append(f"
🔧 Tool Call: {tool_call['name']}")
            parts.append(f"   Args: {json.dumps(tool_call['args'], indent=2)}")
            parts.append(f"   ID: {tool_call['id']}")
    
    return "
".join(parts)
 
def format_messages(messages, title: str = "", border_style: str = "white", msg_subtype: str = ""):
    """Format and display a list of messages with Rich formatting
    
    Args:
        messages: List of messages to format
        title: Title of the panel
        border_style: Border style of the panel
        msg_subtype: Subtype of the message
    """
    # Check if messages is a list
    if isinstance(messages, list):
        for m in messages:
            # Handle different message types
            if isinstance(m, dict):
                # Check if it's a ToolCall dict
                if 'args' in m and 'name' in m:
                    msg_type = 'ToolCall'
                else:
                    msg_type = 'Dict'
            else:
                msg_type = m.__class__.__name__.replace('Message', '')
            
            content = format_message_content(m)
 
            if msg_type == 'Human':
                if msg_subtype == 'RealHumanMessage':
                    if title == "":
                        title = "🧑 Real Human Message"
                    else:
                        title = f"🧑 {title}"
                    console.print(Panel(content, title=title, border_style="#201ADB")) # Blue
                else:
                    if title == "":
                        title = "🧑 Simulated Human Message"
                    else:
                        title = f"🧑 {title}"
                    console.print(Panel(content, title=title, border_style="#1A64DB")) # Blue
            elif msg_type == 'AI':
                if title == "":
                    title = "🤖 Assistant"
                else:
                    title = f"🤖 {title}"
                console.print(Panel(content, title=title, border_style="#24FA00")) # Green
            elif msg_type == 'ClarifyWithUser':
                if title == "":
                    title = "🤖 Clarify With User"
                else:
                    title = f"🤖 {title}"
                console.print(Panel(content, title=title, border_style="#37DB1A")) # Green
            elif msg_type == 'ResearchQuestion':
                if title == "":
                    title = "🤖 Research brief generated"
                else:
                    title = f"🤖 {title}"
                console.print(Panel(content, title=title, border_style="#37DB1A")) # Green
            elif msg_type == 'Tool':
                if title == "":
                    title = "🔧 Tool Output"
                else:
                    title = f"🔧 {title}"
                console.print(Panel(content, title=title, border_style="yellow"))
            elif msg_type == 'ToolCall':
                if title == "":
                    title = "🔧 Tool Calls"
                else:
                    title = f"🔧 {title}"
                console.print(Panel(content, title=title, border_style="magenta"))
            elif msg_type == 'Use tools':
                if title == "":
                    title = "🔧 Tool Output"
                else:
                    title = f"🔧 {title}"
                console.print(Panel(content, title=title, border_style="yellow"))
            elif msg_type == 'System':
                if title == "":
                    title = "🔧 System Message"
                else:
                    title = f"🔧 {title}"
                console.print(Panel(content, title=title, border_style="red"))
            else:
                print(f"msg_type: {msg_type}")
                if title == "":
                    title = f"📝 {msg_type}"
                else:
                    title = f"📝 {title}"
                console.print(Panel(content, title=title, border_style="white"))
    elif isinstance(messages, str):
        console.print(Panel(messages, title=title, border_style=border_style))
    else:
        print(f"messages type: {type(messages)}")
	
	

		
			
		
	
	
Copied

	

		
clarify_with_user_instructions="""
Estos son los mensajes que se han intercambiado hasta ahora con el usuario que solicita el informe:
<Messages>
{messages}
</Messages>
 
La fecha de hoy es {date}.
 
Evalúa si necesitas hacer alguna pregunta aclaratoria o si el usuario ya te ha proporcionado suficiente información para comenzar la investigación.
IMPORTANTE: Si en el historial de mensajes ves que ya has hecho una pregunta aclaratoria, casi siempre no es necesario que hagas otra. Solo haz otra pregunta si es ABSOLUTAMENTE NECESARIO.
 
Si hay acrónimos, abreviaturas o términos desconocidos, pide al usuario que los aclare.
Si necesitas hacer una pregunta, sigue estas pautas:
 - Se conciso al recopilar toda la información necesaria.
 - Asegúrate de recopilar toda la información necesaria para llevar a cabo la tarea de investigación de forma concisa y bien estructurada.
 - Utiliza bullet points o listas numeradas si es necesario para mayor claridad. Asegúrate de que utilizas el formato Markdown y de que se mostrará correctamente si la cadena de salida se pasa a un renderizador Markdown.
 - No solicites información innecesaria o información que el usuario ya haya proporcionado. Si ves que el usuario ya ha proporcionado la información, no se la vuelvas a solicitar.
 
Responde en formato JSON válido con estas claves exactas:
"need_clarification": boolean,
"question": "<pregunta para pedir al usuario que aclare el alcance del informe>",
"verification": "<mensaje de verificación de que comenzaremos la investigación>"
 
Si necesitas hacer una pregunta aclaratoria, devuelve:
"need_clarification": true,
"question": "<tu pregunta aclaratoria>",
"verification": ""
 
Si no necesitas hacer una pregunta aclaratoria, devuelve:
"need_clarification": false,
"question": "",
"verification": "<Mensaje de confirmación de que ahora comenzará la investigación basándose en la información proporcionada.>"
 
Para el mensaje de verificación cuando no se necesita aclaración:
 - Confirma que tienes suficiente información para continuar.
 - Resume brevemente los aspectos clave de lo que entiendes de la solicitud.
 - Confirma que ahora comenzará el proceso de investigación.
 - Mantén el mensaje conciso y profesional.
 
Responde siempre en el idioma en el que el usuario te haya preguntado. En caso de duda, responde en español.
"""
	
	

		
			
		
	
	
Copied

	

		
transform_messages_into_research_topic_prompt = """Se te va a proporcionar un conjunto de mensajes que se han intercambiado hasta ahora entre el usuario y tú. 
Tu trabajo consiste en traducir estos mensajes en una pregunta de investigación más detallada y concreta que se utilizará para orientar la investigación.
 
Los mensajes que se han intercambiado hasta ahora entre el usuario y tú son:
<Messages>
{messages}
</Messages>
 
La fecha de hoy es {date}.
 
Devolverás una única pregunta de investigación que se utilizará para orientar la investigación.
 
Directrices:
1. Maximiza la especificidad y el detalle.
- Incluye todas las preferencias conocidas del usuario y enumera explícitamente los atributos o dimensiones clave a tener en cuenta.
- Es importante que todos los detalles del usuario se incluyan en las instrucciones.
 
2. Trata con cuidado las dimensiones no especificadas
- Cuando la calidad de la investigación requiera tener en cuenta dimensiones adicionales que el usuario no haya especificado, reconócelas como consideraciones abiertas en lugar de preferencias asumidas.
- Ejemplo: en lugar de asumir «opciones económicas», di «considerar todos los rangos de precios a menos que se especifiquen restricciones de coste».
- Solo menciona las dimensiones que sean realmente necesarias para una investigación exhaustiva en ese ámbito.
 
3. Evita suposiciones injustificadas
- Nunca inventes preferencias, restricciones o requisitos específicos del usuario que no se hayan indicado.
- Si el usuario no ha proporcionado un detalle concreto, señala explícitamente esta falta de especificación.
- Orienta al investigador para que trate los aspectos no especificados como flexibles en lugar de hacer suposiciones.
 
4. Distinguir entre el alcance de la investigación y las preferencias del usuario
- Alcance de la investigación: qué temas/dimensiones deben investigarse (puede ser más amplio que las menciones explícitas del usuario)
- Preferencias del usuario: restricciones, requisitos o preferencias específicos (solo debes incluir lo que el usuario haya indicado)
- Ejemplo: «Investigar los factores que influyen en la calidad del café (incluido el origen de los granos, los métodos de tostado y las técnicas de preparación) en las cafeterías de Madrid, centrándose principalmente en el sabor, tal y como ha especificado el usuario».
 
5. Utiliza la primera persona
- Redacta la solicitud desde la perspectiva del usuario.
 
6. Fuentes
- Si se debe dar prioridad a fuentes específicas, entonces específalas en la pregunta de investigación.
- Para la investigación de productos y viajes, es preferible enlazar directamente con sitios web oficiales o primarios (por ejemplo, sitios web oficiales de marcas, páginas de fabricantes o plataformas de comercio electrónico de renombre como Amazon para opiniones de usuarios) en lugar de sitios agregadores o blogs con mucho SEO.
- Para consultas académicas o científicas, es preferible enlazar directamente con el artículo original o la publicación oficial en una revista, en lugar de con artículos de encuestas o resúmenes secundarios.
- En el caso de personas, intente enlazar directamente con su perfil de LinkedIn, su sitio web personal o su perfil de GitHub, si lo tienen.
- Si la consulta está en un idioma específico, dé prioridad a las fuentes publicadas en ese idioma.
 
Responde siempre en el idioma en el que el usuario te haya preguntado. En caso de duda, responde en español.
"""
	
	

		
			
		
	
	
Copied

	

		
import operator
from typing_extensions import Optional, Annotated, Sequence
 
from langchain_core.messages import BaseMessage
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
 
class AgentInputState(MessagesState):
    """Input state for the full agent - only contains messages from user input."""
    pass
 
class AgentState(MessagesState):
    """
    Main state for the full multi-agent research system.
 
    Extends MessagesState with additional fields for research coordination.
    Note: Some fields are duplicated across different state classes for proper
    state management between subgraphs and the main workflow.
    """
 
    # Research brief generated from user conversation history
    research_brief: Optional[str]
    # Messages exchanged with the supervisor agent for coordination
    supervisor_messages: Annotated[Sequence[BaseMessage], add_messages]
    # Raw unprocessed research notes collected during the research phase
    raw_notes: Annotated[list[str], operator.add] = []
    # Processed and structured notes ready for report generation
    notes: Annotated[list[str], operator.add] = []
    # Final formatted research report
    final_report: str
	
	

		
			
		
	
	
Copied

state = {
  "messages": ["Hola, ¿cómo estás?"],
  "research_brief": "Investigar los factores que influyen en la calidad del café",
  "supervisor_messages": [],
  "raw_notes": [],
  "notes": [],
  "final_report": ""
}

agent.invoke(state)

	

		
from pydantic import BaseModel, Field
 
class ClarifyWithUser(BaseModel):
    """Schema for user clarification decision and questions."""
 
    need_clarification: bool = Field(
        description="Whether the user needs to be asked a clarifying question.",
    )
    question: str = Field(
        description="A question to ask the user to clarify the report scope",
    )
    verification: str = Field(
        description="Verify message that we will start research after the user has provided the necessary information.",
    )
 
class ResearchQuestion(BaseModel):
    """Schema for structured research brief generation."""
 
    research_brief: str = Field(
        description="A research question that will be used to guide the research.",
    )
	
	

		
			
		
	
	
Copied

	

		
from langchain.chat_models import init_chat_model
import os
import dotenv
 
dotenv.load_dotenv()
 
SCOPE_MODEL_NAME = "gpt-4.1"
SCOPE_MODEL_PROVIDER = "openai"
SCOPE_MODEL_BASE_URL = "https://api.openai.com/v1"
SCOPE_MODEL_TEMPERATURE = 0.0
SCOPE_MODEL_PROVIDER_API_KEY = os.getenv("OPENAI_API_KEY")
 
scope_model = init_chat_model(
    model=SCOPE_MODEL_NAME,
    model_provider=SCOPE_MODEL_PROVIDER,
    api_key=SCOPE_MODEL_PROVIDER_API_KEY,
    base_url=SCOPE_MODEL_BASE_URL,
    temperature=SCOPE_MODEL_TEMPERATURE
)
	
	

		
			
		
	
	
Copied

	

		
from typing_extensions import Literal
from alive_progress import alive_bar
import traceback
import sys
 
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, get_buffer_string
from langgraph.graph import START, END
from langgraph.types import Command
 
def clarify_with_user(state: AgentState) -> Command[Literal["write_research_brief", "__end__"]]:
    """
    Determine if the user's request contains sufficient information to proceed with research.
    
    Uses structured output to make deterministic decisions and avoid hallucination.
    Routes to either research brief generation or ends with a clarification question.
    """
 
    try:
        print("⏳ Scope agent:")
        format_messages([state.get("messages", [])[-1]], title="Real Human Message", msg_subtype='RealHumanMessage')
 
        # Set up structured output model
        structured_output_model = scope_model.with_structured_output(ClarifyWithUser)
 
        # Invoke the model with clarification instructions
        with alive_bar(monitor=False, stats=False, title="", spinner='dots_waves', bar='blocks') as bar:
            response = structured_output_model.invoke([
                HumanMessage(content=clarify_with_user_instructions.format(
                    messages=get_buffer_string(messages=state["messages"]),
                    date=get_today_str()
                ))
            ])
            bar()
        
        # Format and display the research messages
        format_messages([response], title="Scope Assistant - need clarification?")
        
        # Route based on clarification need
        if response.need_clarification:
            # Create a System message to show the decision
            routing_message = SystemMessage(
                content="Necesita aclaración por parte del usuario. Enviando pregunta aclaratoria..."
            )
            format_messages([routing_message], title="Scope System Message")
            return Command(
                goto=END,
                update={"messages": [AIMessage(content=response.question)]}
            )
        else:
            # Create a System message to show the decision
            routing_message = SystemMessage(
                content="No necesita aclaración por parte del usuario. Enviando mensaje de verificación..."
            )
            format_messages([routing_message], title="Scope System Message")
            return Command(
                goto="write_research_brief",
                update={"messages": [AIMessage(content=response.verification)]}
            )
    
    except Exception as e:
        # Get the traceback information
        exc_type, exc_obj, exc_tb = sys.exc_info()
        line_number = exc_tb.tb_lineno
        
        # Print detailed error information
        print(f"
❌ Error in clarify_with_user function:")
        print(f"   Line number: {line_number}")
        print(f"   Error type: {type(e).__name__}")
        print(f"   Error message: {str(e)}")
        print(f"
   Full traceback:")
        traceback.print_exc()
        
        # Re-raise the exception to let the caller handle it
        raise
	
	

		
			
		
	
	
Copied

	

		
def write_research_brief(state: AgentState):
    """
    Transform the conversation history into a comprehensive research brief.
    
    Uses structured output to ensure the brief follows the required format
    and contains all necessary details for effective research.
    """
    
    try:
        # Set up structured output model
        structured_output_model = scope_model.with_structured_output(ResearchQuestion)
 
        print("⏳ Scope agent - Write research brief:")
        format_messages([state.get("messages", [])[-1]], title="Scope Assistant - Write research brief")
        
        # Generate research brief from conversation history
        with alive_bar(monitor=False, stats=False, title="", spinner='dots_waves', bar='blocks') as bar:
            response = structured_output_model.invoke([
                HumanMessage(content=transform_messages_into_research_topic_prompt.format(
                    messages=get_buffer_string(state.get("messages", [])),
                    date=get_today_str()
                ))
            ])
            bar()
        
        # Format and display the research brief
        format_messages([response], title="Scope Assistant - Research brief generated")
        
        # Update state with generated research brief and pass it to the supervisor
        return {
            "research_brief": response.research_brief,
            "supervisor_messages": [HumanMessage(content=f"{response.research_brief}.")]
        }
    
    except Exception as e:
        # Get the traceback information
        exc_type, exc_obj, exc_tb = sys.exc_info()
        line_number = exc_tb.tb_lineno
        
        # Print detailed error information
        print(f"
❌ Error in write_research_brief function:")
        print(f"   Line number: {line_number}")
        print(f"   Error type: {type(e).__name__}")
        print(f"   Error message: {str(e)}")
        print(f"
   Full traceback:")
        traceback.print_exc()
        
        # Re-raise the exception to let the caller handle it
        raise
	
	

		
			
		
	
	
Copied

	

		
from langgraph.graph import StateGraph
 
# Build the scoping workflow
scope_graph_builder = StateGraph(AgentState, input_schema=AgentInputState)
	
	

		
			
		
	
	
Copied

	

		
# Add workflow nodes
scope_graph_builder.add_node("clarify_with_user", clarify_with_user)
scope_graph_builder.add_node("write_research_brief", write_research_brief)
	
	

		
			
		
	
	
Copied

	

		
<langgraph.graph.state.StateGraph at 0x10a3dc410>
	

	

		
from langgraph.graph import START, END
 
# Add workflow edges
scope_graph_builder.add_edge(START, "clarify_with_user")
scope_graph_builder.add_edge("write_research_brief", END)
	
	

		
			
		
	
	
Copied

	

		
<langgraph.graph.state.StateGraph at 0x10a3dc410>
	

	

		
from langgraph.checkpoint.memory import InMemorySaver
from IPython.display import Image, display
 
checkpointer = InMemorySaver()
scope_agent = scope_graph_builder.compile(checkpointer=checkpointer)
 
# Visualize the workflow
display(Image(scope_agent.get_graph(xray=True).draw_mermaid_png()))
	
	

		
			
		
	
	
Copied

	

		
<IPython.core.display.Image object>
	

	

		
thread = {"configurable": {"thread_id": "1"}}
user_message = "Quiero investigar las mejores cafeterías de Madrid."
 
result = await scope_agent.ainvoke({"messages": [HumanMessage(content=f"{user_message}.")]}, config=thread)
	
	

		
			
		
	
	
Copied

	

		
⏳ Scope agent:
	

	

		
╭───────────────────────────────────────────── 🧑 Real Human Message ─────────────────────────────────────────────╮
│ Quiero investigar las mejores cafeterías de Madrid..                                                            │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

	

		
|▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉| in 2.7s
	

	

		
╭─────────────────────────────────── 🤖 Scope Assistant - need clarification? ────────────────────────────────────╮
│ need_clarification=True question='¿Podrías especificar qué criterios te interesan para considerar una cafetería │
│ como "mejor"? Por ejemplo: calidad del café, ambiente, ubicación, precio, opciones veganas, etc. ¿O prefieres   │
│ una lista general?' verification=''                                                                             │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

	

		
╭──────────────────────────────────────────── 🔧 Scope System Message ────────────────────────────────────────────╮
│ Necesita aclaración por parte del usuario. Enviando pregunta aclaratoria...                                     │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

	

		
user_message = "Examina la calidad del café para evaluar las mejores cafeterías de Madrid."
 
result = await scope_agent.ainvoke({"messages": [HumanMessage(content=f"{user_message}.")]}, config=thread)
	
	

		
			
		
	
	
Copied

	

		
⏳ Scope agent:
	

	

		
╭───────────────────────────────────────────── 🧑 Real Human Message ─────────────────────────────────────────────╮
│ Examina la calidad del café para evaluar las mejores cafeterías de Madrid..                                     │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

	

		
|▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉| in 2.3s
	

	

		
╭─────────────────────────────────── 🤖 Scope Assistant - need clarification? ────────────────────────────────────╮
│ need_clarification=False question='' verification='Gracias por la información. Entiendo que deseas un informe   │
│ sobre las mejores cafeterías de Madrid, evaluadas específicamente por la calidad del café. Comenzaré la         │
│ investigación basándome en este criterio principal. Ahora iniciaré el proceso de investigación.'                │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

	

		
╭──────────────────────────────────────────── 🔧 Scope System Message ────────────────────────────────────────────╮
│ No necesita aclaración por parte del usuario. Enviando mensaje de verificación...                               │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

	

		
⏳ Scope agent - Write research brief:
	

	

		
╭─────────────────────────────────── 🤖 Scope Assistant - Write research brief ───────────────────────────────────╮
│ Gracias por la información. Entiendo que deseas un informe sobre las mejores cafeterías de Madrid, evaluadas    │
│ específicamente por la calidad del café. Comenzaré la investigación basándome en este criterio principal. Ahora │
│ iniciaré el proceso de investigación.                                                                           │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

	

		
|▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉▉| in 3.9s
	

	

		
╭───────────────────────────────── 🤖 Scope Assistant - Research brief generated ─────────────────────────────────╮
│ research_brief='Quiero investigar cuáles son las mejores cafeterías de Madrid evaluadas específicamente por la  │
│ calidad del café. Para ello, me interesa identificar cafeterías que destaquen por la excelencia en el sabor,    │
│ aroma, frescura y preparación del café, considerando factores como el origen de los granos, el método de        │
│ tostado y las técnicas de extracción utilizadas. No tengo preferencias sobre el rango de precios, ubicación,    │
│ ambiente, opciones alimenticias o servicios adicionales, por lo que estos aspectos pueden considerarse abiertos │
│ o secundarios. Solicito priorizar fuentes primarias y confiables, como los sitios web oficiales de las          │
│ cafeterías, reseñas de expertos en café y plataformas reconocidas de opinión de usuarios en español.'           │
╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
	

Si recordamos cómo iba a ser la arquitectura del deep researcher, vemos que hay un Research supervisor que se encarga de asignar investigaciones a diferentes Researchers

	

		
research_agent_prompt = """Eres un asistente de investigación que lleva a cabo una investigación sobre el tema introducido por el usuario. Para contextualizar, hoy es {date}.
 
<Task>
Tu trabajo consiste en utilizar herramientas para recopilar información sobre el tema introducido por el usuario.
Puedes utilizar cualquiera de las herramientas que se te proporcionan para encontrar recursos que te ayuden a responder a la pregunta de investigación. Puedes utilizar estas herramientas en serie o en paralelo, tu investigación se lleva a cabo en un bucle de llamada de herramientas.
</Task>
 
<Available Tools>
Tienes acceso a dos herramientas principales:
1. **tavily_search**: para realizar búsquedas en la web con el fin de recopilar información.
2. **think_tool**: para la reflexión y la planificación estratégica durante la investigación.
 
**IMPORTANTE: utiliza think_tool después de cada búsqueda para reflexionar sobre los resultados y planificar los siguientes pasos**.
**CRÍTICO: Siempre debes proporcionar el parámetro 'reflection' con contenido detallado cuando llames a think_tool. Nunca llames a esta herramienta sin argumentos**.
</Available Tools>
 
<Instructions>
Piensa como un investigador humano con tiempo limitado. Sigue estos pasos:
 
1. **Lee la pregunta con atención**: ¿qué información específica necesita el usuario?
2. **Empieza con búsquedas más amplias**: utiliza primero consultas amplias y exhaustivas.
3. **Después de cada búsqueda, haz una pausa y evalúa**: ¿tengo suficiente para responder? ¿Qué me falta?
4. **Realiza búsquedas más específicas a medida que recopilas información**: completa los huecos.
5. **Detente cuando puedas responder con seguridad**: no sigas buscando la perfección.
</Instructions>
 
<Hard Limits>
**Presupuestos de llamadas a herramientas** (evita búsquedas excesivas):
- **Consultas simples**: utiliza un máximo de 2-3 llamadas a herramientas de búsqueda.
- **Consultas complejas**: utiliza un máximo de 5 llamadas a herramientas de búsqueda
- **Detente siempre**: después de 5 llamadas a herramientas de búsqueda si no puedes encontrar las fuentes adecuadas
 
**Detente inmediatamente cuando**:
- Puedas responder a la pregunta del usuario de forma exhaustiva
- Tengas más de 3 ejemplos/fuentes relevantes para la pregunta
- Tus últimas 2 búsquedas hayan devuelto información similar
</Hard Limits>
 
<Show Your Thinking>
Después de cada llamada a la herramienta de búsqueda, utiliza think_tool para analizar los resultados:
- ¿Qué información clave he encontrado?
- ¿Qué falta?
- ¿Tengo suficiente para responder a la pregunta de forma exhaustiva?
- ¿Debo seguir buscando o dar mi respuesta?
</Show Your Thinking>
"""
	
	

		
			
		
	
	
Copied

	

		
summarize_webpage_prompt = """Tu tarea consiste en resumir el contenido sin procesar de una página web obtenida mediante una búsqueda en Internet. Tu objetivo es crear un resumen que conserve la información más importante de la página web original. Este resumen será utilizado por un agente de investigación posterior, por lo que es fundamental mantener los detalles clave sin perder información esencial.
 
Este es el contenido sin procesar de la página web:
 
<webpage_content>
{webpage_content}
</webpage_content>
 
Sigue estas pautas para crear tu resumen:
 
1. Identifica y conserva el tema principal o el propósito de la página web.
2. Conserva los datos, estadísticas y puntos clave que son fundamentales para el mensaje del contenido.
3. Conserva las citas importantes de fuentes o expertos fiables.
4. Mantén el orden cronológico de los acontecimientos si el contenido es sensible al tiempo o histórico.
5. Conserva las listas o instrucciones paso a paso, si las hay.
6. Incluye fechas, nombres y lugares relevantes que sean cruciales para comprender el contenido.
7. Resume las explicaciones largas manteniendo intacto el mensaje principal.
 
Cuando se trata de diferentes tipos de contenido:
 
- Para artículos de noticias: concéntrate en quién, qué, cuándo, dónde, por qué y cómo.
- Para contenido científico: conserva la metodología, los resultados y las conclusiones.
- Para artículos de opinión: mantén los argumentos principales y los puntos de apoyo.
- Para páginas de productos: mantén las características clave, las especificaciones y los puntos de venta únicos.
 
Tu resumen debe ser significativamente más corto que el contenido original, pero lo suficientemente completo como para constituir una fuente de información por tí mismo. Intenta que tenga entre un 25 % y un 30 % de la longitud original, a menos que el contenido ya sea conciso.
 
Presenta tu resumen en el siguiente formato:
 
```
{{
   "summary": "Tu resumen aquí, estructurado con párrafos o viñetas adecuados según sea necesario",
   "key_excerpts": "Primera cita o extracto importante, Segunda cita o extracto importante, Tercer cita o extracto importante, ...Añade más extractos según sea necesario, hasta un máximo de 5"
}}
```
 
Aquí hay dos ejemplos de buenos resúmenes:
 
Ejemplo 1 (para un artículo de noticias):
```json
{{
   "summary": "El 15 de julio de 2023, la NASA lanzó con éxito la misión Artemis II desde el Centro Espacial Kennedy. Se trata de la primera misión tripulada a la Luna desde el Apolo 17 en 1972. La tripulación de cuatro personas, dirigida por la comandante Jane Smith, orbitará la Luna durante 10 días antes de regresar a la Tierra. Esta misión es un paso crucial en los planes de la NASA para establecer una presencia humana permanente en la Luna para 2030.",
   "key_excerpts": "Artemis II representa una nueva era en la exploración espacial, dijo el administrador de la NASA, John Doe. La misión pondrá a prueba sistemas críticos para futuras estancias de larga duración en la Luna, explicó la ingeniera jefe Sarah Johnson. No solo vamos a volver a la Luna, vamos a avanzar hacia la Luna, declaró la comandante Jane Smith durante la rueda de prensa previa al lanzamiento"
}}
```
 
Ejemplo 2 (para un artículo científico):
```json
{{
   "summary": "Un nuevo estudio publicado en Nature Climate Change revela que el nivel del mar está subiendo más rápido de lo que se pensaba. Los investigadores analizaron datos satelitales de 1993 a 2022 y descubrieron que la tasa de aumento del nivel del mar se ha acelerado en 0,08 mm/año² durante las últimas tres décadas. Esta aceleración se atribuye principalmente al derretimiento de los casquetes polares de Groenlandia y la Antártida. El estudio prevé que, si las tendencias actuales continúan, el nivel global del mar podría aumentar hasta 2 metros para 2100, lo que supondría un riesgo significativo para las comunidades costeras de todo el mundo.",
   "key_excerpts": "Nuestros hallazgos indican una clara aceleración en el aumento del nivel del mar, lo que tiene importantes implicaciones para la planificación costera y las estrategias de adaptación», afirmó la autora principal, la Dra. Emily Brown. La velocidad de deshielo de las capas de hielo de Groenlandia y la Antártida se ha triplicado desde la década de 1990, según el estudio. Sin una reducción inmediata y sustancial de las emisiones de gases de efecto invernadero, nos enfrentamos a un aumento potencialmente catastrófico del nivel del mar a finales de este siglo, advirtió el coautor, el profesor Michael Green."  
}}
```
 
Recuerda que tu objetivo es crear un resumen que pueda ser fácilmente comprendido y utilizado por un agente de investigación posterior, conservando al mismo tiempo la información más importante de la página web original.
 
La fecha de hoy es {date}.
"""
	
	

		
			
		
	
	
Copied

	

		
compress_research_system_prompt = """Eres un asistente de investigación que ha investigado un tema utilizando diversas herramientas y búsquedas en Internet. Tu trabajo ahora consiste en limpiar los resultados, conservando todas las declaraciones e información relevantes recopiladas por el investigador. Para contextualizar, hoy es {date}.
 
<Task>
Debes limpiar la información recopilada de las herramientas y búsquedas en Internet en los mensajes existentes.
Toda la información relevante debe repetirse y reescribirse textualmente, pero en un formato más limpio.
El objetivo de este paso es simplemente eliminar cualquier información obviamente irrelevante o duplicada.
Por ejemplo, si tres fuentes dicen «X», podrías decir «Estas tres fuentes afirman X».
Solo estos resultados completos y limpios se devolverán al usuario, por lo que es esencial que no pierdas ninguna información de los mensajes sin procesar.
</Task>
 
<Tool Call Filtering>
**IMPORTANTE**: Al procesar los mensajes de investigación, céntrate solo en el contenido sustantivo de la investigación:
- **Incluir**: Todos los resultados de tavily_search y los hallazgos de la búsqueda web
- **Excluir**: las llamadas y respuestas de think_tool, ya que son reflexiones internas del agente para la toma de decisiones y no deben incluirse en el informe final de la investigación
- **Céntrate en**: la información real recopilada de fuentes externas, no en el proceso de razonamiento interno del agente.
 
Las llamadas a think_tool contienen reflexiones estratégicas y notas sobre la toma de decisiones que son internas al proceso de investigación, pero no contienen información factual que deba conservarse en el informe final.
</Tool Call Filtering>
 
<Guidelines>
1. Los resultados deben ser exhaustivos e incluir TODA la información y las fuentes que el investigador haya recopilado a partir de las llamadas a herramientas y las búsquedas en la web. Se espera que repita la información clave textualmente.
2. Este informe puede tener la extensión necesaria para incluir TODA la información que el investigador haya recopilado.
3. En tu informe, debes incluir citas en línea para cada fuente que el investigador haya encontrado.
4. Debes incluir una sección «Fuentes» al final del informe en la que se enumeran todas las fuentes que el investigador ha encontrado con las citas correspondientes, referenciadas en relación con las afirmaciones del informe.
5. Asegúrese de incluir TODAS las fuentes recopiladas por el investigador en el informe y cómo se utilizaron para responder a la pregunta.
6. Es muy importante no perder ninguna fuente. Más adelante se utilizará un LLM para fusionar este informe con otros, por lo que es esencial disponer de todas las fuentes.
</Guidelines>
 
<Output Format>
El informe debe estructurarse de la siguiente manera:
**Lista de consultas y llamadas a herramientas realizadas**
**Resultados completos**
**Lista de todas las fuentes relevantes (con citas en el informe)**
</Output Format>
 
<Citation Rules>
- Asigna a cada URL única un número de citación único en tu texto.
- Termina con ### Fuentes, enumerando cada fuente con los números correspondientes.
- IMPORTANTE: Numera las fuentes de forma secuencial sin saltos (1, 2, 3, 4...) en la lista final, independientemente de las fuentes que elijas.
- Ejemplo de formato:
  [1] Título de la fuente: URL
  [2] Título de la fuente: URL
</Citation Rules>
 
Recordatorio importante: Es extremadamente importante que cualquier información que sea remotamente relevante para el tema de investigación del usuario se conserve textualmente (es decir, no la reescribas, no la resumas, no la parafrasees).
"""
	
	

		
			
		
	
	
Copied

	

		
compress_research_human_message = """Todos los mensajes anteriores se refieren a la investigación realizada por un investigador de IA sobre el siguiente tema de investigación:
 
TEMA DE INVESTIGACIÓN: {research_topic}
 
Tu tarea consiste en depurar los resultados de esta investigación conservando TODA la información relevante para responder a esta pregunta de investigación específica.
 
REQUISITOS FUNDAMENTALES:
- NO resumas ni parafrasees la información: consérvala textualmente.
- NO pierdas ningún detalle, dato, nombre, número o hallazgo específico
- NO filtres la información que parezca relevante para el tema de investigación
- Organiza la información en un formato más claro, pero mantén todo el contenido
- Incluye TODAS las fuentes y citas encontradas durante la investigación
- Recuerda que esta investigación se llevó a cabo para responder a la pregunta específica anterior
 
Los resultados depurados se utilizarán para la elaboración del informe final, por lo que es fundamental que sean exhaustivos."""
	
	

		
			
		
	
	
Copied

	

		
from typing_extensions import TypedDict, Annotated, List, Sequence
 
class ResearcherState(TypedDict):
    """
    State for the research agent containing message history and research metadata.
    
    This state tracks the researcher's conversation, iteration count for limiting
    tool calls, the research topic being investigated, compressed findings,
    and raw research notes for detailed analysis.
    """
    researcher_messages: Annotated[Sequence[BaseMessage], add_messages]
    tool_call_iterations: int
    research_topic: str
    compressed_research: str
    raw_notes: Annotated[List[str], operator.add]
 
class ResearcherOutputState(TypedDict):
    """
    Output state for the research agent containing final research results.
    
    This represents the final output of the research process with compressed
    research findings and all raw notes from the research process.
    """
    compressed_research: str
    raw_notes: Annotated[List[str], operator.add]
    researcher_messages: Annotated[Sequence[BaseMessage], add_messages]
	
	

		
			
		
	
	
Copied

	

		
class Summary(BaseModel):
    """Schema for webpage content summarization."""
    summary: str = Field(description="Concise summary of the webpage content")
    key_excerpts: str = Field(description="Important quotes and excerpts from the content")
	
	

		
			
		
	
	
Copied

	

		
def tavily_search_multiple(
    search_queries: List[str],
    max_results: int = 3,
    topic: Literal["general", "news", "finance"] = "general",
    include_raw_content: bool = True,
) -> List[dict]:
    """Perform search using Tavily API for multiple queries.
 
    Args:
        search_queries: List of search queries to execute
        max_results: Maximum number of results per query
        topic: Topic filter for search results
        include_raw_content: Whether to include raw webpage content
 
    Returns:
        List of search result dictionaries
    """
    
    # Execute searches sequentially. Note: yon can use AsyncTavilyClient to parallelize this step.
    search_docs = []
    for query in search_queries:
        result = tavily_client.search(
            query,
            max_results=max_results,
            include_raw_content=include_raw_content,
            topic=topic
        )
        search_docs.append(result)
 
    return search_docs
	
	

		
			
		
	
	
Copied

	

		
def deduplicate_search_results(search_results: List[dict]) -> dict:
    """Deduplicate search results by URL to avoid processing duplicate content.
    
    Args:
        search_results: List of search result dictionaries
        
    Returns:
        Dictionary mapping URLs to unique results
    """
    unique_results = {}
    
    for response in search_results:
        for result in response['results']:
            url = result['url']
            if url not in unique_results:
                unique_results[url] = result
    
    return unique_results
	
	

		
			
		
	
	
Copied

	

		
def summarize_webpage_content(webpage_content: str) -> str:
    """Summarize webpage content using the configured summarization model.
    
    Args:
        webpage_content: Raw webpage content to summarize
        
    Returns:
        Formatted summary with key excerpts
    """
    try:
        # Set up structured output model for summarization
        structured_model = summarization_model.with_structured_output(Summary)
        
        # Generate summary
        summary = structured_model.invoke([
            HumanMessage(content=summarize_webpage_prompt.format(
                webpage_content=webpage_content,
                date=get_today_str()
            ))
        ])
        
        # Format summary with clear structure
        formatted_summary = (
            f"<summary>
{summary.summary}
</summary>

"
            f"<key_excerpts>
{summary.key_excerpts}
</key_excerpts>"
        )
        
        return formatted_summary
        
    except Exception as e:
        print(f"Failed to summarize webpage: {str(e)}")
        return webpage_content[:1000] + "..." if len(webpage_content) > 1000 else webpage_content
	
	

		
			
		
	
	
Copied

	

		
def process_search_results(unique_results: dict) -> dict:
    """Process search results by summarizing content where available.
    
    Args:
        unique_results: Dictionary of unique search results
        
    Returns:
        Dictionary of processed results with summaries
    """
    summarized_results = {}
    
    for url, result in unique_results.items():
        # Use existing content if no raw content for summarization
        if not result.get("raw_content"):
            content = result['content']
        else:
            # Summarize raw content for better processing
            content = summarize_webpage_content(result['raw_content'])
        
        summarized_results[url] = {
            'title': result['title'],
            'content': content
        }
    
    return summarized_results
	
	

		
			
		
	
	
Copied

	

		
def format_search_output(summarized_results: dict) -> str:
    """Format search results into a well-structured string output.
    
    Args:
        summarized_results: Dictionary of processed search results
        
    Returns:
        Formatted string of search results with clear source separation
    """
    if not summarized_results:
        return "No valid search results found. Please try different search queries or use a different search API."
    
    formatted_output = "Search results: 

"
    
    for i, (url, result) in enumerate(summarized_results.items(), 1):
        formatted_output += f"

--- SOURCE {i}: {result['title']} ---
"
        formatted_output += f"URL: {url}

"
        formatted_output += f"SUMMARY:
{result['content']}

"
        formatted_output += "-" * 80 + "
"
    
    return formatted_output
	
	

		
			
		
	
	
Copied

	

		
from langchain_core.tools import tool, InjectedToolArg
 
@tool(parse_docstring=True)
def tavily_search(
    query: str,
    max_results: Annotated[int, InjectedToolArg] = 3,
    topic: Annotated[Literal["general", "news", "finance"], InjectedToolArg] = "general",
) -> str:
    """Fetch results from Tavily search API with content summarization.
 
    Args:
        query: A single search query to execute
        max_results: Maximum number of results to return
        topic: Topic to filter results by ('general', 'news', 'finance')
    """
    # Execute search for single query
    search_results = tavily_search_multiple(
        [query], # Convert single query to list for the internal function
        max_results=max_results,
        topic=topic,
        include_raw_content=True,
    )
 
    # Deduplicate results by URL to avoid processing duplicate content
    unique_results = deduplicate_search_results(search_results)
 
    # Process results with summarization
    summarized_results = process_search_results(unique_results)
 
    # Format output for consumption
    return format_search_output(summarized_results)
	
	

		
			
		
	
	
Copied

{
  "query": str,
  "max_results": int,
  "topic": Literal["general", "news", "finance"]
}

Tengo un buscador que:
- Recibe: 
 - query (string): "Una consulta de búsqueda simple"
 - max_results (int): "Número máximo de resultados a devolver"
 - topic (Literal): "Categoría para filtrar ('general', 'news', 'finance')"

  max_results: Annotated[int, InjectedToolArg] = 3,
  topic: Annotated[Literal["general", "news", "finance"], InjectedToolArg] = "general",

Tengo un buscador que:
- Recibe: 
 - query (string): "Una consulta de búsqueda simple"

	

		
@tool(parse_docstring=True)
def think_tool(reflection: str = "") -> str:
    """Record your strategic reflection on research progress (REQUIRED).
 
    You MUST provide a detailed reflection analyzing your current findings,
    gaps, and next steps. Never call this tool without providing reflection text.
 
    Use this tool after each search to analyze results and plan next steps systematically.
    This creates a deliberate pause in the research workflow for quality decision-making.
 
    When to use:
    - After receiving search results: What key information did I find?
    - Before deciding next steps: Do I have enough to answer comprehensively?
    - When assessing research gaps: What specific information am I still missing?
    - Before concluding research: Can I provide a complete answer now?
 
    Args:
        reflection: (REQUIRED) Your detailed reflection on research progress, findings, gaps, and next steps.
                   Your reflection should include analysis of current findings, gap assessment,
                   quality evaluation, and strategic decision about next steps.
    """
    if not reflection:
        return "Reflection recorded: [No reflection provided]"
    return f"Reflection recorded: {reflection}"
	
	

		
			
		
	
	
Copied

	

		
from langchain.chat_models import init_chat_model
import os
import dotenv
 
dotenv.load_dotenv()
 
SUMMARIZATION_MODEL_NAME = "gpt-4.1-mini"
SUMMARIZATION_MODEL_PROVIDER = "openai"
SUMMARIZATION_MODEL_BASE_URL = "https://api.openai.com/v1"
SUMMARIZATION_MODEL_TEMPERATURE = 0.0
SUMMARIZATION_MODEL_PROVIDER_API_KEY = os.getenv("OPENAI_API_KEY")
 
summarization_model = init_chat_model(
    model=SUMMARIZATION_MODEL_NAME,
    model_provider=SUMMARIZATION_MODEL_PROVIDER,
    api_key=SUMMARIZATION_MODEL_PROVIDER_API_KEY,
    base_url=SUMMARIZATION_MODEL_BASE_URL,
    temperature=SUMMARIZATION_MODEL_TEMPERATURE
)
	
	

		
			
		
	
	
Copied

	

		
RESEARCH_MODEL_NAME = "anthropic:claude-sonnet-4-5-20250929"
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
RESEARCH_MODEL_MAX_TOKENS = 4096
 
research_model = init_chat_model(
    model=RESEARCH_MODEL_NAME,
    max_tokens=RESEARCH_MODEL_MAX_TOKENS
)
	
	

		
			
		
	
	
Copied

	

		
researcher_tools = [tavily_search, think_tool]
 
model_with_tools = research_model.bind_tools(researcher_tools)
	
	

		
			
		
	
	
Copied

	

		
COMPRESS_MODEL_NAME = "gpt-4.1"
COMPRESS_MODEL_PROVIDER = "openai"
COMPRESS_MODEL_BASE_URL = "https://api.openai.com/v1"
COMPRESS_MODEL_TEMPERATURE = 0.0
COMPRESS_MODEL_PROVIDER_API_KEY = os.getenv("OPENAI_API_KEY")
 
compress_model = init_chat_model(
    model=COMPRESS_MODEL_NAME,
    model_provider=COMPRESS_MODEL_PROVIDER,
    api_key=COMPRESS_MODEL_PROVIDER_API_KEY,
    base_url=COMPRESS_MODEL_BASE_URL,
    temperature=COMPRESS_MODEL_TEMPERATURE,
    max_tokens=32000
)
	
	

		
			
		
	
	
Copied