Parallel Tool Use في Claude API
المستوى المطلوب: محترف — تحتاج خبرة سابقة بـ Tool Use الأساسي وبـ async/await في Python.
لو الـ agent بتاعك بياخد 6 ثواني علشان يرجّع dashboard فيه طقس + سعر سهم + آخر إيميل + رصيد حساب + موعد اجتماع، 5 ثواني منهم latency شبكة هتختفي بسطر واحد. Parallel Tool Use بيخلّي Claude يطلب الـ 5 أدوات في رد واحد، وأنت بتنفّذهم بـ asyncio.gather، فالزمن الإجمالي بيساوي أبطأ أداة، مش مجموع الأدوات.
المشكلة باختصار
الـ Tool Use التقليدي بيمشي كده: الموديل بيرد بـ tool_use واحد، أنت بتنفّذه، ترجع النتيجة، الموديل يقرر هل يطلب أداة تانية، وهكذا. كل خطوة فيها round-trip للـ API ≈ 600-900ms + زمن تنفيذ الأداة. خمس أدوات = خمس رحلات = 4-6 ثواني قبل ما المستخدم يشوف حاجة. ده مش مقبول في chat UX.
مثال للمبتدئ: طلب الكافيه
تخيّل إنك في كافيه وطلبت من الويتر: قهوة + كرواسون + كوب مياه. لو الويتر راح جاب القهوة أول، رجعهالك، ثم راح للكرواسون، ثم للمياه — هتستنى 6 دقايق. لو راح ساب الطلب لـ 3 موظفين في الباريستا والمخبز والثلاجة في نفس الوقت — كل حاجة بتوصلك في 2 دقيقة. Claude زي الويتر، والأدوات زي الموظفين. Parallel Tool Use ببساطة بيقول للويتر: "وزّع الطلب، متمشيش لوحدك".
التعريف العلمي الدقيق
منذ Claude 3.5 Sonnet (يونيو 2024) والـ stop_reason أصبح يدعم رد فيه أكثر من tool_use block في نفس الـ assistant turn. الـ schema بيرجع array من content blocks، كل block ليه id فريد. أنت كـ orchestrator مسؤول عن:
- parsing كل blocks ذات النوع
tool_use. - تنفيذها بشكل متوازٍ (asyncio / threading / queue).
- إرجاع
tool_resultblocks بنفس عدد الـ tool_use blocks وبنفس الـtool_use_id.
الباراميتر disable_parallel_tool_use داخل tool_choice بيتحكم في السلوك: false (الافتراضي على Claude 4.x) يسمح بالتوازي، true يفرض tool واحد لكل turn.
الكود الشغّال — Python 3.11 + Anthropic SDK
import anthropic
import asyncio
import time
client = anthropic.AsyncAnthropic()
TOOLS = [
{"name": "get_weather", "description": "طقس مدينة",
"input_schema": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}},
{"name": "get_stock", "description": "سعر سهم",
"input_schema": {"type": "object", "properties": {"ticker": {"type": "string"}}, "required": ["ticker"]}},
{"name": "get_calendar", "description": "اجتماعات اليوم",
"input_schema": {"type": "object", "properties": {"user_id": {"type": "string"}}, "required": ["user_id"]}},
]
async def execute_tool(name, args):
await asyncio.sleep(0.9) # محاكاة latency
return {"get_weather": "27°C", "get_stock": 184.2, "get_calendar": ["10:00 standup"]}[name]
async def run_agent(user_msg):
msgs = [{"role": "user", "content": user_msg}]
while True:
resp = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
tools=TOOLS,
messages=msgs,
)
if resp.stop_reason != "tool_use":
return resp
tool_uses = [b for b in resp.content if b.type == "tool_use"]
results = await asyncio.gather(
*[execute_tool(b.name, b.input) for b in tool_uses],
return_exceptions=True,
)
msgs.append({"role": "assistant", "content": resp.content})
msgs.append({"role": "user", "content": [
{"type": "tool_result", "tool_use_id": b.id,
"content": str(r), "is_error": isinstance(r, Exception)}
for b, r in zip(tool_uses, results)
]})
start = time.perf_counter()
asyncio.run(run_agent("جيبلي الطقس في القاهرة + سعر AAPL + اجتماعاتي"))
print(f"إجمالي: {time.perf_counter()-start:.2f}s")
