Как развернуть автономный рой AI-агентов локально: LangGraph + Ollama

# macOS / Linux: установка Ollama описана на сайте проекта
ollama pull qwen3:8b

python -m venv .venv
source .venv/bin/activate
pip install -U langgraph langchain-ollama langchain-core langgraph-checkpoint-sqlite

ollama run qwen3:8b

from __future__ import annotations

import json
import operator
import os
from pathlib import Path
from typing import Annotated, Literal, TypedDict

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_ollama import ChatOllama
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph

MODEL = os.environ.get('LOCAL_MODEL', 'qwen3:8b')
WORKSPACE = Path(os.environ.get('SWARM_WORKSPACE', '.')).resolve()
MAX_ITERATIONS = int(os.environ.get('MAX_ITERATIONS', '6'))

llm = ChatOllama(
    model=MODEL,
    base_url=os.environ.get('OLLAMA_BASE_URL', 'http://localhost:11434'),
    temperature=0.2,
    num_ctx=8192,
)

json_llm = ChatOllama(
    model=MODEL,
    base_url=os.environ.get('OLLAMA_BASE_URL', 'http://localhost:11434'),
    temperature=0,
    format='json',
    num_ctx=8192,
)

class SwarmState(TypedDict):
    goal: str
    task: str
    draft: str
    findings: Annotated[list[str], operator.add]
    decisions: Annotated[list[str], operator.add]
    errors: Annotated[list[str], operator.add]
    iterations: int
    next: str

def call_model(system: str, prompt: str, *, json_mode: bool = False) -> str:
    model = json_llm if json_mode else llm
    message = model.invoke([
        SystemMessage(content=system),
        HumanMessage(content=prompt),
    ])
    return str(message.content).strip()

def parse_json_object(text: str) -> dict:
    start = text.find('{')
    end = text.rfind('}')
    if start == -1 or end == -1 or end <= start:
        return {}
    try:
        return json.loads(text[start:end + 1])
    except json.JSONDecodeError:
        return {}

def safe_path(relative_path: str) -> Path:
    candidate = (WORKSPACE / relative_path).resolve()
    if candidate != WORKSPACE and WORKSPACE not in candidate.parents:
        raise ValueError('Path escapes workspace')
    return candidate

def local_search(query: str, limit: int = 8) -> list[str]:
    allowed_suffixes = {'.md', '.txt', '.py', '.json', '.yaml', '.yml'}
    terms = [term.lower() for term in query.split() if len(term) > 2]
    results: list[str] = []

    for item in WORKSPACE.rglob('*'):
        if len(results) >= limit:
            break
        if item.is_dir() or item.name.startswith('.'):
            continue
        if item.suffix.lower() not in allowed_suffixes:
            continue
        try:
            text = item.read_text(errors='ignore')[:12000]
        except OSError:
            continue
        lower = text.lower()
        if not terms or any(term in lower for term in terms):
            rel = item.relative_to(WORKSPACE)
            snippet = text.replace('\n', ' ')[:700]
            results.append(f'{rel}: {snippet}')
    return results

def planner(state: SwarmState) -> dict:
    if state.get('iterations', 0) >= MAX_ITERATIONS:
        return {'next': 'final', 'task': 'Finish because iteration limit was reached'}

    prompt = f'''
Goal: {state['goal']}
Current task: {state.get('task', '')}
Findings count: {len(state.get('findings', []))}
Draft exists: {bool(state.get('draft'))}
Recent errors: {state.get('errors', [])[-3:]}

Choose exactly one next step: researcher, builder, critic, final.
Return only JSON with keys next and task.
'''
    raw = call_model('You are a strict workflow planner for a local multi-agent system.', prompt, json_mode=True)
    data = parse_json_object(raw)
    next_step = data.get('next', 'researcher')
    if next_step not in {'researcher', 'builder', 'critic', 'final'}:
        next_step = 'researcher'
    return {'next': next_step, 'task': str(data.get('task', 'Continue the task'))}

def researcher(state: SwarmState) -> dict:
    query = state.get('task') or state['goal']
    hits = local_search(query)
    if not hits:
        return {
            'findings': ['No matching local files found. Continue using only the user goal and existing state.'],
            'iterations': state.get('iterations', 0) + 1,
        }

    prompt = 'Summarize the relevant facts from these local snippets. Do not invent facts.\n\n' + '\n\n'.join(hits)
    summary = call_model('You are a local research agent. Use only provided snippets.', prompt)
    return {
        'findings': [summary],
        'decisions': [f'Researcher searched local workspace for: {query}'],
        'iterations': state.get('iterations', 0) + 1,
    }

def builder(state: SwarmState) -> dict:
    prompt = f'''
Goal: {state['goal']}
Task: {state.get('task', '')}
Findings:
{chr(10).join(state.get('findings', [])[-5:])}

Create a concrete answer or implementation plan. Mark assumptions explicitly.
'''
    draft = call_model('You are a builder agent. Produce useful, testable output.', prompt)
    return {
        'draft': draft,
        'decisions': ['Builder produced or updated the draft'],
        'iterations': state.get('iterations', 0) + 1,
    }

def critic(state: SwarmState) -> dict:
    prompt = f'''
Goal: {state['goal']}
Draft:
{state.get('draft', '')}

Find factual risks, missing checks, security issues and unclear assumptions.
If the draft is good enough, say PASS and list only the remaining caveats.
'''
    critique = call_model('You are a skeptical reviewer. Be concise and specific.', prompt)
    return {
        'errors': [critique],
        'decisions': ['Critic reviewed the draft'],
        'iterations': state.get('iterations', 0) + 1,
    }

def finalizer(state: SwarmState) -> dict:
    prompt = f'''
Goal: {state['goal']}
Draft:
{state.get('draft', '')}
Critique:
{chr(10).join(state.get('errors', [])[-3:])}

Produce the final answer. Keep caveats that matter. Do not mention internal agent names.
'''
    final = call_model('You are the final editor of a local autonomous agent run.', prompt)
    return {'draft': final, 'next': 'done'}

def route(state: SwarmState) -> Literal['researcher', 'builder', 'critic', 'final']:
    value = state.get('next', 'researcher')
    if value in {'researcher', 'builder', 'critic', 'final'}:
        return value
    return 'final'

def build_graph(checkpointer):
    graph = StateGraph(SwarmState)
    graph.add_node('planner', planner)
    graph.add_node('researcher', researcher)
    graph.add_node('builder', builder)
    graph.add_node('critic', critic)
    graph.add_node('final', finalizer)

    graph.add_edge(START, 'planner')
    graph.add_conditional_edges('planner', route)
    graph.add_edge('researcher', 'planner')
    graph.add_edge('builder', 'planner')
    graph.add_edge('critic', 'planner')
    graph.add_edge('final', END)
    return graph.compile(checkpointer=checkpointer)

if __name__ == '__main__':
    goal = os.environ.get('SWARM_GOAL') or 'Analyze the local project and suggest three high-impact improvements.'
    initial_state: SwarmState = {
        'goal': goal,
        'task': '',
        'draft': '',
        'findings': [],
        'decisions': [],
        'errors': [],
        'iterations': 0,
        'next': 'researcher',
    }

    config = {'configurable': {'thread_id': 'local-swarm-demo'}}

    with SqliteSaver.from_conn_string('swarm-checkpoints.sqlite') as checkpointer:
        app = build_graph(checkpointer)
        result = app.invoke(initial_state, config=config)
        print('=== FINAL RESULT ===')
        print(result['draft'])

SWARM_WORKSPACE=./docs \
SWARM_GOAL='Найди пробелы в документации и предложи план улучшений' \
python local_swarm.py