Проектирование архитектуры агентского ИИ с помощью LangGraph и OpenAI: использование адаптивной аргументации, графов памяти и рефлексивных

Проектирование архитектуры агентского ИИ с помощью LangGraph и OpenAI: использование адаптивной аргументации, графов памяти и рефлексивных циклов

В этом руководстве мы создадим продвинутую систему агентского ИИ, используя модели LangGraph и OpenAI, выходя за рамки простых циклов планировщика и исполнителя. Мы реализуем адаптивную аргументацию, где агент динамически выбирает между быстрым и глубоким рассуждением; граф агентской памяти в стиле Zettelkasten, который хранит атомарные знания и автоматически связывает связанные опыты; и управляемый механизм использования инструментов, который обеспечивает соблюдение ограничений во время выполнения.

Установка среды выполнения

Мы устанавливаем все необходимые библиотеки и импортируем основные модули:

LangGraph для оркестрации;

LangChain для абстракций моделей и инструментов;

Поддерживающие библиотеки для графов памяти и числовых операций.

Загрузка ключа API

```python

if not os.environ.get("OPENAIAPIKEY"):

os.environ["OPENAIAPIKEY"] = getpass.getpass("Enter OPENAIAPIKEY: ")

```

Инициализация языковых моделей

```python

MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")

EMBMODEL = os.environ.get("OPENAIEMBED_MODEL", "text-embedding-3-small")

llm_fast = ChatOpenAI(model=MODEL, temperature=0)

llm_deep = ChatOpenAI(model=MODEL, temperature=0)

llm_reflect = ChatOpenAI(model=MODEL, temperature=0)

emb = OpenAIEmbeddings(model=EMB_MODEL)

```

Класс Note

```python

class Note(BaseModel):

note_id: str

title: str

content: str

tags: List[str] = Field(default_factory=list)

createdatunix: float

context: Dict[str, Any] = Field(default_factory=dict)

```

Класс MemoryGraph

```python

class MemoryGraph:

def init(self):

self.g = nx.Graph()

self.note_vectors = {}

def _cos(self, a, b):

return float(np.dot(a, b) / ((np.linalg.norm(a) + 1e-9) * (np.linalg.norm(b) + 1e-9)))

def add_note(self, note, vec):

self.g.addnode(note.noteid, note.model_dump())

self.notevectors[note.noteid] = vec

def topk_related(self, vec, k=5):

scored = [(nid, self.cos(vec, v)) for nid, v in self.notevectors.items()]

scored.sort(key=lambda x: x[1], reverse=True)

return [{"note_id": n, "score": s, "title": self.g.nodes[n]["title"]} for n, s in scored[:k]]

def link_note(self, a, b, w, r):

if a != b:

self.g.add_edge(a, b, weight=w, reason=r)

def evolve_links(self, nid, vec):

for r in self.topk_related(vec, 8):

if r["score"] >= 0.78:

self.linknote(nid, r["noteid"], r["score"], "evolve")

MEM = MemoryGraph()

```

Инструменты

```python

@tool

def web_get(url: str) -> str:

import urllib.request

with urllib.request.urlopen(url, timeout=15) as r:

return r.read(25000).decode("utf-8", errors="ignore")

@tool

def memory_search(query: str, k: int = 5) -> str:

qv = np.array(emb.embed_query(query))

hits = MEM.topk_related(qv, k)

return json.dumps(hits, ensure_ascii=False)

@tool

def memoryneighbors(noteid: str) -> str:

if note_id not in MEM.g:

return "[]"

return json.dumps([

{"noteid": n, "weight": MEM.g[noteid][n...

Читать далее