File size: 4,830 Bytes
b4f404b | 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 151 152 153 | import os
import requests
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from typing import List, Optional
def generate_completion(messages, model="gpt-4.1-nano", max_tokens=1000, temperature=0.7):
"""
Generate a response from Euron API
"""
euron_api_key = os.getenv("EURON_API_KEY", "").strip()
if not euron_api_key:
raise ValueError("EURON_API_KEY is missing. Please set it in your environment.")
url = "https://api.euron.one/api/v1/euri/chat/completions"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {euron_api_key}"
}
# Convert LangChain messages to API format
api_messages = []
for message in messages:
if hasattr(message, 'type'):
role = message.type
if role == "human":
role = "user"
elif role == "ai":
role = "assistant"
api_messages.append({"role": role, "content": message.content})
else:
api_messages.append(message)
payload = {
"messages": api_messages,
"model": model,
"max_tokens": max_tokens,
"temperature": temperature
}
response = requests.post(url, headers=headers, json=payload)
if response.status_code != 200:
raise Exception(f"Euron API Error: {response.status_code} {response.text}")
return response.json()
class EuronChatModel(BaseChatModel):
"""
LangChain compatible chat model using Euron API
"""
model_name: str = "gpt-4.1-nano"
def _generate(self, messages: List, stop: Optional[List[str]] = None) -> ChatResult:
response = generate_completion(messages, model=self.model_name)
# Extract AI message
ai_content = response['choices'][0]['message']['content']
# Wrap in LangChain objects
ai_message = AIMessage(content=ai_content)
generation = ChatGeneration(message=ai_message)
return ChatResult(generations=[generation])
def _llm_type(self) -> str:
return "euron-chat"
# Optional: simple chat function
def simple_chat_completion(user_message: str):
messages = [{"role": "user", "content": user_message}]
response = generate_completion(messages)
return response['choices'][0]['message']['content']
# import os
# import requests
# from langchain_core.language_models import BaseChatModel
# from langchain_core.messages import AIMessage
# from langchain_core.outputs import ChatResult, ChatGeneration
# from typing import List, Optional
# EURON_API_KEY = os.environ.get("EURON_API_KEY")
# def generate_completion(messages, model="gpt-4.1-nano", max_tokens=1000, temperature=0.7):
# """
# Generate a response from Euron API
# """
# url = "https://api.euron.one/api/v1/euri/chat/completions"
# headers = {
# "Content-Type": "application/json",
# "Authorization": f"Bearer {EURON_API_KEY}"
# }
# # Convert LangChain messages to API format
# api_messages = []
# for message in messages:
# if hasattr(message, 'type'):
# role = message.type
# if role == "human":
# role = "user"
# elif role == "ai":
# role = "assistant"
# api_messages.append({"role": role, "content": message.content})
# else:
# api_messages.append(message)
# payload = {
# "messages": api_messages,
# "model": model,
# "max_tokens": max_tokens,
# "temperature": temperature
# }
# response = requests.post(url, headers=headers, json=payload)
# if response.status_code != 200:
# raise Exception(f"Euron API Error: {response.status_code} {response.text}")
# return response.json()
# class EuronChatModel(BaseChatModel):
# """
# LangChain compatible chat model using Euron API
# """
# model_name: str = "gpt-4.1-nano"
# def _generate(self, messages: List, stop: Optional[List[str]] = None) -> ChatResult:
# response = generate_completion(messages, model=self.model_name)
# # Extract AI message
# ai_content = response['choices'][0]['message']['content']
# # Wrap in LangChain objects
# ai_message = AIMessage(content=ai_content)
# generation = ChatGeneration(message=ai_message)
# return ChatResult(generations=[generation])
# def _llm_type(self) -> str:
# return "euron-chat"
# # Optional: simple chat function
# def simple_chat_completion(user_message: str):
# messages = [{"role": "user", "content": user_message}]
# response = generate_completion(messages)
# return response['choices'][0]['message']['content']
|