LangChain & LangGraph for AI Agents
Back to Agent Foundations & LCEL Composition

Case 1 — Autonomous Financial & SQL Tool-Calling Agent

You have a working LCEL chain. It needs loops, retries, and human review. Refactor to LangGraph properly.

63 minutesProject submission
🎯 Free Guest Mode: You are learning for free. Sign in to save your completion progress and quiz answers.

Assignment Instructions

Executive Briefing: A fintech firm requires an autonomous financial intelligence agent to answer ad-hoc investor queries (e.g. "What was Apple's P/E ratio in Q3 2025, and did revenue grow faster than operating expenses?"). The agent must parse multi-step financial inquiries, invoke a PostgreSQL database tool and a real-time stock pricing API, validate arguments with Pydantic v2 schemas, and synthesize a structured, cited investment memo.


1. Enterprise Scenario & Requirements

In production agentic AI, models do not call APIs directly—they output structured arguments that your application runtime validates and executes. In this case, you will build a resilient tool-calling agent using LangChain Expression Language (LCEL) and ChatOpenAI function binding:

  1. Schema Validation: Define strict Pydantic v2 input schemas with field-level descriptions. Malformed user inputs must trigger clean validation errors before API execution.
  2. Tool Binding: Bind tools to the chat model using .bind_tools().
  3. Execution Loop: Implement an execution runner that inspects AIMessage.tool_calls, dispatches calls to verified tools, and passes ToolMessage results back to the model for synthesis.
  4. Error Handling: Gracefully catch simulated database connectivity failures or rate limits (HTTP 429), returning structured fallback messages rather than crashing.

2. Explicit Deliverables

  1. tools.py: Custom @tool definitions (query_financial_db, get_ticker_fundamentals) with Pydantic v2 argument schemas.
  2. agent.py: Complete LCEL tool-calling agent loop that handles reasoning, tool execution, and response synthesis.
  3. test_agent.py: Test suite verifying: (a) single tool invocation, (b) multi-tool sequential reasoning, (c) input schema validation traps, and (d) fallback handling on tool errors.
  4. sample_run.log: Terminal output demonstrating the agent answering 3 real-world investment queries.

3. Starter Code & Scaffold

from typing import Optional
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI

class FinancialMetricInput(BaseModel):
    ticker: str = Field(description="Stock ticker symbol in uppercase, e.g. 'AAPL', 'MSFT'")
    fiscal_period: str = Field(description="Fiscal period string, e.g. 'FY2025', 'Q3-2025'")
    metric: str = Field(description="Target financial metric, e.g. 'revenue', 'net_income', 'pe_ratio'")

@tool(args_schema=FinancialMetricInput)
def get_ticker_fundamentals(ticker: str, fiscal_period: str, metric: str) -> dict:
    '''Retrieve fundamental accounting metrics for a public corporation.'''
    mock_db = {
        ("AAPL", "Q3-2025", "revenue"): 85_200_000_000,
        ("AAPL", "Q3-2025", "pe_ratio"): 32.4,
        ("MSFT", "Q3-2025", "revenue"): 64_700_000_000,
    }
    key = (ticker.upper(), fiscal_period, metric.lower())
    val = mock_db.get(key, 120_500_000)
    return {"ticker": ticker.upper(), "period": fiscal_period, "metric": metric, "value": val}

def build_financial_agent(llm: ChatOpenAI, tools: list):
    '''Construct and return the tool-calling execution runner.'''
    llm_with_tools = llm.bind_tools(tools)
    tool_map = {t.name: t for t in tools}
    
    def run(query: str) -> str:
        messages = [HumanMessage(content=query)]
        response = llm_with_tools.invoke(messages)
        messages.append(response)
        
        while response.tool_calls:
            for tc in response.tool_calls:
                selected_tool = tool_map[tc["name"]]
                result = selected_tool.invoke(tc["args"])
                messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
            response = llm_with_tools.invoke(messages)
            messages.append(response)
            
        return response.content
        
    return run

4. 100-Point Grading Rubric

CriterionPointsExcellent (25 pts)Acceptable (15 pts)Needs Work (5 pts)
Pydantic Schema Design25 ptsSchemas use Pydantic v2 with exhaustive field docstrings, type constraints, and defaults.Basic types without descriptions or error traps.Missing schemas; functions use unstructured args.
Tool Binding & Dispatch25 ptsCorrectly binds tools using bind_tools() and iterates through tool_calls cleanly.Tool calls execute but lack handling for multiple concurrent calls.Manual string parsing instead of native tool binding.
Error Handling & Retries25 ptsCatches tool failures, returns structured ToolMessage(status='error'), and prompts agent recovery.Tool errors print to console but crash the agent loop.Unhandled exceptions cause runtime process exit.
Synthesis & Citations25 ptsOutput summarizes financial data accurately and cites sources and fiscal periods clearly.Accurate response but omits fiscal period citations.Hallucinates numbers not returned by the tools.
Checking your access…