Spaces:
Paused
Paused
File size: 4,801 Bytes
0d3f7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """Toolformer reasoning pattern implementation.
Toolformer: Self-supervised tool usage learning
- The model learns to call APIs by generating [API_CALL(...)] tokens inline
- After the API call, the model observes the result and continues generation
- This pattern teaches the model when and how to appropriately use tools
"""
from __future__ import annotations
import logging
import re
from typing import Any
from hermes.core.types import AgentStrategy
logger = logging.getLogger(__name__)
class ToolformerReasoner:
"""Implements the Toolformer pattern for self-supervised tool use."""
def __init__(self) -> None:
self.strategy = AgentStrategy.TOOLFORMER
def create_prompt(
self, task: str, tools: list[dict[str, Any]], history: list[dict[str, str]] | None = None
) -> str:
"""Create Toolformer prompt with inline API call format."""
tool_descriptions = "\n".join(
f"- {t['name']}: {t['description']}" for t in tools
)
history_text = ""
if history:
history_text = "\nPrevious context:\n"
for h in history:
if "text" in h:
history_text += f"{h['text']}\n"
return f"""You are an AI agent that uses tools by making inline API calls.
Task: {task}
Available tools:
{tool_descriptions}
{history_text}
To use a tool, embed an API call directly in your text using this format:
[API_CALL: tool_name(arg1="value1", arg2="value2")]
The tool result will be provided, and you can continue your response.
Example:
I need to find the latest news about AI.
[API_CALL: search_web(query="latest AI news 2026")]
Now I have the search results, I can answer the question: ...
Rules:
- Only call one API at a time
- Wait for the result before continuing
- Use the tool output to inform your response
- If you don't need any tools, just answer directly"""
def parse_response(self, response: str) -> dict[str, Any]:
"""Parse Toolformer response, extracting API calls."""
result: dict[str, Any] = {
"text": response,
"api_calls": [],
"has_api_call": False,
"final_answer": None,
}
api_call_pattern = re.compile(
r'\[API_CALL:\s*(\w+)\s*\(([^)]*)\)\s*\]'
)
matches = api_call_pattern.findall(response)
for tool_name, args_str in matches:
try:
args = self._parse_args(args_str)
result["api_calls"].append({
"tool": tool_name,
"arguments": args,
})
except Exception as e:
logger.warning(f"Failed to parse API call: {e}")
result["has_api_call"] = len(result["api_calls"]) > 0
if not result["has_api_call"]:
result["final_answer"] = response.strip()
return result
def create_observation_prompt(
self, original: str, api_call: dict[str, Any], observation: str
) -> str:
"""Create prompt to continue generation after tool observation."""
return f"""Continue your response after receiving the tool result.
Your original text:
{original}
You called:
[API_CALL: {api_call['tool']}({api_call['arguments']})]
Tool result:
{observation}
Now incorporate this result into your response and continue.
If you need more information, you can make another API call.
If you have enough information, provide your final answer."""
def parse_observation_response(self, response: str) -> dict[str, Any]:
"""Parse the continued response after an observation."""
result: dict[str, Any] = {"text": response, "has_more_calls": False, "final_answer": None}
if "[API_CALL:" in response:
result["has_more_calls"] = True
parsed = self.parse_response(response)
result["api_calls"] = parsed.get("api_calls", [])
else:
result["final_answer"] = response.strip()
return result
def should_continue(self, parsed: dict[str, Any], max_calls: int, call_count: int) -> bool:
"""Determine if the agent should make more API calls."""
if call_count >= max_calls:
return False
if parsed.get("final_answer"):
return False
return not (not parsed.get("has_api_call") and not parsed.get("has_more_calls"))
def _parse_args(self, args_str: str) -> dict[str, Any]:
"""Parse argument string into a dictionary."""
args: dict[str, Any] = {}
if not args_str.strip():
return args
parts = re.findall(r'(\w+)\s*=\s*("[^"]*"|\'[^\']*\'|\S+)', args_str)
for key, value in parts:
cleaned = value.strip('"').strip("'")
args[key] = cleaned
return args
|