local-agent-core

A standalone Python framework that makes Qwen3-8B behave reliably as an agent when running via Ollama. It solves the three production problems you will hit within the first day of building a Qwen3 tool-calling loop:

Architecture is borrowed from Claude Code's QueryEngine.ts, query.ts, Task.ts, and Tool.ts.

Install

pip install local-agent-core

Prerequisites: Ollama must be running locally with Qwen3:8b pulled.

ollama serve          # terminal tab 1
ollama pull qwen3:8b  # first time only

Quick Start

from local_agent_core import AgentSession, GuardTripped

agent = AgentSession(system_prompt="You are a helpful assistant.")

@agent.tool(
    name="calculator",
    description="Evaluate a math expression",
    parameters={
        "type": "object",
        "properties": {"expression": {"type": "string"}},
        "required": ["expression"]
    }
)
def calculator(expression: str) -> str:
    return str(eval(expression, {"__builtins__": {}}))

response = agent.chat("What is 42 * 7?")   # → "The result is 294."
response2 = agent.chat("Double that.")      # history preserved automatically

Flask Integration

from flask import Flask, request, jsonify
from local_agent_core import AgentSession, GuardTripped, DiskSessionStore

app = Flask(__name__)
store = DiskSessionStore("/var/data/sessions")

SYSTEM_PROMPT = "You are a helpful assistant."

@app.route("/chat", methods=["POST"])
def chat():
    body = request.json
    session_id = body.get("session_id")   # None = new session
    message    = body["message"]

    agent = AgentSession(
        system_prompt=SYSTEM_PROMPT,
        session_id=session_id,
        store=store,
        max_turns=15,
    )
    _register_tools(agent)   # attach your tools here

    try:
        response = agent.chat(message)
        return jsonify({
            "response": response,
            "session_id": agent.session_id,
            "guards": agent.guard_summary(),
        })
    except GuardTripped as g:
        return jsonify({
            "response": f"I had to stop: {g.reason}",
            "session_id": agent.session_id,
            "guard_tripped": g.guard_name,
        })

API Reference

AgentSession

The only class you need to import in application code.

AgentSession(
    system_prompt           = "",               # Injected as role=system at turn 0
    model                   = "qwen3:8b",
    base_url                = "http://localhost:11434/v1",
    max_turns               = 20,               # Hard loop ceiling
    max_tokens_budget       = 32_000,           # Cumulative token ceiling per session
    max_tokens_per_response = 4096,             # Per-response output limit
    temperature             = 0.0,              # 0.0 = deterministic tool calling
    session_id              = None,             # Provide to resume existing session
    store                   = None,             # InMemorySessionStore by default
)

Loop Guards

Three independent loop-breakers. Any one raising GuardTripped terminates the loop.

LoopGuards(
    max_turns            = 20,      # Raise after this many model responses
    max_tokens           = 32_000,  # Raise when cumulative tokens exceed this
    repetition_threshold = 3,       # Raise when same tool+args called this many times
)

Tuning guide:

Catching guard trips:

from local_agent_core import GuardTripped

try:
    response = agent.chat(user_input)
except GuardTripped as g:
    print(g.guard_name)   # "MaxTurns" | "Budget" | "Repetition"
    print(g.reason)       # human-readable explanation

Session Stores

from local_agent_core import InMemorySessionStore, DiskSessionStore

# In-memory (default) — fast, resets on process restart
store = InMemorySessionStore()

# Disk-based — survives restarts, good for single-server deployments
store = DiskSessionStore("/var/data/sessions")

For multi-worker production (gunicorn with >1 worker), implement RedisSessionStore with the same get(), create(), save() interface as DiskSessionStore.

Why temperature=0.0?

Deterministic output means the model makes the same tool call decision given the same history. With temperature > 0, a model might randomly decide not to use a tool, making the agent unreliable.

Why strip Qwen3 reasoning?

Qwen3:8b includes a reasoning field in every response (~150–300 tokens of chain-of-thought). If re-injected into history, a 10-turn conversation wastes 2,000+ tokens on reasoning the model never needs to see again. OllamaClient.extract_message() strips it before adding to session history.

Why not LangChain / LlamaIndex?

Both add abstraction layers that hide the finish_reason state machine and make it harder to inspect and fix broken message history. This framework exposes the raw OpenAI message format directly — you always know exactly what is being sent to the model.

Observed Qwen3:8b Behaviour

Known Limitations

• requires_confirmation flag in ToolDef is defined but not yet wired into the loop.
• RedisSessionStore not yet implemented — needed for multi-worker Flask (gunicorn with >1 worker).
• No async support — OllamaClient uses httpx.Client (sync). For FastAPI or async Flask, replace with httpx.AsyncClient and add async/await to loop.py.
• BudgetGuard.max_tokens=32_000 is conservative — Qwen3:8b context window is 128k. Raise this for long research tasks.

Running the Tests

Tests run against a live Ollama instance (not mocked — that is intentional).

pip install local-agent-core
ollama serve && ollama pull qwen3:8b
python -m pytest tests/ -v

License

MIT