Agent Pipelines

import json
from openai import OpenAI

client = OpenAI()

tools = [
    {"type": "function", "function": {
        "name": "search_web",
        "description": "Search the web for current information",
        "parameters": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    }},
    {"type": "function", "function": {
        "name": "calculate",
        "description": "Evaluate a mathematical expression",
        "parameters": {
            "type": "object",
            "properties": {"expression": {"type": "string"}},
            "required": ["expression"]
        }
    }}
]

def execute_tool(name: str, args: dict) -> str:
    if name == "search_web":
        return search_api(args["query"])  # Your search implementation
    elif name == "calculate":
        return str(eval(args["expression"]))  # Use sandboxed eval in prod
    return f"Unknown tool: {name}"

def run_agent(task: str, max_turns: int = 10) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful research assistant."},
        {"role": "user", "content": task}
    ]

    for turn in range(max_turns):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools
        )
        msg = response.choices[0].message

        # Agent decided to return a final answer
        if response.choices[0].finish_reason == "stop":
            return msg.content

        # Agent wants to call tools
        if msg.tool_calls:
            messages.append(msg)
            for tc in msg.tool_calls:
                result = execute_tool(
                    tc.function.name,
                    json.loads(tc.function.arguments)
                )
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result
                })

    return "Agent reached maximum turns without completing."

import anthropic

client = anthropic.Anthropic()

tools = [
    {
        "name": "search_web",
        "description": "Search the web for current information",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    },
    {
        "name": "calculate",
        "description": "Evaluate a mathematical expression",
        "input_schema": {
            "type": "object",
            "properties": {"expression": {"type": "string"}},
            "required": ["expression"]
        }
    }
]

def run_agent(task: str, max_turns: int = 10) -> str:
    messages = [{"role": "user", "content": task}]

    for turn in range(max_turns):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system="You are a helpful research assistant.",
            tools=tools,
            messages=messages
        )

        # Check if agent is done (no tool use)
        if response.stop_reason == "end_turn":
            return response.content[0].text

        # Process tool calls
        if response.stop_reason == "tool_use":
            messages.append({
                "role": "assistant",
                "content": response.content
            })

            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            messages.append({"role": "user", "content": tool_results})

    return "Agent reached maximum turns."

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

@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    return search_api(query)

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

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

model = ChatOpenAI(model="gpt-4o").bind_tools(
    [search_web, calculate]
)

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

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

# Wire it up
graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode([search_web, calculate]))
graph.set_entry_point("agent")
graph.add_conditional_edges(
    "agent", should_continue,
    {"tools": "tools", END: END}
)
graph.add_edge("tools", "agent")

agent = graph.compile()
result = agent.invoke({
    "messages": [{"role": "user", "content": "NVIDIA market cap?"}]
})