-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Description
Initial Checks
- I confirm that I'm using the latest version of Pydantic AI
- I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue
Description
so im trying to build a django-chatbot-app powered with htmx websocket. when im trying to import the simple agent model into my django websocket app, i got an error of ssl.SSLError: unknown error (_ssl.c:3036). here is the snippets of the code.
then when im importing the agent variable into the websocket of consumers.py, even that im just importing it, i got the error ssl.SSLError: unknown error (_ssl.c:3036)
im using linux Fedora 42 for the OS,
the thing is, when i try it in as unittest, it all works just fine as intended
i tried different ai client, from deepseek and google gemini flash, they both suffer the same issue when importing it to the django app.
NOTE: i will try to ask about this in django community too in the sametime
agents.py
from pydantic_ai import Agent
import os
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.deepseek import DeepSeekProvider
model = OpenAIModel(
"deepseek-chat",
provider=DeepSeekProvider(api_key=os.getenv("DEEPSEEK_API_KEY")),
)
agent = Agent(model)consumers.py
from channels.generic.websocket import WebsocketConsumer
from uuid import UUID
from .agents import agent
from .models import ChatRoom, ChatMessage
from pprint import pprint as print
from django.contrib.auth.models import User
from channels.auth import UserLazyObject
from django.template.loader import render_to_string
import json
class ChatConsumer(WebsocketConsumer):
def connect(self):
# self.scope: dict = self.scope
self.user: User = self.scope["user"]
self.room_uuid = UUID(
self.scope.get("url_route").get("kwargs").get("room_uuid")
)
# print(self.scope)
# print(self.room_uuid)
# print(type(self.room_uuid))
# self.messages = list(ChatMessage.objects.filter(room__user=self.user))
self.chat_room, _ = ChatRoom.objects.get_or_create(
user=self.user, room_uuid=self.room_uuid
)
# TODO: add way to load previous messages by uuid
self.messages: list[ChatMessage] = self.load_messages()
print(self.messages)
return self.accept() if self.chat_room.user == self.user else self.close()
def receive(self, text_data=None, bytes_data=None):
print(text_data)
text_json = json.loads(str(text_data))
print(text_json)
# print(self.scope)
# print(self.scope["user"])
# print(type(self.scope["user"]))
# self.messages.append(text_json)
human_message = ChatMessage(
room=self.chat_room,
content=text_json["message"],
owner=ChatMessage.OwnerEnum.USER,
)
bot_message = ChatMessage(
room=self.chat_room,
content="",
owner=ChatMessage.OwnerEnum.AI,
)
self.messages.append(human_message)
self.messages.append(bot_message)
# user_message_html = render_to_string(
# "chat/partial_chat.html",
# {"message_text": text_json["message"], "is_system": False},
# )
user_message_html = render_to_string(
"cotton/htmxhumanmessage.html",
context={"message": text_json["message"], "is_system": False},
)
bot_message_html = render_to_string(
"cotton/htmxhumanmessage.html",
context={"message": "this is bot reply message", "is_system": True},
)
self.send(text_data=user_message_html)
self.send(text_data=bot_message_html)
def close(self, code=None, reason=None):
print("final result is...")
if len(self.messages) > 0:
print(code)
self.save_messages()
return super().disconnect(code)
self.delete_chatroom()
return super().close(code, reason)
def disconnect(self, code):
print("final result is...")
if len(self.messages) > 0:
self.save_messages()
return super().disconnect(code)
self.delete_chatroom()
print("success disconnect connection")
return super().disconnect(code)
def save_messages(self):
for m in self.messages:
m.save()
def load_messages(self) -> list[ChatMessage]:
try:
chat_room = ChatRoom.objects.get(room_uuid=self.room_uuid)
messages = list(
ChatMessage.objects.filter(room=chat_room).order_by("created_at")
)
if len(messages) > 0:
return messages
except ChatRoom.DoesNotExist:
pass
return []
def delete_chatroom(self) -> None:
self.chat_room.delete()unittest snipet code, run it with python -m unittest tests.TestAgent
tests.py
from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel
from unittest import TestCase
from dotenv import load_dotenv
load_dotenv()
class TestAgent(TestCase):
def test_agent_multiple_chat(self):
"""
NOTE: use gemini flash
"""
model = GeminiModel("gemini-2.0-flash")
agent = Agent(model, system_prompt="You are a very helpful assistant")
result = agent.run_sync("tell me a joke for bapak2 in bahasa indonesia")
print(result.output)
result2 = agent.run_sync(
"Jelaskan maksudnya", message_history=result.new_messages()
)
print(result2.output)Example Code
# this script is combined version of agents.py and consumers.py
from pydantic_ai import Agent
import os
from dotenv import load_dotenv
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.deepseek import DeepSeekProvider
load_dotenv()
model = OpenAIModel(
"deepseek-chat",
provider=DeepSeekProvider(api_key=os.getenv("DEEPSEEK_API_KEY")),
)
agent = Agent(model)
from channels.generic.websocket import WebsocketConsumer
from uuid import UUID
from .agents import agent
from .models import ChatRoom, ChatMessage
from pprint import pprint as print
from django.contrib.auth.models import User
from channels.auth import UserLazyObject
from django.template.loader import render_to_string
import json
class ChatConsumer(WebsocketConsumer):
def connect(self):
# self.scope: dict = self.scope
self.user: User = self.scope["user"]
self.room_uuid = UUID(
self.scope.get("url_route").get("kwargs").get("room_uuid")
)
self.chat_room, _ = ChatRoom.objects.get_or_create(
user=self.user, room_uuid=self.room_uuid
)
# TODO: add way to load previous messages by uuid
self.messages: list[ChatMessage] = self.load_messages()
return self.accept() if self.chat_room.user == self.user else self.close()
def receive(self, text_data=None, bytes_data=None):
print(text_data)
text_json = json.loads(str(text_data))
print(text_json)
human_message = ChatMessage(
room=self.chat_room,
content=text_json["message"],
owner=ChatMessage.OwnerEnum.USER,
)
bot_message = ChatMessage(
room=self.chat_room,
content="",
owner=ChatMessage.OwnerEnum.AI,
)
self.messages.append(human_message)
self.messages.append(bot_message)
user_message_html = render_to_string(
"cotton/htmxhumanmessage.html",
context={"message": text_json["message"], "is_system": False},
)
bot_message_html = render_to_string(
"cotton/htmxhumanmessage.html",
context={"message": "this is bot reply message", "is_system": True},
)
self.send(text_data=user_message_html)
self.send(text_data=bot_message_html)
def close(self, code=None, reason=None):
print("final result is...")
if len(self.messages) > 0:
print(code)
self.save_messages()
return super().disconnect(code)
self.delete_chatroom()
return super().close(code, reason)
def disconnect(self, code):
print("final result is...")
if len(self.messages) > 0:
self.save_messages()
return super().disconnect(code)
self.delete_chatroom()
print("success disconnect connection")
return super().disconnect(code)
def save_messages(self):
for m in self.messages:
m.save()
def load_messages(self) -> list[ChatMessage]:
try:
chat_room = ChatRoom.objects.get(room_uuid=self.room_uuid)
messages = list(
ChatMessage.objects.filter(room=chat_room).order_by("created_at")
)
if len(messages) > 0:
return messages
except ChatRoom.DoesNotExist:
pass
return []
def delete_chatroom(self) -> None:
self.chat_room.delete()Python, Pydantic AI & LLM client version
python 3.13.3
pydantic 2.11.4
pydantic-ai 0.1.9
pydantic-ai-examples 0.1.9
pydantic-ai-slim 0.1.9
pydantic-core 2.33.2
pydantic-evals 0.1.9
pydantic-graph 0.1.9
pydantic-settings 2.9.1
deepseek ai chat client api
gemini flash ai chat client api