LangGraph Yapay Zeka Ajanları

Deeplearning.ai’ın “AI Agents in LangGraph” kursunun Türkçe özetidir.

For English:

AI Agents in LangGraph

Giriş

Geçen yıl içinde birkaç önemli gelişme yaşandı. İlk olarak, fonksiyon çağırma LLM’leri, araç kullanımını çok daha öngörülebilir ve istikrarlı hale getirdi. Ayrıca, arama gibi belirli araçlar ajanik kullanıma uyarlanmıştır. Bir arama motorunda sorgu yaptığınızda, ardından takip edip yanıtları bulabileceğiniz birden fazla bağlantı döndürecektir. Ancak, bir ajanın istediği şey referans olarak kaynağın linkini alabilmektir. Ayrıca, ajan sonuçlar için öngörülebilir formatlara ihtiyaç duyar. Bu, Ajanik Arama’nın sağladığı şeydir.

Üç ajanın daha büyük bir işi paylaştığı durum Ajanik iş akışı olarak adlandırılır. Ajanik iş akışı, operasyonun adımlarını düşünerek planlama ile başlar. Ajanik iş akışının bir diğer önemli unsuru ise araç kullanımıdır. LLM’ler, hangi araçların mevcut olduğunu ve bunların nasıl kullanılacağını bilmelidir, tıpkı bizim arama ajanımız gibi. Yansıtma (reflection), bir sonucu iteratif olarak geliştirmek anlamına gelir; bu, birden fazla LLM’in eleştiri yapması ve bu tür bir düzenleme döngüsünü yönlendirmek için faydalı önerilerde bulunması ile gerçekleşebilir. Çoklu ajan iletişimi, her bir ajanın benzersiz bir istem ile bir rol oynadığı iletişim şeklidir. Son olarak, hafıza, birden fazla adım boyunca sonuçlardaki ilerlemeyi takip edebilmeyi sağlar. Bu yeteneklerin bazıları araç kullanımı için fonksiyon çağırma gibi LLM’in kendisiyle ilişkilidir, ancak bunların çoğu, içinde çalıştıkları çerçeve tarafından LLM’in dışında uygulanmaktadır.

LangGraph Döngüsel Grafları destekler.

ReAct, yukarıdaki görselde en soldaki YZ ajanları hakkında erken dönem bir makaledir. Görselde ortada olan Kendini geliştirme (Self-refine) istemlerin kendilerini iyileştirmeleridir. Görselde en sağda, akış mühendisliğini (flow engineering) kullanarak bir kodlama ajanı oluşturmayı gösteren Alpha codium makalesi yer almaktadır. Bunların ortak noktaları döngüsel yapıdır. Bu yapıları destekleyebilmek için Langchain, LangGraph’ı oluşturmuştur. Bu kursta yalnızca LLM ve Python ile sıfırdan bir ajan oluşturacaksınız.

Sonrasında aynı ajanı LangGraph bileşenlerini doğrudan kullanarak yeniden inşa ederek LangGraph bileşenlerini tanıyacağız. Arama araçları birçok ajan uygulamasının önemli bir parçası olduğu için, ajan arama yeteneklerini ve nasıl kullanılacaklarını öğreneceksiniz. Ajan oluştururken 2 faydalı yetenek vardır. İlki insan girdisi alabilmektir. Bu, kritik noktalarda bir ajana rehberlik etmenizi sağlar. İkincisi ise kalıcılıktır. Bu, mevcut bilgi durumunu saklayabilme yeteneğidir, böylece daha sonra bu duruma geri dönebilirsiniz. Bu hem ajanlarda hata ayıklamak hem de ajanları üretim ortamında konuşlandırmak için harikadır.

Sıfırdan Ajan İnşaa Etmek

Ajanlar oldukça kompleks görevleri gerçekleştirebilirler. Basit bir ajanı inşaa etmek o kadar zor değildir. Hangi işlerin LLM’e düştüğünü ve hangilerinin LLM etrafındaki kod tarafından (çalışma zamanında) yönetildiğini fark etmemiz gerekir.

ReAct Örüntüsü

Sıfırdan oluşturacağımız ajan ReAct modelini temel almaktadır. ReAct, Akıl Yürütme + Harekete Geçme anlamına gelmektedir. LLM önce ne yapacağını düşünür ve ardından ne yapılacağına karar verir. Bu eylem daha sonra bir ortamda gerçekleştirilir ve bir gözlem döndürülür. Bu gözlemin ardından LLM tekrar çalışır. LLM tekrar ne yapacağını düşünür ve başka bir aksiyona karar verir ve görevinin tamamlandığına karar verene kadar çalışmaya devam eder.

import openai
import re
import httpx
import os
from dotenv import load_dotenv

_ = load_dotenv()
from openai import OpenAI

client = OpenAI()

chat_completion = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[{"role": "user", "content": "Hello world"}]
)

print(chat_completion.choices[0].message.content) # Hello! How can I assist you today?

class Agent:
    def __init__(self, system=""):
        self.system = system
        self.messages = []
        if self.system:
            self.messages.append({"role": "system", "content": system})

    def __call__(self, message):
        self.messages.append({"role": "user", "content": message})
        result = self.execute()
        self.messages.append({"role": "assistant", "content": result})
        return result

    def execute(self):
        completion = client.chat.completions.create(
                        model="gpt-4o", 
                        temperature=0,
                        messages=self.messages)
        return completion.choices[0].message.content

prompt = """
You run in a loop of Thought, Action, PAUSE, Observation.
At the end of the loop you output an Answer
Use Thought to describe your thoughts about the question you have been asked.
Use Action to run one of the actions available to you - then return PAUSE.
Observation will be the result of running those actions.

Your available actions are:

calculate:
e.g. calculate: 4 * 7 / 3
Runs a calculation and returns the number - uses Python so be sure to use floating point syntax if necessary

average_dog_weight:
e.g. average_dog_weight: Collie
returns average weight of a dog when given the breed

Example session:

Question: How much does a Bulldog weigh?
Thought: I should look the dogs weight using average_dog_weight
Action: average_dog_weight: Bulldog
PAUSE

You will be called again with this:

Observation: A Bulldog weights 51 lbs

You then output:

Answer: A bulldog weights 51 lbs
""".strip()

def calculate(what):
    return eval(what)

def average_dog_weight(name):
    if name in "Scottish Terrier": 
        return("Scottish Terriers average 20 lbs")
    elif name in "Border Collie":
        return("a Border Collies average weight is 37 lbs")
    elif name in "Toy Poodle":
        return("a toy poodles average weight is 7 lbs")
    else:
        return("An average dog weights 50 lbs")

known_actions = {
    "calculate": calculate,
    "average_dog_weight": average_dog_weight
}

abot = Agent(prompt)

result = abot("How much does a toy poodle weigh?")
print(result)

result = average_dog_weight("Toy Poodle")

print(result) # a toy poodles average weight is 7 lbs

next_prompt = "Observation: {}".format(result)

print(abot(next_prompt)) # Answer: A Toy Poodle weighs 7 lbs.

print(abot.messages)

abot = Agent(prompt) # Reinitializing to clear memory
question = """I have 2 dogs, a border collie and a scottish terrier. \
What is their combined weight"""
print(abot(question))
# Thought: To find the combined weight of a Border Collie and a Scottish
# Terrier, I need to first find the average weight of each breed and then add those weights together.
# I'll start by finding the average weight of a Border Collie.\n\nAction: average_dog_weight: Border Collie\nPAUSE

next_prompt = "Observation: {}".format(average_dog_weight("Border Collie"))
print(next_prompt) # Observation: a Border Collies average weight is 37 lbs
print(abot(next_prompt))
# Thought: Now that I know a Border Collie's average weight is 37 lbs, I need to find the average weight of a 
# Scottish Terrier tocalculate the combined weight. \n\nAction: average_dog_weight: Scottish Terrier\nPAUSE

next_prompt = "Observation: {}".format(average_dog_weight("Scottish Terrier"))
print(next_prompt) # Observation: Scottish Terriers average 20 lbs
print(abot(next_prompt))
# Thought: With the average weight of a Border Collie being 37 lbs and a
# Scottish Terrier being 20 lbs, I can now calculate their combined weight.\n\nAction: calculate: 37+20\nPAUSE

next_prompt = "Observation: {}".format(eval("37 + 20"))
print(next_prompt) # Observation: 57
print(abot(next_prompt)) # Answer: The combined weight of a Border Collie and a Scottish Terrier is 57 lbs.

Döngü Eklemek

Bunu otomatikleştirmek için yukarıdaki kodu bir döngüye koyarız. Eylem dizesini arayan eylem dizesini arayan bir RegEx (DüzAde: Düzenli İfade) oluşturacağız. Bu, LLM’nin yanıtını ayrıştırmamıza ve bir eylemde bulunmak isteyip istemediğimize veya bunun son yanıt olup olmadığına karar vermemize olanak tanıyacaktır.

action_re = re.compile('^Action: (\w+): (.*)$')   # python regular expression to selection action

def query(question, max_turns=5):
    i = 0
    bot = Agent(prompt)
    next_prompt = question
    while i < max_turns:
        i += 1
        result = bot(next_prompt)
        print(result)
        actions = [
            action_re.match(a) 
            for a in result.split('\n') 
            if action_re.match(a)
        ]
        if actions:
            # There is an action to run
            action, action_input = actions[0].groups()
            if action not in known_actions:
                raise Exception("Unknown action: {}: {}".format(action, action_input))
            print(" -- running {} {}".format(action, action_input))
            observation = known_actions[action](action_input)
            print("Observation:", observation)
            next_prompt = "Observation: {}".format(observation)
        else:
            return

question = """I have 2 dogs, a border collie and a scottish terrier. \
What is their combined weight"""
query(question)

LangGraph Bileşenleri

Bir kullanıcı mesajı geldi ve sistem mesajı ile birlikte LLM’e istek attık. LLM bir düşünce ve bir eylem çıktısı verdi. Buna göre bir karar verdik ya geri dönecektik ya da bir fonksyion çağıracaktık. Bütün bunları bir döngüye koyuyoruz. calculate() ve average_dog_weight() olmak üzere 2 aracımız var. Araçları çağırırsak bir gözlem elde ederiz. Daha sonra döngüye girip bu gözlemi girdi olarak geri koyuyoruz ve yeniden başlıyoruz.

Bunu LangChain bileşenlerine ayıralım.

Langchain İstemleri

Bilgi istemi şablonları yeniden kullanılabilir istemlere izin verir.

from langchain.prompts import PromptTemplate prompt_template = PromptTemplate.from_template("Tell me a {adjective} joke about {content}.")

Ayrıca merkez’de (hub) aracılara yönelik istemler de (prompts) mevcuttur:

prompt = hub.pull(“hwchase17/react”)

LangChain Araçları

# get a tool from the library
from langchain_community.tools.tavily_search import TavilySearchResults
tool = TavilySearchResults(max_results=2)

self.tools = {t.name: t for t in tools}
self.model = model.bind_tools([tool])

LangGraph’taki Yeni Yapılar

LangGraph, kontrol akışını tanımlamanıza ve düzenlemenize yardımcı olur. Özellikle, döngüsel graflar oluşturmanıza olanak tanır. Ayrıca, yerleşik kalıcılık (persistence) ile birlikte gelir. Aynı anda birden fazla sohbet yapabilmek veya önceki konuşmaları ve eylemleri hatırlamak faydalıdır. Kalıcılık, insanın devrede olduğu yararlı özellikleri de sağlar. LangGraph, LangChain’in graf destekleyen bir uzantısıdır. LangGraph, son derece kontrollü “akışlar” oluşturmanıza olanak tanır. Tek ve çok ajanlı akışlar graflar olarak tanımlanır ve temsil edilir.

Graflar, ajanları veya işlevleri temsil eden düğümleri, düğümleri bağlayan kenarları ve kararları temsil eden koşullu kenarları içerir.

Veri/Durum

Durum, LangGraph’ta zaman içinde takip edilen önemli şeylerden biridir. Ajan durumuna grafiğin tüm bölümleri tarafından erişilebilir. Ajan durumu grafta yereldir ve kalıcılık katmanında saklanabilir. Buna genellikle ajan durumu denir.

Basit Durum

Basit Ajan Durumu, BaseMessages dizilerinin bir listesinden oluşur. “operator.add” mevcut mesajları geçersiz kılmaz, bunun yerine onları bu duruma ekler.

class AgentState(TypeDict):
  messages: Annotated[Sequence[BaseMessage], operator.add]

Karmaşık Durum

Bu parametrelerden herhangi birine açıklama eklenmemiştir; bu, söz konusu değişkene yeni bir güncelleme gönderildiğinde mevcut değeri geçersiz kıldığı anlamına gelir, ancak ara adımlar “operator.add” ile açıklanır, bu da yeni bir güncelleme gönderildiğinde eklendiği anlamına gelir.

class AgentState(TypeDict):
  input: str
  chat_history: list[BaseMessage]
  agent_outcome: Union[AgentAction, AgentFinish, None]
  intermediate_steps: Annotated[list[tuple[AgentAction, str], operator.add]

Kod

LLM: call_openai, c_edge: exists_action, action:take_action

Durum

class AgentState(TypedDict):
  messages: Annotated[list[AnyMessage], operator.add]

Hadi uygulayalım:

Tavily araç olarak kullanacağımız bir arama motorudur.

from dotenv import load_dotenv
_ = load_dotenv()

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=4) #increased number of results
print(type(tool))
print(tool.name) # <class 'langchain_community.tools.tavily_search.tool.TavilySearchResults'> tavily_search_results_json

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

Not: Aşağıdaki take_action fonksiyonuna, LLM’in var olmayan bir araç adı döndürmesi durumunu kapsayacak bir mantık eklendi. LLM’ler fonksiyon çağrılarıyla bile ara sıra halüsinasyon görebilir. Yapılan tek şeyin LLM’e tekrar denemesi talimatını vermek olduğunu unutmayın! Bu ajanik organizasyonunun bir avantajıdır.

class Agent:

    def __init__(self, model, tools, system=""):
        self.system = system # system_message
        graph = StateGraph(AgentState) # graph canvas that has nothing in it
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges(
            "llm", # node where the edge starts 
            self.exists_action, # function tells us where to go after that
            {True: "action", False: END} # how to map the response of the fnction to the next node to go to
        )
        graph.add_edge("action", "llm") # regular edge
        graph.set_entry_point("llm") 
        self.graph = graph.compile()
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    def exists_action(self, state: AgentState):
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

    def take_action(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []
        for t in tool_calls:
            print(f"Calling: {t}")
            if not t['name'] in self.tools:      # check for bad tool name from LLM
                print("\n ....bad tool name....")
                result = "bad tool name, retry"  # instruct LLM to retry if bad
            else:
                result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print("Back to the model!")
        return {'messages': results}

prompt = """You are a smart research assistant. Use the search engine to look up information. \
You are allowed to make multiple calls (either together or in sequence). \
Only look up information when you are sure of what you want. \
If you need to look up some information before asking a follow up question, you are allowed to do that!
"""

model = ChatOpenAI(model="gpt-3.5-turbo")  #reduce inference cost
abot = Agent(model, [tool], system=prompt)

messages = [HumanMessage(content="What is the weather in sf?")]
result = abot.graph.invoke({"messages": messages})

print(result)
print(result['messages'][-1].content)

messages = [HumanMessage(content="What is the weather in SF and LA?")]
result = abot.graph.invoke({"messages": messages})
print(result['messages'][-1].content)

# Note, the query was modified to produce more consistent results. 
# Results may vary per run and over time as search information and models change.

query = "Who won the super bowl in 2024? In what state is the winning team headquarters located? \
What is the GDP of that state? Answer each question." 
messages = [HumanMessage(content=query)]

model = ChatOpenAI(model="gpt-4o")  # requires more advanced model
abot = Agent(model, [tool], system=prompt)
result = abot.graph.invoke({"messages": messages})

print(result['messages'][-1].content)

Ajanik Arama Araçları

Ajanik aramanın standart aramadan ne kadar farklı olduğunu ve nasıl kullanılacağını öğrenelim.

Neden arama aracı?

İstem, aracı tarafından alınır ve ardından arama aracını aramaya karar verir. Daha sonra bulunan bilgiler ajana iade edilir.

Bir arama aracının içi

Yukarıda, çok temel bir arama aracı uygulaması bulunmaktadır. Ajan, sorguyu arama aracına göndermeye karar verirse, ilk adım soruyu anlamaya çalışmak ve gerekiyorsa alt sorulara bölmek olacaktır. Bu önemli bir adımdır çünkü böylece karmaşık sorgular ele alınabilir. Daha sonra her alt soru için, arama aracı birden fazla entegrasyon arasından en iyi kaynağı bulmak zorunda kalacaktır. İş, doğru kaynağı bulmakla bitmez. Arama aracı, alt soruya uygun yalnızca ilgili bilgileri çıkarmak zorunda olacaktır. Bunun temel bir uygulaması, kaynağı parçalara ayırma ve en üst k öğeleri getirmek için hızlı bir vektör araması yapma süreciyle gerçekleştirilebilir.

# libraries
from dotenv import load_dotenv
import os
from tavily import TavilyClient

# load environment variables from .env file
_ = load_dotenv()

# connect
client = TavilyClient(api_key=os.environ.get("TAVILY_API_KEY"))

# run search
result = client.search("What is in Nvidia's new Blackwell GPU?",
                       include_answer=True)

# print the answer
result["answer"]

'The new Nvidia Blackwell GPU architecture, specifically the Blackwell B200 GPU, is designed to power the next generation of AI supercomputers. It is expected to deliver up to 20 petaflops of compute performance and potentially more than quadruple the performance of its predecessor. Additionally, the Blackwell platform is aimed at enabling organizations to build and run real-time generative AI on trillion-parameter large language models at a significantly lower cost and energy consumption compared to its predecessor.'

Düzenli Arama

# choose location (try to change to your own city!)

city = "San Francisco"

query = f"""
    what is the current weather in {city}?
    Should I travel there today?
    "weather.com"
"""

Not: Bir istisna durumunda beklenen sonuçları döndürmek için arama değiştirildi. Öğrenci trafiğinin yoğun olduğu durumlarda bazen hız sınırlama istisnaları meydana gelebilir.

import requests
from bs4 import BeautifulSoup
from duckduckgo_search import DDGS
import re

ddg = DDGS()

def search(query, max_results=6):
    try:
        results = ddg.text(query, max_results=max_results)
        return [i["href"] for i in results]
    except Exception as e:
        print(f"returning previous results due to exception reaching ddg.")
        results = [ # cover case where DDG rate limits due to high deeplearning.ai volume
            "https://weather.com/weather/today/l/USCA0987:1:US",
            "https://weather.com/weather/hourbyhour/l/54f9d8baac32496f6b5497b4bf7a277c3e2e6cc5625de69680e6169e7e38e9a8",
        ]
        return results  


for i in search(query):
    print(i)
# https://weather.com/weather/tenday/l/San Francisco CA USCA0987:1:US
# https://weather.com/storms/hurricane/news/2024-07-05-hurricane-beryl-forecast-mexico-texas

def scrape_weather_info(url):
    """Scrape content from the given URL"""
    if not url:
        return "Weather information could not be found."
    
    # fetch data
    headers = {'User-Agent': 'Mozilla/5.0'}
    response = requests.get(url, headers=headers)
    if response.status_code != 200:
        return "Failed to retrieve the webpage."

    # parse result
    soup = BeautifulSoup(response.text, 'html.parser')
    return soup

# use DuckDuckGo to find websites and take the first result
url = search(query)[0]

# scrape first wesbsite
soup = scrape_weather_info(url)

print(f"Website: {url}\n\n") # Website: https://weather.com/weather/today/l/San+Francisco+CA+USCA0987:1:US
print(str(soup.body)[:50000]) # limit long outputs

recents Specialty Forecasts San Francisco, CA Today's Forecast for San Francisco, CA Morning Afternoon Evening Overnight Weather Today in San Francisco, CA 5:58 am 8:32 pm Don't Miss Hourly Forecast Now 1 pm 2 pm 3 pm 4 pm Outside That's Not What Was Expected Daily Forecast Today Sun 14 Mon 15 Tue 16 Wed 17 Radar We Love Our Critters Summer Skin Essentials Home, Garage & Garden The Real Toll Of Extreme Heat Health News For You Weather in your inbox Your local forecast, plus daily trivia, stunning photos and our meteorologists’ top picks. All in one place, every weekday morning. By signing up, you're opting in to receive the Morning Brief email newsletter. To manage your data, visit Data Rights . Terms of Use | Privacy Policy Stay Safe Fact Versus Fiction Air Quality Index Air quality is considered satisfactory, and air pollution poses little or no risk. Health & Activities Seasonal Allergies and Pollen Count Forecast Grass pollen is low in your area Cold & Flu Forecast Flu risk is low in your area We recognize our responsibility to use data and technology for good. We may use or share your data with our data vendors. Take control of your data. © The Weather Company, LLC 2024

Ajanik Arama

# run search
result = client.search(query, max_results=1)

# print first result
data = result["results"][0]["content"]

print(data) # Weather.com brings you the most accurate monthly weather forecast for San Francisco, ... 13 69 ° 54 ° 14. 72 ° 54 ° 15 ... July: 66 ° 54 ° 0: August: 68 ° 56 ...

import json
from pygments import highlight, lexers, formatters

# parse JSON
parsed_json = json.loads(data.replace("'", '"'))

# pretty print JSON with syntax highlighting
formatted_json = json.dumps(parsed_json, indent=4)
colorful_json = highlight(formatted_json,
                          lexers.JsonLexer(),
                          formatters.TerminalFormatter())

print(colorful_json)

Kalıcılık ve Akış (Persistence and Streaming)

Ajanlar genellikle daha uzun süreli görevler üzerinde çalışırlar. Bu tür görevler için gerçekten önemli iki tür görev vardır: kalıcılık ve akış.

Kalıcılık (persistence), belirli bir zamanda bir ajanın durumunu korumanızı sağlar. Bu, gelecekte o duruma geri dönüp o durumdan devam etmemizi sağlar ve uzun süreli çalışan uygulamalar için önemli bir özelliktir. Aynı şekilde, akış ile o anda neler olduğunu belirten bir sinyaller listesi yayabilirsiniz. Böylece uzun süreli uygulamalarda ajanın tam olarak ne yaptığını bilirsiniz. Bireysel mesajları da akışa alabiliriz (stream). Bu, hangi eylemin yapılacağını belirleyen AI mesajı ve bu eylemi gerçekleştirdikten sonra ortaya çıkan gözlem mesajı olabilir. Tokenleri de akışa alabiliriz. LLM çağrısının her tokeni için çıktıyı akışa almak isteyebiliriz. Bir kontrol noktası belirleyici, her düğümden sonra ve düğümler arasında durumu kontrol noktasına kaydeder. Bu ajan için kalıcılık eklemek amacıyla SqlliteSaver kullanacağız. Ayrıca bunu Redis veya Postgres gibi harici bir veritabanına da bağlayabilirsiniz. Farklı izleri (track) kalıcı kontrol noktası belirleyicisi (persistent checkpointer) içinde takip etmek için iş parçacığı (thread) konfigürasyonu kullanılacaktır. Bu, aynı anda birden fazla sohbetin devam etmesine olanak tanır.

from dotenv import load_dotenv

_ = load_dotenv()

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=2)

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(":memory:")

class Agent:
    def __init__(self, model, tools, checkpointer, system=""):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        self.graph = graph.compile(checkpointer=checkpointer)
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

    def exists_action(self, state: AgentState):
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

    def take_action(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []
        for t in tool_calls:
            print(f"Calling: {t}")
            result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print("Back to the model!")
        return {'messages': results}

prompt = """You are a smart research assistant. Use the search engine to look up information. \
You are allowed to make multiple calls (either together or in sequence). \
Only look up information when you are sure of what you want. \
If you need to look up some information before asking a follow up question, you are allowed to do that!
"""
model = ChatOpenAI(model="gpt-4o")
abot = Agent(model, [tool], system=prompt, checkpointer=memory)

messages = [HumanMessage(content="What is the weather in sf?")]

thread = {"configurable": {"thread_id": "1"}}

for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v['messages'])

messages = [HumanMessage(content="What about in la?")]
thread = {"configurable": {"thread_id": "1"}}
for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

messages = [HumanMessage(content="Which one is warmer?")]
thread = {"configurable": {"thread_id": "1"}}
for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

messages = [HumanMessage(content="Which one is warmer?")]
thread = {"configurable": {"thread_id": "2"}}
for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

Akış Tokenleri

Ayrıca eşzamansız kontrol noktalarınız (checpoint) ve olaylarınız (event) da olabilir.

from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver

memory = AsyncSqliteSaver.from_conn_string(":memory:")
abot = Agent(model, [tool], system=prompt, checkpointer=memory)

messages = [HumanMessage(content="What is the weather in SF?")]
thread = {"configurable": {"thread_id": "4"}}
async for event in abot.graph.astream_events({"messages": messages}, thread, version="v1"):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            # Empty content in the context of OpenAI means
            # that the model is asking for a tool to be invoked.
            # So we only print non-empty content
            print(content, end="|")

Döngüdeki İnsan (Human in the loop)

Bir ajan ne yaptığını takip etmek için insanın döngüde kalmasını istediğimiz birçok durum bulunmaktadır. Bu, LangGraph’ta kolaydır. Ek durum bilgileri hafızada saklanır ve get_state() veya get_state_history() kullanıldığında görüntülenir. Durum, ayrıca her durum geçişinde saklanır, oysa daha önce bir kesintide (interrupt) veya sonunda saklanıyordu. Bu, komut çıktısını biraz değiştirir, ancak mevcut bilgilere faydalı bir eklemedir.

from dotenv import load_dotenv
_ = load_dotenv()

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")

from uuid import uuid4
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, AIMessage

"""
In previous examples we've annotated the `messages` state key
with the default `operator.add` or `+` reducer, which always
appends new messages to the end of the existing messages array.

Now, to support replacing existing messages, we annotate the
`messages` key with a customer reducer function, which replaces
messages with the same `id`, and appends them otherwise.
"""
def reduce_messages(left: list[AnyMessage], right: list[AnyMessage]) -> list[AnyMessage]:
    # assign ids to messages that don't have them
    for message in right:
        if not message.id:
            message.id = str(uuid4())
    # merge the new messages with the existing messages
    merged = left.copy()
    for message in right:
        for i, existing in enumerate(merged):
            # replace any existing messages with the same id
            if existing.id == message.id:
                merged[i] = message
                break
        else:
            # append any new messages to the end
            merged.append(message)
    return merged

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], reduce_messages]

tool = TavilySearchResults(max_results=2)

Manuel insan onayı

Grafiği derlerken, kontrol noktası (checkpointer) yerine, ‘equations’ eylemi parametresinden önce bu kesintiyi (interrupt) ileteceğiz. Eylem düğümünü çağırmadan önce bir kesinti ekleyeceğiz. Eylem düğümü, araçları (tools) çağırdığımız yerdir. Bunu yapmamızın nedeni, herhangi bir aracı çalıştırmadan önce manuel onay gerektiren bir şey ekleyecek olmamızdır. Bazen sadece belirli bir araç çağrıldığında kesinti yapmak isteyebilirsiniz.

class Agent:
    def __init__(self, model, tools, system="", checkpointer=None):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        self.graph = graph.compile(
            checkpointer=checkpointer,
            interrupt_before=["action"]
        )
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

    def exists_action(self, state: AgentState):
        print(state)
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

    def take_action(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []
        for t in tool_calls:
            print(f"Calling: {t}")
            result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print("Back to the model!")
        return {'messages': results}

prompt = """You are a smart research assistant. Use the search engine to look up information. \
You are allowed to make multiple calls (either together or in sequence). \
Only look up information when you are sure of what you want. \
If you need to look up some information before asking a follow up question, you are allowed to do that!
"""
model = ChatOpenAI(model="gpt-3.5-turbo")
abot = Agent(model, [tool], system=prompt, checkpointer=memory)

messages = [HumanMessage(content="Whats the weather in SF?")]
thread = {"configurable": {"thread_id": "1"}}
for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

abot.graph.get_state(thread)

# The next node
abot.graph.get_state(thread).next # ('action', )

Ara verdikten sonra devam etme

Devam etmek için aynı iş parçacığı konfigürasyonuyla akışı tekrar çağırabilir ve ‘none’ girdi olarak iletebiliriz. Bu, sonuçların akışını sağlar ve aracın çağrılmasından araç mesajını alırız.

for event in abot.graph.stream(None, thread):
    for v in event.values():
        print(v)

abot.graph.get_state(thread)

abot.graph.get_state(thread).next

messages = [HumanMessage("Whats the weather in LA?")]
thread = {"configurable": {"thread_id": "2"}}
for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)
while abot.graph.get_state(thread).next:
    print("\n", abot.graph.get_state(thread),"\n")
    _input = input("proceed?")
    if _input != "y":
        print("aborting")
        break
    for event in abot.graph.stream(None, thread):
        for v in event.values():
            print(v)

Durum Hafızası

Bir graf yürütülürken her durumun anlık görüntüsü bellekte saklanır. AgentState ve her anlık görüntü için bir iş parçacığı ve benzersiz bir tanımlayıcı gibi başka yararlı şeyler vardır. Anlık görüntülere erişmek için bu iş parçacığı tanımlayıcılarını kullanabilirsiniz.

StateSnapshot: {AgentState, useful_things}

Belleğe erişim sağlamak için kullanılan bazı komutlar vardır, örneğin, g.get_state(thread). Eğer thread'i ID olmadan sağlarsanız ve sadece thread ID'sini döndürürseniz, mevcut durumu döndürecektir. Ayrıca, tüm durum anlık görüntülerinin (state snapshot) bir yineleyicisini döndüren durum geçmişi komutu da vardır. Bu yineleyiciyi, her bir durum için benzersiz tanımlayıcılara erişmek için kullanabilirsiniz. Thread tanımlayıcısı veya o thread TS benzersiz tanımlayıcısı verildiğinde, örneğin, ilk duruma, durum bir'e erişebilir ve bunu bir invoke komutunda kullanabilirsiniz. Bu, durum bir'i mevcut durum veya grafiğin geri kalanı için başlangıç noktası olarak kullanacaktır. Bu, esasen zamanda yolculuktur.

g.invoke(None, {thread, thread_ts})
g.stream(None, {thread, thread_ts})

Durum Değişikliği

Kesintiye kadar çalıştırın ve ardından durumu değiştirin.

messages = [HumanMessage("Whats the weather in LA?")]
thread = {"configurable": {"thread_id": "3"}}
for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

abot.graph.get_state(thread)

current_values = abot.graph.get_state(thread)

current_values.values['messages'][-1]

current_values.values['messages'][-1].tool_calls

# Tool calls associated with message. Does not actually work untill we update state on the graph.
_id = current_values.values['messages'][-1].tool_calls[0]['id']
current_values.values['messages'][-1].tool_calls = [
    {'name': 'tavily_search_results_json',
  'args': {'query': 'current weather in Louisiana'},
  'id': _id}
]

abot.graph.update_state(thread, current_values.values)

abot.graph.get_state(thread)

for event in abot.graph.stream(None, thread):
    for v in event.values():
        print(v)

Zamanda Yolculuk

states = []
for state in abot.graph.get_state_history(thread):
    print(state)
    print('--')
    states.append(state)

Filme alınan durumun aynısını getirmek için aşağıdaki uzaklık -1'den -3'e değiştirildi. Bu, __start__ başlangıç ​​durumunu ve en son yazılım sürümüyle birlikte durum belleğinde saklanan ilk durumu açıklar.

to_replay = states[-3]

to_replay

for event in abot.graph.stream(None, to_replay.config):
    for k, v in event.items():
        print(v)

Zamanda geriye gidip düzenleme

to_replay

_id = to_replay.values['messages'][-1].tool_calls[0]['id']
to_replay.values['messages'][-1].tool_calls = [{'name': 'tavily_search_results_json',
  'args': {'query': 'current weather in LA, accuweather'},
  'id': _id}]

branch_state = abot.graph.update_state(to_replay.config, to_replay.values)

for event in abot.graph.stream(None, branch_state):
    for k, v in event.items():
        if k != "__end__":
            print(v)

Belirli bir zamanda bir duruma mesaj ekleme

to_replay
_id = to_replay.values['messages'][-1].tool_calls[0]['id']
state_update = {"messages": [ToolMessage(
    tool_call_id=_id,
    name="tavily_search_results_json",
    content="54 degree celcius",
)]}
branch_and_add = abot.graph.update_state(
    to_replay.config, 
    state_update, 
    as_node="action")
for event in abot.graph.stream(None, branch_and_add):
    for k, v in event.items():
        print(v)

Ekstra Pratik

Küçük bir graf oluşturun

Bu, durum belleğini kontrol etmeye ilişkin daha fazla bilgi edinmek istiyorsanız üzerinde çalışabileceğiniz küçük, basit bir graf’tır.

from dotenv import load_dotenv
_ = load_dotenv()

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langgraph.checkpoint.sqlite import SqliteSaver

Aşağıdaki duruma sahip basit bir 2 düğümlü graf tanımlayın: -lnode: son düğümdür -scratch: karalama defteri (scratchpad) konumunudur, -count : her adımda artan bir sayaçtır.

class AgentState(TypedDict):
    lnode: str
    scratch: str
    count: Annotated[int, operator.add]
def node1(state: AgentState):
    print(f"node1, count:{state['count']}")
    return {"lnode": "node_1",
            "count": 1,
           }
def node2(state: AgentState):
    print(f"node2, count:{state['count']}")
    return {"lnode": "node_2",
            "count": 1,
           }

Graf N1->N2->N1…’e gidiyor ancak sayım 3'e ulaştığında kesiliyor.

def should_continue(state):
    return state["count"] < 3

builder = StateGraph(AgentState)
builder.add_node("Node1", node1)
builder.add_node("Node2", node2)

builder.add_edge("Node1", "Node2")
builder.add_conditional_edges("Node2", 
                              should_continue, 
                              {True: "Node1", False: END})
builder.set_entry_point("Node1")

memory = SqliteSaver.from_conn_string(":memory:")
graph = builder.compile(checkpointer=memory)

Artık iş parçacığını ayarlayabilir ve çalıştırabiliriz:

thread = {"configurable": {"thread_id": str(1)}}
graph.invoke({"count":0, "scratch":"hi"},thread)

Mevcut duruma bakmak

Mevcut durumu alın. AgentState olan değerleri not edin. config ve thread_ts’yi not edin. Bunları aşağıdaki anlık görüntülere atıfta bulunmak için kullanacaksınız.

graph.get_state(thread)

Bellekteki tüm durum anlık görüntülerini görüntüleyin. Gördüğünüzü izlemenize yardımcı olması için görüntülenen sayım aracısı durumu (count agent state) değişkenini kullanabilirsiniz. Yineleyici tarafından ilk olarak en son anlık görüntülerin döndürüldüğüne dikkat edin. Ayrıca meta verilerdeki kullanışlı bir adım değişkeninin, graf yürütmedeki adım sayısını saydığını unutmayın. Bu biraz ayrıntılı — ancak parent_config’in önceki düğümün konfigürasyonu olduğunu da fark edebilirsiniz. İlk kez başlatılırken, bir ebeveyn (parent) oluşturmak için belleğe ek durumlar eklenir. Bu, aşağıda dallara ayrıldığınızda veya zaman yolculuğu yaptığınızda kontrol etmeniz gereken bir şeydir.

Durum Geçmişine Bakmak

for state in graph.get_state_history(thread):
    print(state, "\n")

Yalnızca config’i bir listede saklayın. Sağdaki sayım sırasına dikkat edin. get_state_history ilk önce en son anlık görüntüleri döndürür.

states = []
for state in graph.get_state_history(thread):
    states.append(state.config)
    print(state.config, state.values['count'])

İlk durumu yakalayın.

states[-3]

Bu, Düğüm1'in ilk kez tamamlanmasından sonraki durumdur. Sonrakinin Düğüm2 olduğunu ve sayının 1 olduğunu unutmayın.

graph.get_state(states[-3])

Zamanda Geriye Gitmek

Zamanda geriye gitmek için bu durumu çağırmada kullanın. Şu current_state olarak “states[-3]” kullandığına ve düğüm2'ye devam ettiğine dikkat edin

graph.invoke(None, states[-3])

Yeni durumların artık durum geçmişinde olduğuna dikkat edin. En sağdaki sayımlara dikkat edin.

thread = {"configurable": {"thread_id": str(1)}}
for state in graph.get_state_history(thread):
    print(state.config, state.values['count'])

Aşağıda ayrıntıları görebilirsiniz. Çok fazla metin var ancak yeni dalı başlatan düğümü bulmaya çalışın. Ana yapılandırmanın yığındaki önceki girdi olmadığına, ancak “state[-3]” girdi olduğuna dikkat edin.

thread = {"configurable": {"thread_id": str(1)}}
for state in graph.get_state_history(thread):
    print(state,"\n")

Durum Değişikliği

Yeni bir konu başlatıp geçmişi temizlemek ve çalıştırmak için başlayalım.

thread2 = {"configurable": {"thread_id": str(2)}}
graph.invoke({"count":0, "scratch":"hi"},thread2)

from IPython.display import Imag
Image(graph.get_graph().draw_png())

states2 = []
for state in graph.get_state_history(thread2):
    states2.append(state.config)
    print(state.config, state.values['count'])   

Bir durumu ele alarak başlayın.

save_state = graph.get_state(states2[-3])
save_state

Şimdi değerleri değiştirin. Dikkat edilmesi gereken bir nokta: Ajan durumunun ne zaman tanımlandığını hatırlayın, değerlerin mevcut değere eklendiğini belirtmek için sayım operator.add’i kullandı. Burada -3, mevcut sayım değeri ile değiştirilmek yerine eklenecektir.

save_state.values["count"] = -3
save_state.values["scratch"] = "hello"
save_state

Şimdi durumu güncelleyin. Bu, üstte yeni bir girdi veya hafızadaki en son girdiyi oluşturur. Bu mevcut durum haline gelecektir.

graph.update_state(thread2,save_state.values)

Mevcut durum en üstte. thread_ts ile eşleştirebilirsiniz. Yeni düğümün parent_config, thread_ts değerlerine dikkat edin; bu önceki düğümdür.

for i, state in enumerate(graph.get_state_history(thread2)):
    if i >= 3:  #print latest 3
        break
    print(state, '\n')

as_node ile tekrar deneme

update_state() kullanarak yazarken, graf mantığına hangi düğümün yazar olarak kabul edilmesi gerektiğini tanımlamak istersiniz. Bunun yaptığı şey, graf mantığının graftaki düğümü bulmasına izin vermektir. Değerler yazıldıktan sonra next() değeri, yeni durum kullanılarak grafiğin üzerinden geçilerek hesaplanır. Bu durumda sahip olduğumuz durum Node1 tarafından yazılmıştır. Graf daha sonra bir sonraki durumu Düğüm2 olarak hesaplayabilir. Bazı graflarda bunun koşullu kenarlardan geçmeyi içerebileceğini unutmayın! Hadi bunu deneyelim.

graph.update_state(thread2,save_state.values, as_node="Node1")

for i, state in enumerate(graph.get_state_history(thread2)):
    if i >= 3:  #print latest 3
        break
    print(state, '\n')

invoke, belirli bir thread_ts verilmediği takdirde mevcut durumdan çalıştırılacaktır. Bu şimdi yeni eklenen girdidir.

graph.invoke(None,thread2)

Durum geçmişini yazdırın, en son girdilerdeki çizik (scratch) değeri değişimine dikkat edin.

for state in graph.get_state_history(thread2):
    print(state,"\n")

Denemeler yapmaya devam edin.

Deneme Yazarı

Bir yapay zeka araştırmacısının, bizim senaryomuzda bir makale yazarının kompakt bir versiyonunu oluşturacağız.

plan: You are an expert writer tasked with writing a high level outline of an essay. Write such an outline for the user provided topic. Give an outline of the essay along with any relevant notes or instructions for the sections.
research_plan: You are a researcher charged with providing information that can be used when writing the following essay. Generate a list of search queries that will gather any relevant information. Only generate 3 queries max.
generate: You are an essay assistant tasked with writing excellent 5-paragraph essays. Generate … if the user provides critique , respond with a revised version of your previous attempts. Utilize information:

from dotenv import load_dotenv
_ = load_dotenv()

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, AIMessage, ChatMessage
memory = SqliteSaver.from_conn_string(":memory:")

class AgentState(TypedDict):
    task: str
    plan: str
    draft: str
    critique: str
    content: List[str]
    revision_number: int
    max_revisions: int

from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

PLAN_PROMPT = """You are an expert writer tasked with writing a high level outline of an essay. \
Write such an outline for the user provided topic. Give an outline of the essay along with any relevant notes \
or instructions for the sections."""

WRITER_PROMPT = """You are an essay assistant tasked with writing excellent 5-paragraph essays.\
Generate the best essay possible for the user's request and the initial outline. \
If the user provides critique, respond with a revised version of your previous attempts. \
Utilize all the information below as needed: 
------
{content}"""

REFLECTION_PROMPT = """You are a teacher grading an essay submission. \
Generate critique and recommendations for the user's submission. \
Provide detailed recommendations, including requests for length, depth, style, etc."""

RESEARCH_PLAN_PROMPT = """You are a researcher charged with providing information that can \
be used when writing the following essay. Generate a list of search queries that will gather \
any relevant information. Only generate 3 queries max."""

RESEARCH_CRITIQUE_PROMPT = """You are a researcher charged with providing information that can \
be used when making any requested revisions (as outlined below). \
Generate a list of search queries that will gather any relevant information. Only generate 3 queries max."""

from langchain_core.pydantic_v1 import BaseModel
class Queries(BaseModel):
    queries: List[str]

from tavily import TavilyClient
import os
tavily = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

def plan_node(state: AgentState):
    messages = [
        SystemMessage(content=PLAN_PROMPT), 
        HumanMessage(content=state['task'])
    ]
    response = model.invoke(messages)
    return {"plan": response.content}

def research_plan_node(state: AgentState):
    queries = model.with_structured_output(Queries).invoke([
        SystemMessage(content=RESEARCH_PLAN_PROMPT),
        HumanMessage(content=state['task'])
    ])
    content = state['content'] or []
    for q in queries.queries:
        response = tavily.search(query=q, max_results=2)
        for r in response['results']:
            content.append(r['content'])
    return {"content": content}

def generation_node(state: AgentState):
    content = "\n\n".join(state['content'] or [])
    user_message = HumanMessage(
        content=f"{state['task']}\n\nHere is my plan:\n\n{state['plan']}")
    messages = [
        SystemMessage(
            content=WRITER_PROMPT.format(content=content)
        ),
        user_message
        ]
    response = model.invoke(messages)
    return {
        "draft": response.content, 
        "revision_number": state.get("revision_number", 1) + 1
    }

def reflection_node(state: AgentState):
    messages = [
        SystemMessage(content=REFLECTION_PROMPT), 
        HumanMessage(content=state['draft'])
    ]
    response = model.invoke(messages)
    return {"critique": response.content}

def research_critique_node(state: AgentState):
    queries = model.with_structured_output(Queries).invoke([
        SystemMessage(content=RESEARCH_CRITIQUE_PROMPT),
        HumanMessage(content=state['critique'])
    ])
    content = state['content'] or []
    for q in queries.queries:
        response = tavily.search(query=q, max_results=2)
        for r in response['results']:
            content.append(r['content'])
    return {"content": content}

def should_continue(state):
    if state["revision_number"] > state["max_revisions"]:
        return END
    return "reflect"

builder = StateGraph(AgentState)

builder.add_node("planner", plan_node)
builder.add_node("generate", generation_node)
builder.add_node("reflect", reflection_node)
builder.add_node("research_plan", research_plan_node)
builder.add_node("research_critique", research_critique_node)

builder.set_entry_point("planner")

builder.add_conditional_edges(
    "generate", 
    should_continue, 
    {END: END, "reflect": "reflect"}
)

builder.add_edge("planner", "research_plan")
builder.add_edge("research_plan", "generate")
builder.add_edge("reflect", "research_critique")
builder.add_edge("research_critique", "generate")

graph = builder.compile(checkpointer=memory)

from IPython.display import Image
Image(graph.get_graph().draw_png())

thread = {"configurable": {"thread_id": "1"}}
for s in graph.stream({
    'task': "what is the difference between langchain and langsmith",
    "max_revisions": 2,
    "revision_number": 1,
}, thread):
    print(s)

Deneme Yazarı Arayüzü

import warnings
warnings.filterwarnings("ignore")
from helper import ewriter, writer_gui

MultiAgent = ewriter()
app = writer_gui(MultiAgent.graph)
app.launch()

LangChain Kaynakları

Langchain dokümantasyonu: https://python.langchain.com/v0.2/docs/introduction/

LangServe: https://python.langchain.com/v0.1/docs/langserve/

LangSmith: https://docs.smith.langchain.com/

LangGraph: https://langchain-ai.github.io/langgraph/

Sonuç

Bu kursta oluşturamadığımız ancak bilmeniz gereken bazı aracı akışlarından bahsedeceğiz.

Çoklu Ajan Mimarisi (Multi-agent Architecture)

Çok Ajanlı Mimari

Bu ajanı yazarın ajanı yaparken işlemiştik. Çoklu ajan mimarisi, birden fazla ajanın aynı paylaşılan durum üzerinde çalıştığı mimaridir. Bu etmenler yazar ajanında olduğu gibi sadece bir istem ve bir dil modeli barındırabilirler. Ayrıca bunları çağırabilecekleri farklı araçlara da sahip olabilirler. İçlerinde döngülere sahip olabilirler. Tüm ajanlar aynı paylaşılan durumda çalışır. Böylece etmenler aynı paylaşılan durumu bir etmenden diğerine aktarırlar.

Denetleyici Mimari (The Supervisor Architecture)

Bu mimaride bazı alt ajanları çağıran bir denetleyici ajanımız bulunmaktadır. Burada denetleyici, bu ajana gelecek girdinin ne olduğunu belirler. Ayrıca bu ajanların kendi içlerinde farklı durumları olabilir, bunlar graflardır. Ve mutlaka aynı ortak durum kavramının olması da gerekmez. Bunun dışında denetleyici mimari, çoklu etmen mimarisine oldukça benzer. Tek önemli fark, işçi ajanlarından sorumlu bir denetleyicinin olmasıdır. Planlama çok fazla zeka gerektirdiğinden, yönetici olarak daha akıllı bir ajan kullanmak isteyebilirsiniz.

Akış Mühendisliği Mimarisi (Flow Engineering Architecture)

Bu mimari, araştırmacıların bir döngü içeren işlem hattına benzer yapı gibi graf bir çözümle en son teknoloji kodlama performansını elde ettiği AlphaCodium makalesinde tanıtılmıştır.

Akış mühendisliği mimarisi, ilk kod çözümünde, genel test yineleme bölümünde ve yapay zeka görev yineleme bölümünde döngüler içerir. Bu ilginç bir graftır çünkü belli bir noktaya kadar çok yönlü bir akış bulunmaktadır. Bu, belirli bir kodlama sorununa yönelik özel bir mimaridir ve genellikle ajanlarınızın harekete geçmesi ve düşünmesi için doğru bilgi akışının ne olduğunu düşünmeyi ifade eder.

Planlama ve Uygulama

Bu doğrultuda, yaygın bir paradigma, bir planlama ve uygulama tarzı akışına sahip olmaktır; burada ilk önce açık bir planlama adımı atarsınız ve bu planı uygulamaya başlarsınız. Böylece bir alt ajanın yapması gereken birkaç adımı belirleyebilirsiniz. Daha sonra ajan gidip bir eylem gerçekleştirip geri gelir. Sonrasında planı güncelleyebilirsiniz, bu değişiklikler bu mimarinin farklı varyantları arasında değişiklik gösterir. Ajan ardından bir sonraki göreve geçer ve bunu gerçekleştirir ve planı bitirene kadar geri gelir. Daha sonra planın başarılı olup olmadığını veya yeniden planlamanız gerekip gerekmediğini kontrol edebiliriz.

Dil Ajanı Ağacı Arama Mimarisi (Language Agent Tree Search Architecture)

Bu, olası eylem durumları üzerinde bir ağaç araması yapar. LATS önce bir eylem üretir. Sonra bu eyleme dayanarak aşağı iner ve bu eylem hakkında yansıma yapmak (reflection) için başka alt eylemler üretir. Tüm bu yansımalar aracılığıyla, hangi eylem durumu ağacında geriye gitmek istediğine dair düşünebilir. Bu şekilde geriye doğru yayılım yapabilir ve üst düğümleri daha fazla bilgiyle güncelleyebilir, bu da geçmiş durumdan gelecekteki yönlendirmeleri etkileyebilir. Özellikle bu durumda, neden sürekliliğin önemli olduğunu görebilirsiniz, çünkü süreklilik sayesinde önceki durumlara geri gitme yeteneğine sahip olmanız gerekmektedir.

LangGraph, yüksek kontrol edilebilirlik hedefleyerek ve döngüsel veya döngüsüz akışlar oluşturmanıza izin vererek tasarlanmıştır. Bu kontrol edilebilirlik, LangGraph’ı diğer çerçevelerden ayıran şeydir.

Kaynak

[1] Deeplearning.ai, (2024), AI Agents in LangGraph:

[https://learn.deeplearning.ai/courses/ai-agents-in-langgraph/]