День 4: Tool Calling и Structured Outputs

Цель: Научить агентов использовать инструменты и получать структурированные ответы

Время изучения: 2-3 часа
Код: src/day4_tools_structured.py
Тесты: tests/test_day4_tools_structured.py

🤖 Связь с агентами

Вчера: Создали систему Planner → Executor → Critic
Сегодня: Даём агентам инструменты — они становятся по-настоящему полезными!

Что мы строим?

В E Э " ч x м C е e у o р c л m а u ↓ и p : t р l o у e r е t т e A d g в : e ы n п S t о t л e н p е н 1 и " е Р - - - С E е е x а П В Р г e л о ы а о c ь и ч б д u ↓ н с и о н t о к с т я o л а : r в в е ы н с A п и и g о н я ф e л т а n н е й t я р л е н а + т е м : т и T е o o l s

Зачем это нужно?

Реальный сценарий: Исследовательский агент

З P E C а l x r д a e - - - i а n c t ч n u И И И i а e t с с с c : r o п п п : : r о о о " : л л л " Н " ь ь ь О а 1 з з з т й . у у у л д е е е и и П т т т ч о н и и t t t о н с o o o , ф к o o o о l l l д р в а м " " " н а и s e c н ц н e x a ы и т a t l е ю е r r c р c a u п о н h c l о е " t a л L т _ t у a е → d e ч n , a " е g н t н G 2 а a → ы r . х " ! a о с " p И д → у h з и м в т п м и л о и е G л р п ч i у у о ь t ч е с H а т ч д u е и а b т д т н а а н р к н й ы е о н е п л ы к , о и е о з ч л 3 и е и . т с ч о т е П р в с о и о т с й в ч з о и в т ё з а з в т д ё ь з " д н а G i t H u b "

Tools = способность агента взаимодействовать с внешним миром!

Часть 1: Что такое Tools?

Определение

Tool (инструмент) — это функция, которую агент может вызвать для выполнения действия.

def search_tool(query: str) -> str:
    """Поиск информации в интернете."""
    # В реальности: вызов API поиска
    return f"Search results for: {query}"

def calculator_tool(expression: str) -> float:
    """Вычисление матем��тического выражения."""
    # В реальности: безопасное вычисление
    return eval(expression)

def read_file_tool(filepath: str) -> str:
    """Чтение содержимого файла."""
    with open(filepath, 'r') as f:
        return f.read()

Как агент выбирает tool?

А г е н т п о л у ч а е т з а д а ч у → L L M а н а л и з и р у е т → В ы б и р а е т н у ж н ы й t o o l → В ы з ы в а е т → П о л у ч а е т р е з у л ь т а т

Часть 2: Создание Tools

Простые инструменты

from typing import Dict, Any

# Tool 1: Поиск
def search_internet(query: str) -> str:
    """
    Поиск информации в интернете.
    
    Args:
        query: Поисковый запрос
        
    Returns:
        Результаты поиска
    """
    # Эмуляция поиска
    results = {
        "langgraph": "LangGraph is a library for building stateful, multi-actor applications with LLMs.",
        "python": "Python is a high-level programming language.",
        "ai": "Artificial Intelligence is the simulation of human intelligence."
    }
    
    for key in results:
        if key in query.lower():
            return results[key]
    
    return f"No results found for: {query}"


# Tool 2: Калькулятор
def calculator(expression: str) -> str:
    """
    Вычисление математического выражения.
    
    Args:
        expression: Математическое выражение (например, "2 + 2")
        
    Returns:
        Результат вычисления
    """
    try:
        # В реальности: использовать безопасный парсер
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {str(e)}"


# Tool 3: Работа с файлами
def read_file(filepath: str) -> str:
    """
    Чтение содержимого файла.
    
    Args:
        filepath: Путь к файлу
        
    Returns:
        Содержимое файла
    """
    try:
        with open(filepath, 'r') as f:
            content = f.read()
        return f"File content ({len(content)} chars): {content[:100]}..."
    except FileNotFoundError:
        return f"Error: File not found: {filepath}"
    except Exception as e:
        return f"Error: {str(e)}"

Реестр инструментов

# Словарь досту��ных инструментов
TOOLS = {
    "search": search_internet,
    "calculator": calculator,
    "read_file": read_file
}


def execute_tool(tool_name: str, **kwargs) -> str:
    """
    Выполнение инструмента по имени.
    
    Args:
        tool_name: Имя инструмента
        **kwargs: Аргументы для инструмента
        
    Returns:
        Результат выполнения
    """
    if tool_name not in TOOLS:
        return f"Error: Unknown tool '{tool_name}'"
    
    tool = TOOLS[tool_name]
    
    try:
        result = tool(**kwargs)
        return result
    except Exception as e:
        return f"Error executing {tool_name}: {str(e)}"

Часть 3: Structured Outputs

Зачем нужны?

Проблема: LLM возвращает текст, а нам нужна структура

# ❌ Неструктурированный ответ
response = "I will search for LangGraph and then calculate the sum"

# ✅ Структурированный ответ
response = {
    "steps": [
        {"action": "search", "query": "LangGraph"},
        {"action": "calculator", "expression": "10 + 20"}
    ]
}

Pydantic для валидации

from pydantic import BaseModel, Field
from typing import List, Literal


class ToolCall(BaseModel):
    """Вызов инструмента."""
    tool_name: Literal["search", "calculator", "read_file"]
    arguments: Dict[str, Any]
    reasoning: str = Field(description="Почему выбран этот инструмент")


class Plan(BaseModel):
    """Структурированный план."""
    task: str
    steps: List[str]
    estimated_time: int = Field(description="Оценка времени в минутах")


class ExecutionResult(BaseModel):
    """Результат выполнения."""
    success: bool
    result: str
    tool_used: str
    execution_time: float


class Critique(BaseModel):
    """Оценка критика."""
    quality_score: int = Field(ge=1, le=10, description="Оценка качества от 1 до 10")
    feedback: str
    should_retry: bool
    suggestions: List[str]

Часть 4: Executor с Tools

from typing import TypedDict, List, Optional

class AgentState(TypedDict, total=False):
    task: str
    plan: List[str]
    current_step: int
    step_result: str
    tools_used: List[str]  # 🆕 История использованных tools
    critique: str
    status: str
    iteration: int
    max_iterations: int
    final_result: str


def executor_agent_with_tools(state: AgentState) -> AgentState:
    """
    Executor Agent с поддержкой tools.
    
    В реальной системе здесь LLM выбирает tool и вызывает его.
    """
    updated = dict(state)
    plan = state.get("plan", [])
    current_step = state.get("current_step", 0)
    
    if current_step >= len(plan):
        updated["status"] = "done"
        updated["final_result"] = state.get("step_result", "Task completed")
        return updated
    
    step = plan[current_step]
    
    # Простая эмуляция выбора tool
    # В реальности: LLM анализирует шаг и выбирает tool
    tool_name = None
    tool_args = {}
    
    if "search" in step.lower() or "find" in step.lower():
        tool_name = "search"
        tool_args = {"query": step}
    elif "calculate" in step.lower() or "compute" in step.lower():
        tool_name = "calculator"
        tool_args = {"expression": "10 + 20"}  # Упрощение
    elif "read" in step.lower() or "file" in step.lower():
        tool_name = "read_file"
        tool_args = {"filepath": "example.txt"}
    
    # Выполняем tool
    if tool_name:
        result = execute_tool(tool_name, **tool_args)
        tools_used = updated.get("tools_used", [])
        tools_used.append(tool_name)
        updated["tools_used"] = tools_used
        print(f"   🔧 Used tool: {tool_name}")
    else:
        result = f"Completed: {step}"
    
    updated["step_result"] = result
    updated["status"] = "critiquing"
    updated["iteration"] = updated.get("iteration", 0) + 1
    
    print(f"\n⚙️  Executor: Executing step {current_step + 1}/{len(plan)}")
    print(f"   Step: {step}")
    print(f"   Result: {result}")
    
    return updated

Часть 5: Planner со Structured Output

def planner_agent_structured(state: AgentState) -> AgentState:
    """
    Planner Agent с структурированным выводом.
    
    В реальной системе здесь LLM возвращает Plan (Pydantic модель).
    """
    updated = dict(state)
    task = state.get("task", "")
    
    # Эмуляция структурированного планирования
    # В реальности: LLM возвращает Plan модель
    plan_obj = Plan(
        task=task,
        steps=[
            f"Search for information about '{task}'",
            f"Calculate statistics for '{task}'",
            f"Summarize findings for '{task}'"
        ],
        estimated_time=15
    )
    
    updated["plan"] = plan_obj.steps
    updated["current_step"] = 0
    updated["status"] = "executing"
    updated["iteration"] = updated.get("iteration", 0) + 1
    
    print(f"\n📋 Planner: Created structured plan")
    print(f"   Task: {plan_obj.task}")
    print(f"   Steps: {len(plan_obj.steps)}")
    print(f"   Estimated time: {plan_obj.estimated_time} minutes")
    for i, step in enumerate(plan_obj.steps, 1):
        print(f"   {i}. {step}")
    
    return updated

Полный пример

from langgraph.graph import StateGraph, END


def create_agent_with_tools():
    """Создаёт граф с поддержкой tools."""
    
    graph = StateGraph(AgentState)
    
    # Используем версии с tools
    graph.add_node("planner", planner_agent_structured)
    graph.add_node("executor", executor_agent_with_tools)
    graph.add_node("critic", critic_agent)
    
    graph.set_entry_point("planner")
    graph.add_edge("planner", "executor")
    graph.add_edge("executor", "critic")
    
    graph.add_conditional_edges(
        "critic",
        should_continue,
        {
            "execute": "executor",
            "critique": "critic",
            "end": END
        }
    )
    
    return graph.compile()


def main():
    """Пример использования агентов с tools."""
    
    workflow = create_agent_with_tools()
    
    initial_state: AgentState = {
        "task": "Research LangGraph and calculate usage statistics",
        "iteration": 0,
        "max_iterations": 10,
        "status": "planning",
        "tools_used": []
    }
    
    print("="*60)
    print("🚀 Agents with Tools")
    print("="*60)
    
    result = workflow.invoke(initial_state)
    
    print("\n" + "="*60)
    print("📊 Final Result")
    print("="*60)
    print(f"Task: {result.get('task')}")
    print(f"Tools used: {result.get('tools_used', [])}")
    print(f"Iterations: {result.get('iteration')}")
    print(f"Status: {result.get('status')}")
    print(f"Final result: {result.get('final_result')}")

Интеграция с реальным LLM

С LangChain

from langchain_openai import ChatOpenAI
from langchain.tools import Tool

# Определяем tools для LangChain
tools = [
    Tool(
        name="search",
        func=search_internet,
        description="Search for information on the internet"
    ),
    Tool(
        name="calculator",
        func=calculator,
        description="Calculate mathematical expressions"
    )
]

# LLM с tools
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)


def executor_with_real_llm(state: AgentState) -> AgentState:
    """Executor с реальным LLM."""
    step = plan[current_step]
    
    # LLM выбирает tool
    response = llm_with_tools.invoke(f"Execute this step: {step}")
    
    # Если LLM вызвал tool
    if response.tool_calls:
        tool_call = response.tool_calls[0]
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        
        # Выполняем tool
        result = execute_tool(tool_name, **tool_args)
    else:
        result = response.content
    
    # ... обновляем state

Ключевые выводы

✅ Tools — функции, которые агенты могут вызывать
✅ Реестр tools — словарь доступных инструментов
✅ Structured Outputs — Pydantic модели для валидации
✅ LLM выбирает tool — на основе анализа задачи
✅ История tools — отслеживание использованных инструментов

🤖 Связь с агентами

Сегодня вы научились:

  • ✅ Создавать инструменты для агентов
  • ✅ Агенты выбирают и используют tools
  • ✅ Структурировать ответы агентов (Pydantic)
  • ✅ Интегрировать с реальным LLM

Следующий шаг: Добавим память (Memory) — агенты будут помнить контекст и историю!

Следующий шаг: День 5

В следующей статье мы изучим:

  • Memory — агенты помнят контекст
  • Checkpointing — сохранение состояния
  • Финальный пайплайн — всё вместе

👉 День 5: Memory и финальный пайплайн (скоро)

Ресурсы

Прогресс: 4/10 дней ✅