Agent Libraries: LangChain & Framework Deep Dive

from langchain_core.tools import tool
from typing import Annotated

@tool
def calculate(expression: Annotated[str, "A mathematical expression to evaluate"]) -> str:
    """Evaluate a mathematical expression and return the result.
    Use this for any arithmetic or math questions."""
    try:
        result = eval(expression)  # In production, use a safe math parser
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@tool
def get_weather(
    city: Annotated[str, "The city name"],
    unit: Annotated[str, "Temperature unit: celsius or fahrenheit"] = "celsius"
) -> str:
    """Get the current weather for a city.
    Use this when the user asks about weather conditions."""
    # In production, this would call a real weather API
    return f"The weather in {city} is 22 degrees {unit} and sunny."

# Inspect the tool's schema (this is what the model sees)
print(calculate.name)         # "calculate"
print(calculate.description)  # "Evaluate a mathematical expression..."
print(calculate.args_schema.schema())  # JSON schema for arguments

from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

# Web search via DuckDuckGo (no API key required)
search_tool = DuckDuckGoSearchRun()
result = search_tool.invoke("LangChain framework latest version 2025")
print(result[:200])

# Wikipedia lookup
wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(top_k_results=1))
result = wiki_tool.invoke("Transformer neural network architecture")
print(result[:200])

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Define the prompt with a placeholder for agent scratchpad
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to tools. "
               "Use them when needed to answer questions accurately."),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# Initialize model and tools
model = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [calculate, get_weather]

# Create the agent
agent = create_tool_calling_agent(model, tools, prompt)

# Wrap in AgentExecutor to handle the tool-call loop
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,    # Print each step for debugging
    max_iterations=5  # Safety limit on tool-call loops
)

# Run the agent
result = executor.invoke({
    "input": "What is the weather in Paris? Also, what is 15% of 847?"
})
print(result["output"])

from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict

class LoggingHandler(BaseCallbackHandler):
    """Logs every LLM call and tool invocation."""

    def on_llm_start(self, serialized: Dict[str, Any], prompts: list, **kwargs):
        print(f"[LLM START] Model: {serialized.get('id', ['unknown'])}")

    def on_llm_end(self, response, **kwargs):
        print(f"[LLM END] Tokens used: {response.llm_output}")

    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs):
        print(f"[TOOL START] {serialized.get('name', 'unknown')}: {input_str}")

    def on_tool_end(self, output: str, **kwargs):
        print(f"[TOOL END] Result: {output[:100]}")

    def on_agent_action(self, action, **kwargs):
        print(f"[AGENT ACTION] {action.tool}: {action.tool_input}")

    def on_agent_finish(self, finish, **kwargs):
        print(f"[AGENT FINISH] {finish.return_values}")

# Attach the handler to the executor
result = executor.invoke(
    {"input": "What is 2^10?"},
    config={"callbacks": [LoggingHandler()]}
)

import os

# Enable LangSmith tracing (set these before importing LangChain)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."      # Your LangSmith API key
os.environ["LANGCHAIN_PROJECT"] = "my-agent-app"  # Project name in LangSmith

# All subsequent chain/agent invocations are automatically traced.
# View traces at https://smith.langchain.com

# You can also add custom metadata to traces
result = executor.invoke(
    {"input": "Summarize the weather in Tokyo"},
    config={
        "metadata": {"user_id": "user-456", "request_id": "req-789"},
        "tags": ["production", "weather-agent"]
    }
)

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

model = ChatOpenAI(model="gpt-4o", temperature=0)
model_with_tools = model.bind_tools([calculate, get_weather])

def run_agent(user_input: str, max_steps: int = 5) -> str:
    """A custom agent loop with full control over execution."""
    messages = [HumanMessage(content=user_input)]
    tools_map = {"calculate": calculate, "get_weather": get_weather}

    for step in range(max_steps):
        # Ask the model
        response = model_with_tools.invoke(messages)
        messages.append(response)

        # If no tool calls, we have the final answer
        if not response.tool_calls:
            return response.content

        # Execute each tool call
        for tool_call in response.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]

            print(f"Step {step + 1}: Calling {tool_name}({tool_args})")

            # Optional: add approval logic here
            # if tool_name == "dangerous_tool":
            #     approval = input("Approve? (y/n): ")
            #     if approval != "y":
            #         continue

            # Execute the tool
            tool_fn = tools_map[tool_name]
            result = tool_fn.invoke(tool_args)

            # Add the tool result as a ToolMessage
            messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"]
            ))

    return "Agent reached maximum steps without a final answer."

# Run the custom agent
answer = run_agent("What is the weather in London and what is 42 * 17?")
print(answer)

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

llm = ChatOpenAI(model="gpt-4o").bind_tools([search_tool, calculator_tool])

def call_model(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState):
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return END

graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode([search_tool, calculator_tool]))
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile()
result = app.invoke({"messages": [HumanMessage("What is 15% of the GDP of France?")]})

from crewai import Agent, Task, Crew, LLM

llm = LLM(model="gpt-4o")

researcher = Agent(
    role="Senior Research Analyst",
    goal="Find comprehensive data on the given topic",
    backstory="You are an expert researcher with 20 years of experience.",
    tools=[search_tool, web_scraper],
    llm=llm,
)

writer = Agent(
    role="Technical Writer",
    goal="Create a clear, well-structured report from research findings",
    backstory="You specialize in making complex topics accessible.",
    llm=llm,
)

research_task = Task(
    description="Research the current state of LLM inference optimization.",
    expected_output="A detailed list of findings with sources.",
    agent=researcher,
)

writing_task = Task(
    description="Write a 1000-word report based on the research findings.",
    expected_output="A polished report in markdown format.",
    agent=writer,
)

crew = Crew(agents=[researcher, writer], tasks=[research_task, writing_task], verbose=True)
result = crew.kickoff()

import autogen

coder = autogen.AssistantAgent(name="coder", llm_config={"model": "gpt-4o"})
user = autogen.UserProxyAgent(name="user", human_input_mode="NEVER",
                              code_execution_config={"work_dir": "out"})
chat = autogen.GroupChat(agents=[user, coder], messages=[], max_round=8)
manager = autogen.GroupChatManager(groupchat=chat,
                                   llm_config={"model": "gpt-4o"})
user.initiate_chat(manager, message="Plot prime gaps below 1000 with matplotlib.")

Code Fragment 30.2.2b: An AutoGen group chat with a Coder and a User Proxy. The GroupChatManager is itself an LLM-driven router that picks the next speaker on every turn, the topology drawn on the right side of Figure 30.2.f1.

from agents import Agent, Runner, function_tool

@function_tool
def search(query: str) -> str:
    """Search the web for information."""
    return web_search(query)

@function_tool
def calculate(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return eval(expression)

agent = Agent(
    name="Research Assistant",
    instructions="You help users find information and perform calculations.",
    tools=[search, calculate],
    model="gpt-4o",
)

result = Runner.run_sync(agent, "What is 15% of the GDP of France?")

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class CountState(TypedDict):
    value: int

def increment(state: CountState) -> dict:
    return {"value": state["value"] + 1}

def double(state: CountState) -> dict:
    return {"value": state["value"] * 2}

graph = StateGraph(CountState)
graph.add_node("inc", increment)
graph.add_node("dbl", double)
graph.add_edge(START, "inc")
graph.add_edge("inc", "dbl")
graph.add_edge("dbl", END)

app = graph.compile()
print(app.invoke({"value": 3}))   # {'value': 8}: (3+1)*2

from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage
from typing import Annotated, TypedDict

class ChatState(TypedDict):
    # The reducer is the second tuple element. add_messages appends new
    # messages instead of overwriting the list and dedupes by message id.
    messages: Annotated[list[AnyMessage], add_messages]

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o")

def chat_node(state: ChatState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}   # the reducer appends, not replaces

graph = StateGraph(ChatState)
graph.add_node("chat", chat_node)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)

app = graph.compile()
out = app.invoke({"messages": [HumanMessage("What is the boiling point of water?")]})
print(out["messages"][-1].content)

from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Return the current weather for a city."""
    return f"{city}: 18C, partly cloudy"

@tool
def calculator(expression: str) -> str:
    """Evaluate a simple arithmetic expression."""
    return str(eval(expression))   # toy: in production use a safe parser

tools = [get_weather, calculator]
llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools(tools)

def chat_node(state: ChatState) -> dict:
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph = StateGraph(ChatState)
graph.add_node("chat", chat_node)
graph.add_node("tools", ToolNode(tools))          # the reserved TOOLS node
graph.add_edge(START, "chat")
graph.add_conditional_edges("chat", tools_condition)  # routes to "tools" or END
graph.add_edge("tools", "chat")                   # close the ReAct loop

app = graph.compile()
out = app.invoke({"messages": [HumanMessage(
    "What is the weather in Paris, and what is 23 * 17?")]})
for m in out["messages"]:
    print(type(m).__name__, getattr(m, "content", "")[:80])

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# One thread (one conversation), two turns: the second sees the first.
config = {"configurable": {"thread_id": "session-42"}}
app.invoke({"messages": [HumanMessage("My name is Yossi.")]}, config=config)
out = app.invoke({"messages": [HumanMessage("What is my name?")]}, config=config)
print(out["messages"][-1].content)   # "Your name is Yossi."

# Inspect the thread: a list of checkpoints, newest first.
for ckpt in app.get_state_history(config):
    print(ckpt.config["configurable"]["checkpoint_id"], len(ckpt.values["messages"]))

# Resume from step 5: pick a past checkpoint and continue from it.
old_config = list(app.get_state_history(config))[4].config
app.invoke(None, config=old_config)   # replays from that checkpoint forward

# Full state after every reducer (UI-friendly, verbose).
for event in app.stream({"messages": [HumanMessage("Weather in Tokyo?")]},
                        config=config, stream_mode="values"):
    print("STATE", len(event["messages"]), "msgs")

# Just the per-node update (logs / tracing-friendly, compact).
for event in app.stream({"messages": [HumanMessage("And in Lima?")]},
                        config=config, stream_mode="updates"):
    # event is e.g. {"chat": {"messages": [AIMessage(...)]}}
    node_name = next(iter(event))
    print("UPDATE from", node_name)

app = graph.compile(checkpointer=memory, interrupt_before=["tools"])
config = {"configurable": {"thread_id": "approval-1"}}

# Step 1: run until just before any tool call.
app.invoke({"messages": [HumanMessage(
    "Send a thank-you email to alice@example.com.")]}, config=config)
pending = app.get_state(config)
print("Proposed action:", pending.values["messages"][-1].tool_calls)

# Step 2 (human review): operator edits the args, then resumes.
edited = pending.values["messages"][-1]
edited.tool_calls[0]["args"]["body"] = "Thanks, Alice. From the LangGraph agent."
app.update_state(config, {"messages": [edited]})

# Step 3: resume from the saved checkpoint; the tool now executes.
final = app.invoke(None, config=config)
print(final["messages"][-1].content)