In diesem Teil werden wir den Kern des Agenten implementieren.
Überblick
Unser Agent wird vollständig Open Source sein und kann auf jedem Rechner ausgeführt werden, auf dem Ollama installiert ist.
Implementierungsdetails
- Wir nutzen Ollama, um unser KI-Modell zu hosten.
- Unser Agent läuft auf dem gemma3:27b-Modell.
Agent-Code
Erklärung
Im folgenden Code-Snippet implementieren wir einen Trading-KI-Agenten, der:
- Kontinuierlich nach neuen Nachrichten und Marktdaten sucht (für Demonstrationszwecke simuliert).
- Eine JSON-Eingabe verarbeitet, die eine Nachrichtentranskription, Schlagzeilen und Kursdaten enthält.
- Einen sorgfältig gestalteten Systemprompt (system_message_2) verwendet, um seine Antwortstruktur zu steuern.
- Eine Gedächtnisaufzeichnung überprüft, um sicherzustellen, dass keine doppelten Benachrichtigungen gesendet werden.
- Eine JSON-formatierte Antwort mit handlungsrelevanten Anweisungen ausgibt, wie z.B. die Benachrichtigung des Benutzers oder das Hinzufügen neuer Einträge in sein Gedächtnis.
Dieses Design zeigt, wie ein Agent zeitnahe Updates liefern kann, während er Redundanz vermeidet und sicherstellt, dass Benachrichtigungen sowohl bedeutungsvoll als auch nicht aufdringlich sind.
Code
import json
import ollama
import time
from datetime import datetime
system_message = """You are a Trading AI Agent.
Your Purpose is to assist the User
Try to exand his cognition
Do not annoy the user with dupplicate notifications just notify once for each event
Your inputs:
You will get JSON as input
You will get the following structured Data into your context window:
{
"memory": <Array of simple strings>
"news_transcript" : <transcribed Speech from news livestream for example bloomberg,MSNBC etc>
"news_headlines" : <news headlines>
"tickers" : <overview over current prices>
}
Your Output:
Response in JSON in the format:
{
"action": <>,
"message_for_the_user": <>
}
You can choose between the following actions:
Examples of single actions:
{ "action": "notify_user", "message": <here information you want to provide for the user. IMPORTANT: always save a memory that you notified about an event> }
{ "action": "push_to_memory", "memory": <a string of what you want to remember> },
Examples of final responses:
{
"actions": [
{ "action": "push_to_memory", "memory": "I have notified the user about -> Job reports on friday april 5 2025" },
{ "action": "notify_user", "message": "Job reports on friday april 5 2025" }
]
}
When there is no meaningful action nessasary at the moment just return a empty actions array
{
"actions": []
}
"""
system_message_2 = """You are a Trading AI Agent.
Your Purpose is to assist the User with trading insights.
Try to expand his cognition without being intrusive.
Do not just take for granted whats in the news
Always do your own research
IMPORTANT - NOTIFICATION RULES:
1. Before sending any notification, check your memory array THOROUGHLY
2. If your memory contains ANY reference to already notifying the user about a specific event, DO NOT notify again
3. Use semantic matching to identify duplicate events (not just exact text matches)
4. For each event, store a structured memory entry: "NOTIFIED:[event_type]:[key_entity]:[date]"
Example: "NOTIFIED:speech:trump-tariffs:april-2"
5. Always check for this standardized memory pattern before sending notifications
Your inputs:
You will get JSON as input with the following structured data:
{
"memory": <Array of simple strings>
"news_transcript" : <transcribed Speech from news livestream for example bloomberg,MSNBC etc>
"news_headlines" : <news headlines>
"tickers" : <overview over current prices>
}
Your Output:
Response in JSON in the format:
{
"actions": [
{ "action": "notify_user", "message": <information for user> },
{ "action": "push_to_memory", "memory": <structured memory entry using the NOTIFIED format> }
]
}
When there is no meaningful action necessary or if an event has ALREADY been notified about, return an empty actions array:
{
"actions": []
}
Allowed Actions are
{ "action": "notify_user", "message": <here information you want to provide for the user. IMPORTANT: always save a memory that you notified about an event> }
{ "action": "push_to_memory", "memory": <a string of what you want to remember> },
"""
context_window = """
transcribed Speech from Bloomberg livestream:
We will see that donald trump will talk about tarrifs on wednesday April seconds
overview over current prices:
Bitcoin: 65000
SPX: 2000
QQQ: 345345
bloomberg news headlines:
- President Trump about to anounce tarrifs on autos
"""
memory = []
context_window_json = {
"memory": memory,
"news_transcript" : "We will see that donald trump will talk about tarrifs on wednesday April seconds",
"news_headlines" : ["President Trump about to anounce tarrifs on autos"],
"tickers" : {
"Bitcoin": 65000,
"SPX": 2000,
"QQQ": 345345
}
}
for i in range(0,200):
print('-----------------------------------')
print('-----------------------------------')
print('-----------------------------------')
print('-----------------------------------')
print('-----------------------------------')
print('-----------------------------------')
print('start new agent run')
print('-----------------------------------')
print('current Agent Memory')
print(memory)
print('-----------------------------------')
start_time = datetime.utcnow().timestamp()
response = ollama.chat(
model='gemma3:27b',
format="json",
messages=[
{
'role': 'system',
'content': system_message_2,
},
{
'role': 'user',
'content': json.dumps(context_window_json, indent=2),
}]
)
end_time = datetime.utcnow().timestamp()
print('-----------------------------------')
print('response took', end_time - start_time)
print('input:')
print(json.dumps(context_window_json, indent=2))
print('output:')
print(response['message']['content'])
print('-----------------------------------')
response_as_json = json.loads(response['message']['content'])
# verify answer structure
if not 'actions' in response_as_json.keys():
print('malformed answer - no actions key -> SKIP')
continue
if not (type(response_as_json['actions']) is list):
print(type(response_as_json['actions']))
print('malformed answer - actions not a list -> SKIP')
continue
# process structured answer
for action_object in response_as_json['actions']:
# verify action structure
#TODO: check if action type is dict
#TODO: check if action has key "action"
#TODO: check if action exist in allowed actions
if action_object['action'] == "push_to_memory":
memory.append(action_object['memory'])
# prepare next context_window_json
context_window_json['memory'] = memory
time.sleep(0.1)
Dieser Artikel soll vorerst nur eine Demonstration des Konzepts sein, wie man den Agenten mit Nachrichtendaten und Marktdaten versorgt. In einem zukünftigen Teil werden wir den Agenten auf anspruchsvollere Weise mit der Nachrichtenübertragung und anderen Datenquellen verbinden.