在本綜合教程中，我們指導用戶使用Langgraph和Claude創建功能強大的多工具AI代理，並針對包括數學計算，Web搜索，天氣查詢，文本分析和實時信息檢索的各種任務進行了優化。它首先要簡化依賴安裝，以確保即使對於初學者，也可以輕鬆設置。然後，將用戶介紹了專用工具的結構化實現，例如安全計算器，有效的網絡搜索實用程序，利用DuckDuckgo，模擬天氣信息提供商，詳細的文本分析儀和時間提高功能。該教程還清楚地描述了這些工具在使用Langgraph構建的複雜代理體系結構中的集成，通過交互式示例和清晰的解釋說明了實際用法，從而促進了初學者和高級開發人員，以快速部署自定義的多功能AI代理。
import subprocess
import sys
def install_packages():
packages = (
"langgraph",
"langchain",
"langchain-anthropic",
"langchain-community",
"requests",
"python-dotenv",
"duckduckgo-search"
)
for package in packages:
try:
subprocess.check_call((sys.executable, "-m", "pip", "install", package, "-q"))
print(f"✓ Installed {package}")
except subprocess.CalledProcessError:
print(f"✗ Failed to install {package}")
print("Installing required packages...")
install_packages()
print("Installation complete!\n")我們自動化構建基於Langgraph的多工具AI代理所需的必需Python軟件包的安裝。它利用子過程默默地運行PIP命令，並確保成功安裝了從長鏈組件到Web搜索和環境處理工具的每個軟件包。此設置簡化了環境準備過程，使筆記本電腦便攜式和初學者友好。
import os
import json
import math
import requests
from typing import Dict, List, Any, Annotated, TypedDict
from datetime import datetime
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from duckduckgo_search import DDGS我們導入用於構建多工具AI代理的所有必要庫和模塊。它包括用於通用功能的Python標準庫，例如OS，JSON，MATH和DATETIME，以及外部庫，例如HTTP呼叫的請求以及用於實現Web搜索的DuckDuckgo_Search。 Langchain和Langgraph生態系統帶來消息類型，工具裝飾器，狀態圖組件和檢查點實用程序，而ChatanThropic則可以與Claude模型集成以進行對話智能。這些進口構成了定義工具，代理工作流程和交互的基礎構建塊。
os.environ("ANTHROPIC_API_KEY") = "Use Your API Key Here"
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")我們設置並檢索了對Claude模型進行身份驗證和交互所需的人類API密鑰。 OS.Environ系列分配了API密鑰（您應該用有效的密鑰替換），而OS.GetEnv則將其安全檢索以供以後在模型初始化中使用。此方法可確保在整個腳本中訪問密鑰，而無需多次進行硬編碼。
from typing import TypedDict
class AgentState(TypedDict):
messages: Annotated(List(BaseMessage), operator.add)
@tool
def calculator(expression: str) -> str:
"""
Perform mathematical calculations. Supports basic arithmetic, trigonometry, and more.
Args:
expression: Mathematical expression as a string (e.g., "2 + 3 * 4", "sin(3.14159/2)")
Returns:
Result of the calculation as a string
"""
try:
allowed_names = {
'abs': abs, 'round': round, 'min': min, 'max': max,
'sum': sum, 'pow': pow, 'sqrt': math.sqrt,
'sin': math.sin, 'cos': math.cos, 'tan': math.tan,
'log': math.log, 'log10': math.log10, 'exp': math.exp,
'pi': math.pi, 'e': math.e
}
expression = expression.replace('^', '**')
result = eval(expression, {"__builtins__": {}}, allowed_names)
return f"Result: {result}"
except Exception as e:
return f"Error in calculation: {str(e)}"我們定義代理的內部狀態並實現強大的計算器工具。 AgentState類使用TypedDict來構建Agent Memory，特別是跟踪對話期間交換的消息。用@tool裝飾的計算器功能將其註冊為AI可使用的實用程序，可以安全地評估數學表達式。它可以通過將可用功能限制為從數學模塊的預定義集進行安全計算，並用Python的指示操作員替換common語法。這樣可以確保該工具可以處理簡單的算術和高級功能，例如三角法或對數，同時防止不安全的代碼執行。
@tool
def web_search(query: str, num_results: int = 3) -> str:
"""
Search the web for information using DuckDuckGo.
Args:
query: Search query string
num_results: Number of results to return (default: 3, max: 10)
Returns:
Search results as formatted string
"""
try:
num_results = min(max(num_results, 1), 10)
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=num_results))
if not results:
return f"No search results found for: {query}"
formatted_results = f"Search results for '{query}':\n\n"
for i, result in enumerate(results, 1):
formatted_results += f"{i}. **{result('title')}**\n"
formatted_results += f" {result('body')}\n"
formatted_results += f" Source: {result('href')}\n\n"
return formatted_results
except Exception as e:
return f"Error performing web search: {str(e)}"我們定義了一個Web_search工具，該工具可以通過DuckDuckgo搜索API通過DuckDuckgo_Search Python軟件包從Internet獲取實時信息。該工具接受搜索查詢和可選的num_results參數，確保返回的結果數在1到10之間。它打開了DuckDuckgo搜索會話，檢索結果並將其整齊地格式化以使其用於用戶友好的顯示。如果找不到結果或發生錯誤，則該功能通過返回信息性消息來優雅處理。該工具為代理提供了實時搜索功能，增強了響應能力和實用性。
@tool
def weather_info(city: str) -> str:
"""
Get current weather information for a city using OpenWeatherMap API.
Note: This is a mock implementation for demo purposes.
Args:
city: Name of the city
Returns:
Weather information as a string
"""
mock_weather = {
"new york": {"temp": 22, "condition": "Partly Cloudy", "humidity": 65},
"london": {"temp": 15, "condition": "Rainy", "humidity": 80},
"tokyo": {"temp": 28, "condition": "Sunny", "humidity": 70},
"paris": {"temp": 18, "condition": "Overcast", "humidity": 75}
}
city_lower = city.lower()
if city_lower in mock_weather:
weather = mock_weather(city_lower)
return f"Weather in {city}:\n" \
f"Temperature: {weather('temp')}°C\n" \
f"Condition: {weather('condition')}\n" \
f"Humidity: {weather('humidity')}%"
else:
return f"Weather data not available for {city}. (This is a demo with limited cities: New York, London, Tokyo, Paris)"我們定義了一個Weather_Info工具，該工具模擬了為給定城市檢索當前的天氣數據。儘管它沒有連接到現場天氣API，但它使用了紐約，倫敦，東京和巴黎等主要城市的模擬數據的預定詞典。收到城市名稱後，該功能將其歸一化為小寫，並檢查其在模擬數據集中是否存在。如果發現，它將以可讀格式返回溫度，天氣狀況和濕度。否則，它會通知用戶天氣數據不可用。該工具是佔位符，後來可以升級以從實際天氣API獲取實時數據。
@tool
def text_analyzer(text: str) -> str:
"""
Analyze text and provide statistics like word count, character count, etc.
Args:
text: Text to analyze
Returns:
Text analysis results
"""
if not text.strip():
return "Please provide text to analyze."
words = text.split()
sentences = text.split('.') + text.split('!') + text.split('?')
sentences = (s.strip() for s in sentences if s.strip())
analysis = f"Text Analysis Results:\n"
analysis += f"• Characters (with spaces): {len(text)}\n"
analysis += f"• Characters (without spaces): {len(text.replace(' ', ''))}\n"
analysis += f"• Words: {len(words)}\n"
analysis += f"• Sentences: {len(sentences)}\n"
analysis += f"• Average words per sentence: {len(words) / max(len(sentences), 1):.1f}\n"
analysis += f"• Most common word: {max(set(words), key=words.count) if words else 'N/A'}"
return analysisText_Analyzer工具提供了給定文本輸入的詳細統計分析。它計算諸如字符計數（有或沒有空格），單詞計數，句子計數和平均單詞等指標，並標識最常見的單詞。該工具通過提示用戶提供有效的文本來優雅地處理空輸入。它使用簡單的字符串操作以及Python的集合和最大功能來提取有意義的見解。它是AI代理工具包中語言分析或內容質量檢查的有價值的實用程序。
@tool
def current_time() -> str:
"""
Get the current date and time.
Returns:
Current date and time as a formatted string
"""
now = datetime.now()
return f"Current date and time: {now.strftime('%Y-%m-%d %H:%M:%S')}"Current_time工具提供了一種以人類可讀格式檢索當前系統日期和時間的簡單方法。使用Python的DateTime模塊，它捕獲了當前時刻，並將其格式化為Yyyy-MM-DD HH：MM：SS。該實用程序對於時間stamp的響應或回答有關AI代理交互流中當前日期和時間的用戶查詢特別有用。
tools = (calculator, web_search, weather_info, text_analyzer, current_time)
def create_llm():
if ANTHROPIC_API_KEY:
return ChatAnthropic(
model="claude-3-haiku-20240307",
temperature=0.1,
max_tokens=1024
)
else:
class MockLLM:
def invoke(self, messages):
last_message = messages(-1).content if messages else ""
if any(word in last_message.lower() for word in ('calculate', 'math', '+', '-', '*', '/', 'sqrt', 'sin', 'cos')):
import re
numbers = re.findall(r'(\d\+\-\*/\.\(\)\s\w)+', last_message)
expr = numbers(0) if numbers else "2+2"
return AIMessage(content="I'll help you with that calculation.",
tool_calls=({"name": "calculator", "args": {"expression": expr.strip()}, "id": "calc1"}))
elif any(word in last_message.lower() for word in ('search', 'find', 'look up', 'information about')):
query = last_message.replace('search for', '').replace('find', '').replace('look up', '').strip()
if not query or len(query) < 3:
query = "python programming"
return AIMessage(content="I'll search for that information.",
tool_calls=({"name": "web_search", "args": {"query": query}, "id": "search1"}))
elif any(word in last_message.lower() for word in ('weather', 'temperature')):
city = "New York"
words = last_message.lower().split()
for i, word in enumerate(words):
if word == 'in' and i + 1 < len(words):
city = words(i + 1).title()
break
return AIMessage(content="I'll get the weather information.",
tool_calls=({"name": "weather_info", "args": {"city": city}, "id": "weather1"}))
elif any(word in last_message.lower() for word in ('time', 'date')):
return AIMessage(content="I'll get the current time.",
tool_calls=({"name": "current_time", "args": {}, "id": "time1"}))
elif any(word in last_message.lower() for word in ('analyze', 'analysis')):
text = last_message.replace('analyze this text:', '').replace('analyze', '').strip()
if not text:
text = "Sample text for analysis"
return AIMessage(content="I'll analyze that text for you.",
tool_calls=({"name": "text_analyzer", "args": {"text": text}, "id": "analyze1"}))
else:
return AIMessage(content="Hello! I'm a multi-tool agent powered by Claude. I can help with:\n• Mathematical calculations\n• Web searches\n• Weather information\n• Text analysis\n• Current time/date\n\nWhat would you like me to help you with?")
def bind_tools(self, tools):
return self
print("⚠️ Note: Using mock LLM for demo. Add your ANTHROPIC_API_KEY for full functionality.")
return MockLLM()
llm = create_llm()
llm_with_tools = llm.bind_tools(tools)
我們初始化為AI代理提供動力的語言模型。如果有有效的擬人API鍵，它將使用Claude 3 Haiku模型進行高質量響應。如果沒有API鍵，則定義了一個無用的模擬以模擬基於關鍵字匹配的基本工具路由行為，從而使代理可以脫機功能有限的功能。 bind_tools方法將定義的工具鏈接到模型，使其可以根據需要調用它們。
def agent_node(state: AgentState) -> Dict(str, Any):
"""Main agent node that processes messages and decides on tool usage."""
messages = state("messages")
response = llm_with_tools.invoke(messages)
return {"messages": (response)}
def should_continue(state: AgentState) -> str:
"""Determine whether to continue with tool calls or end."""
last_message = state("messages")(-1)
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
return "tools"
return END我們定義代理的核心決策邏輯。 Agent_Node函數處理傳入消息，調用語言模型（使用工具），然後返回模型的響應。然後，syse_continue函數評估模型的響應是否包括工具調用。如果是這樣，它將控件路由到工具執行節點；否則，它將引導流程以結束相互作用。這些功能可以在代理的工作流程內進行動態和條件過渡。
def create_agent_graph():
tool_node = ToolNode(tools)
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
workflow.add_edge("tools", "agent")
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
return app
print("Creating LangGraph Multi-Tool Agent...")
agent = create_agent_graph()
print("✓ Agent created successfully!\n")我們構建了定義AI代理的操作結構的langgraph-Power工作流程。它初始化了一個工具句來處理工具執行並使用狀態圖來組織代理決策和工具使用之間的流程。添加節點和邊緣以管理過渡：從代理開始，有條件地路由到工具，然後根據需要向後循環。集成了一個存儲器，以跨回合進行持續狀態跟踪。該圖被編譯到可執行的應用程序（APP）中，啟用了一個結構化的，內存的多工具代理，準備部署。
def test_agent():
"""Test the agent with various queries."""
config = {"configurable": {"thread_id": "test-thread"}}
test_queries = (
"What's 15 * 7 + 23?",
"Search for information about Python programming",
"What's the weather like in Tokyo?",
"What time is it?",
"Analyze this text: 'LangGraph is an amazing framework for building AI agents.'"
)
print("🧪 Testing the agent with sample queries...\n")
for i, query in enumerate(test_queries, 1):
print(f"Query {i}: {query}")
print("-" * 50)
try:
response = agent.invoke(
{"messages": (HumanMessage(content=query))},
config=config
)
last_message = response("messages")(-1)
print(f"Response: {last_message.content}\n")
except Exception as e:
print(f"Error: {str(e)}\n")test_agent函數是驗證實用程序，可確保langgraph代理在不同用例中正確響應。它運行預定義的查詢，算術，網絡搜索，天氣，時間和文本分析，並打印代理的響應。使用一致的螺紋_ID進行配置，它使用每個查詢調用代理。它整齊地顯示結果，幫助開發人員在轉向交互式或生產使用之前驗證工具集成和對話邏輯。
def chat_with_agent():
"""Interactive chat function."""
config = {"configurable": {"thread_id": "interactive-thread"}}
print("🤖 Multi-Tool Agent Chat")
print("Available tools: Calculator, Web Search, Weather Info, Text Analyzer, Current Time")
print("Type 'quit' to exit, 'help' for available commands\n")
while True:
try:
user_input = input("You: ").strip()
if user_input.lower() in ('quit', 'exit', 'q'):
print("Goodbye!")
break
elif user_input.lower() == 'help':
print("\nAvailable commands:")
print("• Calculator: 'Calculate 15 * 7 + 23' or 'What's sin(pi/2)?'")
print("• Web Search: 'Search for Python tutorials' or 'Find information about AI'")
print("• Weather: 'Weather in Tokyo' or 'What's the temperature in London?'")
print("• Text Analysis: 'Analyze this text: (your text)'")
print("• Current Time: 'What time is it?' or 'Current date'")
print("• quit: Exit the chat\n")
continue
elif not user_input:
continue
response = agent.invoke(
{"messages": (HumanMessage(content=user_input))},
config=config
)
last_message = response("messages")(-1)
print(f"Agent: {last_message.content}\n")
except KeyboardInterrupt:
print("\nGoodbye!")
break
except Exception as e:
print(f"Error: {str(e)}\n")CHAT_WITH_AGENT函數提供了與Langgraph Multi-Tool代理的實時對話的交互式命令行接口。它支持自然語言查詢並識別諸如使用指導的“幫助”之類的命令，並“退出”退出。每個用戶輸入都是通過代理處理的，該代理會動態選擇並調用適當的響應工具。該功能通過模擬對話體驗並展示代理在處理各種查詢（從數學和網絡搜索到天氣，文本分析和時間檢索）方面的功能來增強用戶參與度。
if __name__ == "__main__":
test_agent()
print("=" * 60)
print("🎉 LangGraph Multi-Tool Agent is ready!")
print("=" * 60)
chat_with_agent()
def quick_demo():
"""Quick demonstration of agent capabilities."""
config = {"configurable": {"thread_id": "demo"}}
demos = (
("Math", "Calculate the square root of 144 plus 5 times 3"),
("Search", "Find recent news about artificial intelligence"),
("Time", "What's the current date and time?")
)
print("🚀 Quick Demo of Agent Capabilities\n")
for category, query in demos:
print(f"({category}) Query: {query}")
try:
response = agent.invoke(
{"messages": (HumanMessage(content=query))},
config=config
)
print(f"Response: {response('messages')(-1).content}\n")
except Exception as e:
print(f"Error: {str(e)}\n")
print("\n" + "="*60)
print("🔧 Usage Instructions:")
print("1. Add your ANTHROPIC_API_KEY to use Claude model")
print(" os.environ('ANTHROPIC_API_KEY') = 'your-anthropic-api-key'")
print("2. Run quick_demo() for a quick demonstration")
print("3. Run chat_with_agent() for interactive chat")
print("4. The agent supports: calculations, web search, weather, text analysis, and time")
print("5. Example: 'Calculate 15*7+23' or 'Search for Python tutorials'")
print("="*60)最後，我們協調Langgraph多工具代理的執行。如果腳本是直接運行的，它將啟動test_agent（）以示例查詢驗證功能，然後啟動Interactive CHAT_WITH_WITH_AGENT（）模式進行實時交互。 Quick_Demo（）函數還簡要展示了代理在數學，搜索和時間查詢中的功能。清晰的用法說明在末尾打印，指導用戶配置API鍵，運行演示並與代理進行交互。這為用戶提供了平穩的入門體驗，可以探索和擴展代理的功能。
總之，該分步教程為建立有效的多工具AI代理提供了利用Langgraph和Claude的生成能力的寶貴見解。通過直接的解釋和動手演示，該指南使用戶能夠將各種實用程序整合到一個凝聚力和互動系統中。代理執行任務的靈活性，從復雜的計算到動態信息檢索，展示了現代AI開發框架的多功能性。此外，包括用於測試和交互式聊天的用戶友好功能可以增強實踐理解，從而在各種情況下立即應用。開發人員可以通過這種基本知識自信地擴展和自定義其AI代理。
在Github上查看筆記本。這項研究的所有信用都歸該項目的研究人員。另外，請隨時關注我們 嘰嘰喳喳 而且不要忘記加入我們的 95k+ ml子雷迪特 並訂閱 我們的新聞通訊。
Asif Razzaq是Marktechpost Media Inc.的首席執行官。作為一位有遠見的企業家和工程師，ASIF致力於利用人工智能的潛力來實現社會利益。他最近的努力是推出了人工智能媒體平台Marktechpost，該平台的深入覆蓋了機器學習和深度學習新聞，既在技術上都可以聽起來，既可以通過技術上的聲音，又可以被廣泛的受眾理解。該平台每月有超過200萬個觀點，說明了其在受眾中的受歡迎程度。
