Проектирование продвинутой архитектуры агентного ИИ

Обзор

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

Основные компоненты

Объединяя структурированное управление состоянием, извлечение с учетом памяти, рефлексивное обучение и контролируемое использование инструментов, мы демонстрируем, как современные агентные системы могут размышлять, действовать, учиться и эволюционировать, а не отвечать одним проходом.

Настройка среды

Чтобы настроить среду выполнения, мы устанавливаем все необходимые библиотеки и импортируем основные модули. Мы интегрируем LangGraph для оркестрации, LangChain для абстракций моделей и инструментов, и вспомогательные библиотеки для графов памяти и числовых операций.

!pip -q install -U langgraph langchain-openai langchain-core pydantic numpy networkx requests
 
import os, getpass, json, time, operator
from typing import List, Dict, Any, Optional, Literal
from typing_extensions import TypedDict, Annotated
import numpy as np
import networkx as nx
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage, AnyMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

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

Мы безопасно загружаем API-ключ OpenAI во время выполнения и инициализируем языковые модели, используемые для быстрого, глубокого и рефлексивного рассуждения. Мы также настраиваем модель встраивания, которая обеспечивает семантическое сходство в памяти.

if not os.environ.get("OPENAI_API_KEY"):
   os.environ["OPENAI_API_KEY"] = getpass.getpass("Введите OPENAI_API_KEY: ")
 
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
EMB_MODEL = os.environ.get("OPENAI_EMBED_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)

Создание агентного графа памяти

Мы строим агентный граф памяти, вдохновленный методом Zettelkasten, где каждое взаимодействие хранится как атомарная заметка.

class Note(BaseModel):
   note_id: str
   title: str
   content: str
   tags: List[str] = Field(default_factory=list)
   created_at_unix: float
   context: Dict[str, Any] = Field(default_factory=dict)
 
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.add_node(note.note_id, **note.model_dump())
       self.note_vectors[note.note_id] = vec
 
   def topk_related(self, vec, k=5):
       scored = [(nid, self._cos(vec, v)) for nid, v in self.note_vectors.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.link_note(nid, r["note_id"], r["score"], "evolve")
MEM = MemoryGraph()

Реализация внешних инструментов

Мы определяем внешние инструменты, которые агент может вызывать, такие как веб-доступ и поиск по памяти.

@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 memory_neighbors(note_id: str) -> str:
   if note_id not in MEM.g:
       return "[]"
   return json.dumps([
       {"note_id": n, "weight": MEM.g[note_id][n]["weight"]}
       for n in MEM.g.neighbors(note_id)
   ])
TOOLS = [web_get, memory_search, memory_neighbors]
TOOLS_BY_NAME = {t.name: t for t in TOOLS}

Основные агентные действия

Основные агентные действия, включая делиберацию, действие, выполнение инструментов, финализацию и рефлексию, представлены как узлы LangGraph.

def deliberate(st):
   spec = RunSpec.model_validate(st["run_spec"])
   d = llm_fast.with_structured_output(DeliberationDecision).invoke([
       SystemMessage(content=DECIDER_SYS),
       HumanMessage(content=json.dumps(spec.model_dump()))
   ])
   return {"decision": d.model_dump(), "budget_calls_remaining": st["budget_calls_remaining"] - 1}

Заключение

Мы продемонстрировали, как системы агентного ИИ могут приблизиться к автономии, адаптируя глубину своего мышления и повторно используя предыдущее знание. Этот подход представляет собой практическую основу для создания масштабируемых, самосовершенствующихся агентов в реальных приложениях.

Для полной реализации смотрите ПОЛНЫЕ КОДЫ здесь.