Интеграция внешних инструментов: tool use, структурированные выводы, API integration, multi-tool workflows
Интеграция внешних инструментов с LLM: API, базы данных, калькуляторы и другие сервисы
Function Calling (вызов функций) — это возможность LLM генерировать структурированный вызов функции вместо свободного текста. Это позволяет модели взаимодействовать с внешними инструментами: API, базами данных, файловыми системами, калькуляторами и другими сервисами.
Зачем это нужно:
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ User │────▶│ LLM │────▶│ Your Code │
│ Question │ │ + Tools │ │ (Function) │
└─────────────┘ └──────────────┘ └─────────────┘
│ │
│ Function Call │
│ {name, args} │
│◀───────────────────┘
│
│ Function Result
│◀────────────────────
│
▼
Final Response
from openai import OpenAI
import json
client = OpenAI()
# Определение функций
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Получить текущую погоду в указанном городе",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "Город, например: Москва, London"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "Единица измерения температуры"
}
},
"required": ["location"]
}
}
}
]
# Запрос к модели
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "user", "content": "Какая погода в Москве?"}
],
tools=tools
)
# Проверка наличия вызова функции
if response.choices[0].message.tool_calls:
tool_call = response.choices[0].message.tool_calls[0]
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
print(f"Вызов функции: {function_name}")
print(f"Аргументы: {function_args}")
# Вызов функции: get_weather
# Аргументы: {"location": "Москва", "unit": "celsius"}def get_weather(location: str, unit: str = "celsius") -> str:
# Реальная реализация через API
# Для примера — заглушка
return f"В {location} сейчас 20°C, ясно"
# Полный цикл
def chat_with_tools(user_message: str) -> str:
# Шаг 1: Запрос к модели
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": user_message}],
tools=tools
)
message = response.choices[0].message
# Шаг 2: Проверка на вызов функции
if message.tool_calls:
tool_call = message.tool_calls[0]
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
# Шаг 3: Выполнение функции
if function_name == "get_weather":
result = get_weather(**function_args)
# Шаг 4: Возврат результата модели
second_response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "user", "content": user_message},
{"role": "assistant", "tool_calls": [tool_call]},
{"role": "tool", "tool_call_id": tool_call.id, "content": result}
]
)
return second_response.choices[0].message.content
return message.content
# Использование
answer = chat_with_tools("Какая погода в Москве?")
print(answer)
# В Москве сейчас 20°C, ясно. Одежда для тёплой погоды подойдёт.tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Получить текущую погоду в указанном городе",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
},
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "Получить текущую цену акции компании",
"parameters": {
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Тикер акции, например: AAPL, GOOGL"
}
},
"required": ["symbol"]
}
}
},
{
"type": "function",
"function": {
"name": "search_database",
"description": "Поиск пользователей в базе данных",
"parameters": {
"type": "object",
"properties": {
"email": {"type": "string"},
"name": {"type": "string"},
"limit": {"type": "integer", "default": 10}
}
}
}
}
]def execute_function_call(tool_call) -> str:
"""Выполняет вызов функции и возвращает результат"""
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
functions = {
"get_weather": get_weather,
"get_stock_price": get_stock_price,
"search_database": search_database
}
if function_name in functions:
return functions[function_name](**function_args)
else:
return f"Ошибка: функция {function_name} не найдена"
def process_tool_calls(messages: list) -> str:
"""Обрабатывает все tool calls в сообщении"""
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools
)
message = response.choices[0].message
tool_calls = message.tool_calls
# Если есть вызовы функций
if tool_calls:
# Добавляем сообщение ассистента с вызовами
messages.append(message)
# Выполняем каждую функцию и добавляем результаты
for tool_call in tool_calls:
result = execute_function_call(tool_call)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# Рекурсивный вызов для получения финального ответа
return process_tool_calls(messages)
return message.contentМодель может генерировать несколько вызовов функций одновременно, если это эффективно для решения задачи.
user_message = "Сравни погоду в Москве и Санкт-Петербурге"
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": user_message}],
tools=tools
)
# Модель может сгенерировать два параллельных вызова:
# tool_calls = [
# {name: "get_weather", args: {location: "Москва"}},
# {name: "get_weather", args: {location: "Санкт-Петербург"}}
# ]
# Выполняем параллельно
import asyncio
async def execute_parallel(tool_calls):
tasks = [execute_function_call(tc) for tc in tool_calls]
results = await asyncio.gather(*tasks)
return resultsFunction Calling также используется для получения структурированного вывода (JSON schema).
tools = [
{
"type": "function",
"function": {
"name": "extract_user_info",
"description": "Извлечь информацию о пользователе из текста",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string"},
"email": {"type": "string"},
"age": {"type": "integer"},
"city": {"type": "string"}
},
"required": ["name", "email"]
}
}
}
]
text = "Меня зовут Иван Петров, мне 28 лет. Живу в Москве. Мой email: ivan@example.com"
response = client.chat.completions.create(
model="gpt-4",
messages=[{
"role": "user",
"content": f"Извлеки информацию из текста: {text}"
}],
tools=tools,
tool_choice={"type": "function", "function": {"name": "extract_user_info"}}
)
tool_call = response.choices[0].message.tool_calls[0]
user_data = json.loads(tool_call.function.arguments)
print(user_data)
# {
# "name": "Иван Петров",
# "email": "ivan@example.com",
# "age": 28,
# "city": "Москва"
# }from pydantic import BaseModel
class Userinfo(BaseModel):
name: str
email: str
age: int | None = None
city: str | None = None
response = client.beta.chat.completions.parse(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"Извлеки информацию: {text}"
}],
response_format=Userinfo
)
user_info = response.choices[0].message.parsed
# user_info — это инстанс Pydantic моделиfrom fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
import httpx
app = FastAPI()
# Модели запросов/ответов
class ChatRequest(BaseModel):
message: str
conversation_history: List[dict] = []
class ToolResponse(BaseModel):
response: str
tool_calls: Optional[List[dict]] = None
# Реальные инструменты
class WeatherAPI:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.weather.com"
async def get_weather(self, location: str, unit: str = "celsius") -> str:
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{self.base_url}/current",
params={"q": location, "key": self.api_key}
)
data = resp.json()
temp = data["temp_c"] if unit == "celsius" else data["temp_f"]
return f"В {location} сейчас {temp}°, {data['condition']}"
class StockAPI:
async def get_stock_price(self, symbol: str) -> str:
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.alphavantage.co/query",
params={"function": "GLOBAL_QUOTE", "symbol": symbol}
)
data = resp.json()
price = data["Global Quote"]["05. price"]
return f"${symbol}: ${price}"
# Глобальные инструменты
weather_api = WeatherAPI(api_key="your_key")
stock_api = StockAPI()
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Получить текущую погоду в указанном городе",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
},
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "Получить текущую цену акции",
"parameters": {
"type": "object",
"properties": {
"symbol": {"type": "string"}
},
"required": ["symbol"]
}
}
}
]
TOOLS_MAP = {
"get_weather": weather_api.get_weather,
"get_stock_price": stock_api.get_stock_price
}
@app.post("/chat", response_model=ToolResponse)
async def chat(request: ChatRequest):
messages = request.conversation_history + [
{"role": "user", "content": request.message}
]
# Запрос к OpenAI
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=TOOLS
)
message = response.choices[0].message
# Если есть вызовы функций
if message.tool_calls:
messages.append(message)
# Выполняем каждую функцию
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
if function_name in TOOLS_MAP:
result = await TOOLS_MAP[function_name](**function_args)
else:
result = f"Ошибка: функция {function_name} не найдена"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# Получаем финальный ответ
final_response = client.chat.completions.create(
model="gpt-4",
messages=messages
)
return ToolResponse(
response=final_response.choices[0].message.content,
tool_calls=[tc.model_dump() for tc in message.tool_calls]
)
return ToolResponse(response=message.content)from fastapi.responses import StreamingResponse
import json
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
async def generate():
messages = request.conversation_history + [
{"role": "user", "content": request.message}
]
stream = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=TOOLS,
stream=True
)
tool_calls_buffer = {}
for chunk in stream:
delta = chunk.choices[0].delta
# Обработка tool calls
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if idx not in tool_calls_buffer:
tool_calls_buffer[idx] = {
"name": tc.function.name if tc.function else "",
"arguments": ""
}
if tc.function and tc.function.arguments:
tool_calls_buffer[idx]["arguments"] += tc.function.arguments
# Отправка контента
if delta.content:
yield f"data: {json.dumps({'type': 'content', 'data': delta.content})}\n\n"
# Если были tool calls, выполняем их
if tool_calls_buffer:
# ... логика выполнения функций ...
yield f"data: {json.dumps({'type': 'tool_calls', 'data': list(tool_calls_buffer.values())})}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")from ollama import chat
from pydantic import BaseModel
class GetWeather(BaseModel):
location: str
unit: str = "celsius"
response = chat(
model='llama3.1',
messages=[{'role': 'user', 'content': 'Какая погода в Москве?'}],
tools=[{
'type': 'function',
'function': {
'name': 'get_weather',
'description': 'Получить погоду в городе',
'parameters': GetWeather.model_json_schema()
}
}]
)
# Проверка на вызов функции
if response.message.tool_calls:
tool_call = response.message.tool_calls[0]
print(tool_call.function)import llama_cpp_python as llama
llm = llama.Llama(
model_path="./models/llama-3.1-8b.Q4_K_M.gguf",
n_ctx=4096
)
output = llm.create_chat_completion(
messages=[
{"role": "user", "content": "Какая погода в Москве?"}
],
tools=[{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"}
},
"required": ["location"]
}
}
}]
)
tool_call = output["choices"][0]["message"]["tool_calls"][0]# Плохо
{
"name": "get_weather",
"description": "Получить погоду"
}
# Хорошо
{
"name": "get_weather",
"description": "Получить текущую погоду и прогноз на 3 дня для указанного города. "
"Используйте эту функцию, когда пользователь спрашивает о погоде, "
"температуре, осадках, ветре.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "Название города на английском или русском языке"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "Единица измерения температуры. По умолчанию celsius."
}
},
"required": ["location"]
}
}async def safe_execute_function(tool_call) -> str:
try:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
if function_name not in TOOLS_MAP:
return f"Ошибка: функция '{function_name}' не найдена"
result = await TOOLS_MAP[function_name](**function_args)
return result
except json.JSONDecodeError:
return "Ошибка: некорректные аргументы функции"
except TypeError as e:
return f"Ошибка: неверные параметры функции — {str(e)}"
except Exception as e:
return f"Ошибка выполнения: {str(e)}"MAX_TOOL_CALLS = 5
async def process_with_limit(messages: list, max_calls: int = MAX_TOOL_CALLS):
call_count = 0
while call_count < max_calls:
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=TOOLS
)
message = response.choices[0].message
if not message.tool_calls:
return message.content
messages.append(message)
for tool_call in message.tool_calls:
result = await safe_execute_function(tool_call)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
call_count += 1
return "Превышено максимальное количество вызовов функций"import logging
from datetime import datetime
logger = logging.getLogger("function_calling")
async def logged_execute_function(tool_call) -> str:
start_time = datetime.now()
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
logger.info(f"Вызов функции: {function_name}, args: {function_args}")
try:
result = await TOOLS_MAP[function_name](**function_args)
duration = (datetime.now() - start_time).total_seconds()
logger.info(f"Функция {function_name} выполнена за {duration:.2f}s")
return result
except Exception as e:
logger.error(f"Ошибка функции {function_name}: {e}")
raiseДалее: Fine-Tuning