File size: 4,929 Bytes
0e02b97
 
 
ec335c4
0e02b97
ccb1848
0e02b97
86c5368
 
 
 
 
 
 
27b075e
0e02b97
 
ecc9e42
0e02b97
 
 
 
 
27b075e
 
 
 
 
 
 
bedb8e2
0e02b97
 
bf45c7d
27b075e
 
 
 
86c5368
0e02b97
 
 
 
bedb8e2
 
 
 
86c5368
 
 
 
 
 
 
 
 
27b075e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
ec335c4
 
ccb1848
ec335c4
 
 
 
 
 
 
 
27b075e
ec335c4
0e02b97
 
 
 
 
ecc9e42
4c7374d
0e02b97
 
 
 
 
 
bedb8e2
0e02b97
 
 
 
 
 
ecc9e42
 
7bd7366
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0e02b97
 
 
 
27b075e
 
 
 
 
 
0e02b97
 
 
27b075e
 
 
0e02b97
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
from __future__ import annotations

from typing import List
import json

from ollama import AsyncClient, ChatResponse, Message

from .config import (
    MAX_TOOL_CALL_DEPTH,
    MODEL_NAME,
    NUM_CTX,
    OLLAMA_HOST,
    SYSTEM_PROMPT,
)
from .db import Conversation, Message as DBMessage, User, _db, init_db
from .log import get_logger
from .schema import Msg
from .tools import execute_terminal

_LOG = get_logger(__name__)


class ChatSession:
    def __init__(
        self,
        user: str = "default",
        session: str = "default",
        host: str = OLLAMA_HOST,
        model: str = MODEL_NAME,
    ) -> None:
        init_db()
        self._client = AsyncClient(host=host)
        self._model = model
        self._user, _ = User.get_or_create(username=user)
        self._conversation, _ = Conversation.get_or_create(
            user=self._user, session_name=session
        )
        self._messages: List[Msg] = self._load_history()
        self._ensure_system_prompt()

    async def __aenter__(self) -> "ChatSession":
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        if not _db.is_closed():
            _db.close()

    def _ensure_system_prompt(self) -> None:
        if any(m.get("role") == "system" for m in self._messages):
            return

        DBMessage.create(
            conversation=self._conversation, role="system", content=SYSTEM_PROMPT
        )
        self._messages.insert(0, {"role": "system", "content": SYSTEM_PROMPT})

    def _load_history(self) -> List[Msg]:
        messages: List[Msg] = []
        for msg in self._conversation.messages.order_by(DBMessage.created_at):
            if msg.role == "assistant":
                try:
                    calls = json.loads(msg.content)
                except json.JSONDecodeError:
                    messages.append({"role": "assistant", "content": msg.content})
                else:
                    messages.append(
                        {
                            "role": "assistant",
                            "tool_calls": [Message.ToolCall(**c) for c in calls],
                        }
                    )
            elif msg.role == "user":
                messages.append({"role": "user", "content": msg.content})
            else:
                messages.append({"role": "tool", "content": msg.content})
        return messages

    @staticmethod
    def _store_assistant_message(
        conversation: Conversation, message: Message
    ) -> None:
        """Persist assistant messages, storing tool calls when present."""

        if message.tool_calls:
            content = json.dumps([c.model_dump() for c in message.tool_calls])
        else:
            content = message.content or ""

        DBMessage.create(conversation=conversation, role="assistant", content=content)

    async def ask(self, messages: List[Msg], *, think: bool = True) -> ChatResponse:
        return await self._client.chat(
            self._model,
            messages=messages,
            think=think,
            tools=[execute_terminal],
            options={"num_ctx": NUM_CTX},
        )

    async def _handle_tool_calls(
        self,
        messages: List[Msg],
        response: ChatResponse,
        conversation: Conversation,
        depth: int = 0,
    ) -> ChatResponse:
        if depth >= MAX_TOOL_CALL_DEPTH or not response.message.tool_calls:
            return response

        for call in response.message.tool_calls:
            if call.function.name == "execute_terminal":
                result = execute_terminal(**call.function.arguments)
            else:
                continue

            messages.append(
                {
                    "role": "tool",
                    "name": call.function.name,
                    "content": str(result),
                }
            )
            DBMessage.create(
                conversation=conversation,
                role="tool",
                content=str(result),
            )
            nxt = await self.ask(messages, think=True)
            self._store_assistant_message(conversation, nxt.message)
            return await self._handle_tool_calls(
                messages, nxt, conversation, depth + 1
            )

        return response

    async def chat(self, prompt: str) -> str:
        DBMessage.create(conversation=self._conversation, role="user", content=prompt)
        self._messages.append({"role": "user", "content": prompt})

        response = await self.ask(self._messages)
        self._messages.append(response.message.model_dump())
        self._store_assistant_message(self._conversation, response.message)

        _LOG.info("Thinking:\n%s", response.message.thinking or "<no thinking trace>")

        final_resp = await self._handle_tool_calls(
            self._messages, response, self._conversation
        )
        return final_resp.message.content