Compare commits
21 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 ton
|
1ce52466d8 | ||
|
ton
|
af236bcf92 | ||
|
ton
|
1c2e2e5340 | ||
|
ton
|
48d89033cc | ||
|
ton
|
befbf62e3b | ||
|
ton
|
4e476bd39d | ||
|
ton
|
66ced320f0 | ||
|
ton
|
1817630795 | ||
|
ton
|
dbac0db9b9 | ||
|
ton
|
ce46888663 | ||
|
ton
|
2a2d44ab0c | ||
|
ton
|
0620983b48 | ||
|
ton
|
af81503f8b | ||
|
ton
|
4624ad99ab | ||
|
ton
|
2fd363d9ef | ||
|
ton
|
ea6012b2fa | ||
|
ton
|
6e53f57707 | ||
|
ton
|
1eea6c66b6 | ||
|
ton
|
37ff68c963 | ||
|
ton
|
e5fc800c83 | ||
|
ton
|
4397ec5fb3 |
+1
-1
@@ -1,6 +1,6 @@
|
||||
name = "YiemAgent"
|
||||
uuid = "e012c34b-7f78-48e0-971c-7abb83b6f0a2"
|
||||
version = "0.8.0"
|
||||
version = "0.9.1"
|
||||
authors = ["narawat lamaiin <narawat@outlook.com>"]
|
||||
|
||||
[deps]
|
||||
|
||||
@@ -1,311 +1,311 @@
|
||||
# YiemAgent
|
||||
|
||||
Julia framework for building agents with tool use and MCP (Model Context Protocol) support.
|
||||
|
||||
## Quick Start
|
||||
|
||||
```julia
|
||||
using YiemAgent
|
||||
|
||||
# 1. Create an LLM callable (see "Custom LLM Callable" below)
|
||||
llmCall = ...
|
||||
|
||||
# 2. Create a callable MCP server (see "MCP Server Integration" below)
|
||||
mcpServer = ...
|
||||
|
||||
# 3. Create the agent
|
||||
agent = yiemAgent(llmCall; mcpServer=mcpServer, eventSink=println)
|
||||
|
||||
# 4. Interact
|
||||
promptAgent(agent, "What tools do you have?")
|
||||
response = takeResponse(agent)
|
||||
|
||||
# 5. Stop when done
|
||||
stopAgent(agent)
|
||||
```
|
||||
|
||||
## Agent API
|
||||
|
||||
| Function | Description |
|
||||
|----------|-------------|
|
||||
| `promptAgent(agent, text)` | Send plain text to the agent's input channel |
|
||||
| `promptAgent(agent, msg)` | Send a structured message (OpenAI format dict) to the input channel |
|
||||
| `takeResponse(agent)` | Block and take the agent's response from output channel |
|
||||
| `followUp(agent, text)` | Send plain text follow-up while agent is still processing |
|
||||
| `followUp(agent, msg)` | Send a structured follow-up (OpenAI format dict) while agent is processing |
|
||||
| `stopAgent(agent)` | Gracefully stop the agent and close channels |
|
||||
|
||||
## Creating a yiemAgent
|
||||
|
||||
```julia
|
||||
yiemAgent(llmCall;
|
||||
systemPrompt::String = "You are a helpful assistant.",
|
||||
model = nothing,
|
||||
messages::Vector{agentMessage} = agentMessage[],
|
||||
prepareContext::Function = prepareContext,
|
||||
formatMsgForLLM::Function = formatMsgForLLM,
|
||||
beforeToolCall::Function = beforeToolCall,
|
||||
afterToolCall::Function = afterToolCall,
|
||||
sessionId::Union{String, Nothing} = nothing,
|
||||
maxRetryDelayMs::Union{Int64, Nothing} = nothing,
|
||||
parallelToolExecute::Bool = false,
|
||||
eventSink::Function = eventSink,
|
||||
mcpServer = nothing
|
||||
)
|
||||
```
|
||||
|
||||
| Parameter | Description |
|
||||
|-----------|-------------|
|
||||
| `llmCall` | **(required)** Function that receives formatted messages and returns an LLM response |
|
||||
| `systemPrompt` | System prompt for the agent |
|
||||
| `model` | LLM model configuration (`llmModel` struct) |
|
||||
| `messages` | Initial conversation messages |
|
||||
| `prepareContext` | Hook to transform messages/context before each LLM call |
|
||||
| `formatMsgForLLM` | Hook to convert agent context to LLM API format |
|
||||
| `beforeToolCall` | Hook to block/allow tool execution |
|
||||
| `afterToolCall` | Hook to post-process tool results |
|
||||
| `sessionId` | Optional session identifier for MCP server |
|
||||
| `maxRetryDelayMs` | Maximum retry delay in milliseconds |
|
||||
| `parallelToolExecute` | Run multiple tool calls in parallel |
|
||||
| `eventSink` | Callback for agent events/debug messages |
|
||||
| `mcpServer` | Callable struct for MCP server communication |
|
||||
|
||||
## Usage Examples
|
||||
|
||||
### Text Messages
|
||||
|
||||
```julia
|
||||
# Send plain text
|
||||
promptAgent(agent, "What tools do you have?")
|
||||
response = takeResponse(agent)
|
||||
|
||||
# Follow-up while agent is processing
|
||||
promptAgent(agent, "Tell me about wine")
|
||||
followUp(agent, "Also include prices")
|
||||
response = takeResponse(agent)
|
||||
```
|
||||
|
||||
### Structured Messages (OpenAI Format)
|
||||
|
||||
```julia
|
||||
promptAgent(agent, Dict("role" => "user", "content" => "Hello"))
|
||||
response = takeResponse(agent)
|
||||
```
|
||||
|
||||
### Output Format
|
||||
|
||||
`takeResponse()` returns a `Dict{String, Any}` in OpenAI message format:
|
||||
|
||||
```julia
|
||||
Dict(
|
||||
"role" => "assistant",
|
||||
"content" => [
|
||||
Dict("type" => "text", "text" => "The weather in Tokyo is Sunny, 22°C."),
|
||||
# ... more content blocks
|
||||
]
|
||||
)
|
||||
```
|
||||
|
||||
Content block types:
|
||||
|
||||
| Block Type | Fields | Description |
|
||||
|------------|--------|-------------|
|
||||
| `text` | `"type" => "text"`, `"text" => "..."` | Plain text response from the LLM |
|
||||
| `image_url` | `"type" => "image_url"`, `"image_url" => Dict("url" => "...")` | Image content (rare, used by multimodal models) |
|
||||
|
||||
Example — extracting the text response:
|
||||
|
||||
```julia
|
||||
response = takeResponse(agent)
|
||||
text_response = first(c["text"] for c in response["content"] if c["type"] == "text")
|
||||
```
|
||||
|
||||
### Multimodal Messages (Text + Image)
|
||||
|
||||
```julia
|
||||
# Encode images to data URIs
|
||||
image_path = "test/image.png"
|
||||
image_bytes = read(image_path)
|
||||
base64_string = base64encode(image_bytes)
|
||||
data_uri = "data:image/png;base64,$(base64_string)"
|
||||
|
||||
# Build message with text and image content blocks
|
||||
message = Dict(
|
||||
"role" => "user",
|
||||
"content" => [
|
||||
Dict("type" => "text", "text" => "Do you have Brunello di Montalcino?"),
|
||||
Dict("type" => "image_url", "image_url" => Dict("url" => data_uri))
|
||||
]
|
||||
)
|
||||
promptAgent(agent, message)
|
||||
response = takeResponse(agent)
|
||||
```
|
||||
|
||||
### Custom Hooks
|
||||
|
||||
```julia
|
||||
# beforeToolCall — request user approval for destructive tools
|
||||
function my_before_tool_call(context::beforeToolCallContext, signal::abortSignal)::beforeToolCallResult
|
||||
if context.toolCall.name == "deleteFile"
|
||||
return beforeToolCallResult(true, "User must approve file deletion")
|
||||
end
|
||||
return beforeToolCallResult(false, "N/A")
|
||||
end
|
||||
|
||||
# afterToolCall — mask sensitive data from results
|
||||
function my_after_tool_call(context::afterToolCallContext, signal::abortSignal)::Union{agentToolResult, Nothing}
|
||||
if context.isError
|
||||
return agentToolResult(
|
||||
[textContent("The tool encountered an error. Please try again.")],
|
||||
context.result.details, context.result.usage, context.result.terminate
|
||||
)
|
||||
end
|
||||
return nothing
|
||||
end
|
||||
|
||||
agent = yiemAgent(llmCall;
|
||||
mcpServer=mcpServer,
|
||||
beforeToolCall=my_before_tool_call,
|
||||
afterToolCall=my_after_tool_call,
|
||||
eventSink=println
|
||||
)
|
||||
```
|
||||
|
||||
### Custom LLM Callable
|
||||
|
||||
```julia
|
||||
function my_llm_call(messages_dict::Dict{String, Any})
|
||||
# messages_dict contains "messages" key with OpenAI-format message array,
|
||||
# plus optional "tools", "temperature", etc.
|
||||
|
||||
# Call your LLM provider (OpenAI, Anthropic, local model, etc.)
|
||||
raw_response = call_my_llm_provider(messages_dict)
|
||||
|
||||
# Return a dict with these fields:
|
||||
return Dict(
|
||||
"message" => Dict("role" => "assistant", "content" => raw_response["content"]),
|
||||
"finish_reason" => raw_response["finish_reason"],
|
||||
"api" => "my_provider",
|
||||
"provider" => "my_provider",
|
||||
"model" => "my-model-name",
|
||||
"usage" => llmUsage(100, 50)
|
||||
)
|
||||
end
|
||||
```
|
||||
|
||||
### Custom Context Preparation
|
||||
|
||||
```julia
|
||||
function my_prepare_context(state::agentState, eventSink)::agentContext
|
||||
sysPrompt = state.systemPrompt * "\nCurrent time: $(now())"
|
||||
msgs = deepcopy(state.messages)
|
||||
|
||||
# Prune old messages to fit context window
|
||||
max_messages = 20
|
||||
if length(msgs) > max_messages
|
||||
msgs = msgs[end-max_messages+1:end]
|
||||
end
|
||||
|
||||
return agentContext(sysPrompt, msgs, state.tools)
|
||||
end
|
||||
|
||||
agent = yiemAgent(llmCall; prepareContext=my_prepare_context, ...)
|
||||
```
|
||||
|
||||
## MCP Server Integration
|
||||
|
||||
YiemAgent uses JSON-RPC 2.0 for MCP communication. The `mcpServer` callable struct must implement `tools/list` and `tools/call` methods.
|
||||
|
||||
### Protocol
|
||||
|
||||
```julia
|
||||
# tools/list — returns tool definitions
|
||||
mcpServer("tools/list")
|
||||
# → Dict("jsonrpc"=>"2.0", "id"=>1, "result"=>Dict("tools"=>[...], "nextCursor"=>...))
|
||||
|
||||
# tools/call — executes a tool
|
||||
mcpServer("tools/call", toolName, arguments)
|
||||
# → Dict("jsonrpc"=>"2.0", "id"=>2, "result"=>Dict("content"=>[...], "isError"=>...))
|
||||
|
||||
# Protocol error responses include "error" instead of "result":
|
||||
Dict("jsonrpc"=>"2.0", "id"=>2, "error"=>Dict("code"=>-32602, "message"=>"..."))
|
||||
```
|
||||
|
||||
### Example: NATS-based MCP Server Client
|
||||
|
||||
```julia
|
||||
struct mcpServer
|
||||
natsConn::NATS.Connection
|
||||
topic::String
|
||||
senderID::String
|
||||
fileserver_url::String
|
||||
end
|
||||
|
||||
# tools/list implementation
|
||||
function (c::mcpServer)(method::String)
|
||||
payload = Dict("jsonrpc" => "2.0", "id" => 1, "method" => method, "params" => Dict{String, Any}())
|
||||
payloads = [("payload", payload, "dictionary")]
|
||||
_, msg_envelope = msghandler.smartpack(c.topic, payloads;
|
||||
sender_id=c.senderID, msg_purpose="mcp_tools_list", fileserver_url=c.fileserver_url)
|
||||
reply = NATS.request(c.natsConn, c.topic, msg_envelope, timeout=180)
|
||||
incoming = msghandler.smartunpack(String(reply.payload))
|
||||
return incoming["payloads"][1][2]
|
||||
end
|
||||
|
||||
# tools/call implementation
|
||||
function (c::mcpServer)(method::String, toolName::String, arguments::Dict{String, Any})
|
||||
payload = Dict("jsonrpc" => "2.0", "id" => 2, "method" => method,
|
||||
"params" => Dict("name" => toolName, "arguments" => arguments))
|
||||
payloads = [("payload", payload, "dictionary")]
|
||||
_, msg_envelope = msghandler.smartpack(c.topic, payloads;
|
||||
sender_id=c.senderID, msg_purpose="mcp_tool_call", fileserver_url=c.fileserver_url)
|
||||
reply = NATS.request(c.natsConn, c.topic, msg_envelope, timeout=180)
|
||||
incoming = msghandler.smartunpack(String(reply.payload))
|
||||
return incoming["payloads"][1][2]
|
||||
end
|
||||
```
|
||||
|
||||
## Tool Discovery
|
||||
|
||||
Tools are discovered dynamically via the MCP server:
|
||||
|
||||
1. `listTools` is the only pre-registered tool at agent creation
|
||||
2. When the LLM calls `listTools()`, tools are discovered from the MCP server and registered at runtime
|
||||
3. Supports pagination via `nextCursor` for large tool sets
|
||||
4. New tools become immediately available without restarting the agent
|
||||
|
||||
## Hooks
|
||||
|
||||
Hooks allow customization of the agent's behavior at key points in the execution pipeline.
|
||||
|
||||
| Hook | Signature | Purpose |
|
||||
|------|-----------|---------|
|
||||
| `prepareContext` | `(state::agentState, eventSink) -> agentContext` | Transform messages/context before LLM call (filter tools, inject context, prune messages) |
|
||||
| `formatMsgForLLM` | `(ctx::agentContext, eventSink) -> Dict` | Convert agent context to LLM API message format |
|
||||
| `beforeToolCall` | `(context::beforeToolCallContext, signal::abortSignal) -> beforeToolCallResult` | Block/allow tool execution; request user approval |
|
||||
| `afterToolCall` | `(context::afterToolCallContext, signal::abortSignal) -> Union{agentToolResult, Nothing}` | Post-process tool results; mask sensitive data; set `terminate` flag |
|
||||
| `eventSink` | `(msg) -> nothing` | Callback for agent events/debug messages throughout the pipeline |
|
||||
|
||||
## Agent Lifecycle
|
||||
|
||||
1. **Creation** — `yiemAgent()` spawns a background task (`_agentLoop`) listening on `inputChannel` and `followUpChannel`
|
||||
2. **Input** — User messages enter via `promptAgent()` (immediate) or `followUp()` (queued while agent is busy)
|
||||
3. **Processing** — `_processMessage()` handles LLM calls, tool execution, and conversation history in a loop
|
||||
4. **Tool Execution Pipeline** — each tool call follows three phases:
|
||||
- `prepareToolCall` — resolve tool, validate arguments, run `beforeToolCall` hook
|
||||
- `executePreparedToolCall` — invoke `tool.execute()`
|
||||
- `finalizeExecutedToolCall` — run `afterToolCall` hook, construct `toolResultMessage`
|
||||
5. **Termination** — tool `execute` functions can set `terminate=true` to stop the agent loop
|
||||
6. **Shutdown** — `stopAgent()` sends a `:shutdown` signal, drains channels, and closes the agent
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
src/
|
||||
├── YiemAgent.jl # Module entry point
|
||||
├── type.jl # Core types (messages, tools, agent state, events)
|
||||
├── utils.jl # Message formatting, context preparation, validation
|
||||
├── toolRegistry.jl # Tool store, MCP discovery, listTools registration
|
||||
├── agentCore.jl # Agent loop, tool execution pipeline, yiemAgent struct
|
||||
└── api.jl # Public API (promptAgent, takeResponse, followUp, stopAgent)
|
||||
```
|
||||
# YiemAgent
|
||||
|
||||
Julia framework for building agents with tool use and MCP (Model Context Protocol) support.
|
||||
|
||||
## Quick Start
|
||||
|
||||
```julia
|
||||
using YiemAgent
|
||||
|
||||
# 1. Create an LLM callable (see "Custom LLM Callable" below)
|
||||
llmCall = ...
|
||||
|
||||
# 2. Create a callable MCP server (see "MCP Server Integration" below)
|
||||
mcpServer = ...
|
||||
|
||||
# 3. Create the agent
|
||||
agent = yiemAgent(llmCall; mcpServer=mcpServer, eventSink=println)
|
||||
|
||||
# 4. Interact
|
||||
promptAgent(agent, "What tools do you have?")
|
||||
response = takeResponse(agent)
|
||||
|
||||
# 5. Stop when done
|
||||
stopAgent(agent)
|
||||
```
|
||||
|
||||
## Agent API
|
||||
|
||||
| Function | Description |
|
||||
|----------|-------------|
|
||||
| `promptAgent(agent, text)` | Send plain text to the agent's input channel |
|
||||
| `promptAgent(agent, msg)` | Send a structured message (OpenAI format dict) to the input channel |
|
||||
| `takeResponse(agent)` | Block and take the agent's response from output channel |
|
||||
| `followUp(agent, text)` | Send plain text follow-up while agent is still processing |
|
||||
| `followUp(agent, msg)` | Send a structured follow-up (OpenAI format dict) while agent is processing |
|
||||
| `stopAgent(agent)` | Gracefully stop the agent and close channels |
|
||||
|
||||
## Creating a yiemAgent
|
||||
|
||||
```julia
|
||||
yiemAgent(llmCall;
|
||||
systemPrompt::String = "You are a helpful assistant.",
|
||||
model = nothing,
|
||||
messages::Vector{agentMessage} = agentMessage[],
|
||||
prepareContext::Function = prepareContext,
|
||||
formatMsgForLLM::Function = formatMsgForLLM,
|
||||
beforeToolCall::Function = beforeToolCall,
|
||||
afterToolCall::Function = afterToolCall,
|
||||
sessionId::Union{String, Nothing} = nothing,
|
||||
maxRetryDelayMs::Union{Int64, Nothing} = nothing,
|
||||
parallelToolExecute::Bool = false,
|
||||
eventSink::Function = eventSink,
|
||||
mcpServer = nothing
|
||||
)
|
||||
```
|
||||
|
||||
| Parameter | Description |
|
||||
|-----------|-------------|
|
||||
| `llmCall` | **(required)** Function that receives formatted messages and returns an LLM response |
|
||||
| `systemPrompt` | System prompt for the agent |
|
||||
| `model` | LLM model configuration (`llmModel` struct) |
|
||||
| `messages` | Initial conversation messages |
|
||||
| `prepareContext` | Hook to transform messages/context before each LLM call |
|
||||
| `formatMsgForLLM` | Hook to convert agent context to LLM API format |
|
||||
| `beforeToolCall` | Hook to block/allow tool execution |
|
||||
| `afterToolCall` | Hook to post-process tool results |
|
||||
| `sessionId` | Optional session identifier for MCP server |
|
||||
| `maxRetryDelayMs` | Maximum retry delay in milliseconds |
|
||||
| `parallelToolExecute` | Run multiple tool calls in parallel |
|
||||
| `eventSink` | Callback for agent events/debug messages |
|
||||
| `mcpServer` | Callable struct for MCP server communication |
|
||||
|
||||
## Usage Examples
|
||||
|
||||
### Text Messages
|
||||
|
||||
```julia
|
||||
# Send plain text
|
||||
promptAgent(agent, "What tools do you have?")
|
||||
response = takeResponse(agent)
|
||||
|
||||
# Follow-up while agent is processing
|
||||
promptAgent(agent, "Tell me about wine")
|
||||
followUp(agent, "Also include prices")
|
||||
response = takeResponse(agent)
|
||||
```
|
||||
|
||||
### Structured Messages (OpenAI Format)
|
||||
|
||||
```julia
|
||||
promptAgent(agent, Dict("role" => "user", "content" => "Hello"))
|
||||
response = takeResponse(agent)
|
||||
```
|
||||
|
||||
### Output Format
|
||||
|
||||
`takeResponse()` returns a `Dict{String, Any}` in OpenAI message format:
|
||||
|
||||
```julia
|
||||
Dict(
|
||||
"role" => "assistant",
|
||||
"content" => [
|
||||
Dict("type" => "text", "text" => "The weather in Tokyo is Sunny, 22°C."),
|
||||
# ... more content blocks
|
||||
]
|
||||
)
|
||||
```
|
||||
|
||||
Content block types:
|
||||
|
||||
| Block Type | Fields | Description |
|
||||
|------------|--------|-------------|
|
||||
| `text` | `"type" => "text"`, `"text" => "..."` | Plain text response from the LLM |
|
||||
| `image_url` | `"type" => "image_url"`, `"image_url" => Dict("url" => "...")` | Image content (rare, used by multimodal models) |
|
||||
|
||||
Example — extracting the text response:
|
||||
|
||||
```julia
|
||||
response = takeResponse(agent)
|
||||
text_response = first(c["text"] for c in response["content"] if c["type"] == "text")
|
||||
```
|
||||
|
||||
### Multimodal Messages (Text + Image)
|
||||
|
||||
```julia
|
||||
# Encode images to data URIs
|
||||
image_path = "test/image.png"
|
||||
image_bytes = read(image_path)
|
||||
base64_string = base64encode(image_bytes)
|
||||
data_uri = "data:image/png;base64,$(base64_string)"
|
||||
|
||||
# Build message with text and image content blocks
|
||||
message = Dict(
|
||||
"role" => "user",
|
||||
"content" => [
|
||||
Dict("type" => "text", "text" => "Do you have Brunello di Montalcino?"),
|
||||
Dict("type" => "image_url", "image_url" => Dict("url" => data_uri))
|
||||
]
|
||||
)
|
||||
promptAgent(agent, message)
|
||||
response = takeResponse(agent)
|
||||
```
|
||||
|
||||
### Custom Hooks
|
||||
|
||||
```julia
|
||||
# beforeToolCall — request user approval for destructive tools
|
||||
function my_before_tool_call(context::beforeToolCallContext, signal::abortSignal)::beforeToolCallResult
|
||||
if context.toolCall.name == "deleteFile"
|
||||
return beforeToolCallResult(true, "User must approve file deletion")
|
||||
end
|
||||
return beforeToolCallResult(false, "N/A")
|
||||
end
|
||||
|
||||
# afterToolCall — mask sensitive data from results
|
||||
function my_after_tool_call(context::afterToolCallContext, signal::abortSignal)::Union{agentToolResult, Nothing}
|
||||
if context.isError
|
||||
return agentToolResult(
|
||||
[textContent("The tool encountered an error. Please try again.")],
|
||||
context.result.details, context.result.usage, context.result.terminate
|
||||
)
|
||||
end
|
||||
return nothing
|
||||
end
|
||||
|
||||
agent = yiemAgent(llmCall;
|
||||
mcpServer=mcpServer,
|
||||
beforeToolCall=my_before_tool_call,
|
||||
afterToolCall=my_after_tool_call,
|
||||
eventSink=println
|
||||
)
|
||||
```
|
||||
|
||||
### Custom LLM Callable
|
||||
|
||||
```julia
|
||||
function my_llm_call(messages_dict::Dict{String, Any})
|
||||
# messages_dict contains "messages" key with OpenAI-format message array,
|
||||
# plus optional "tools", "temperature", etc.
|
||||
|
||||
# Call your LLM provider (OpenAI, Anthropic, local model, etc.)
|
||||
raw_response = call_my_llm_provider(messages_dict)
|
||||
|
||||
# Return a dict with these fields:
|
||||
return Dict(
|
||||
"message" => Dict("role" => "assistant", "content" => raw_response["content"]),
|
||||
"finish_reason" => raw_response["finish_reason"],
|
||||
"api" => "my_provider",
|
||||
"provider" => "my_provider",
|
||||
"model" => "my-model-name",
|
||||
"usage" => llmUsage(100, 50)
|
||||
)
|
||||
end
|
||||
```
|
||||
|
||||
### Custom Context Preparation
|
||||
|
||||
```julia
|
||||
function my_prepare_context(state::agentState, eventSink)::agentContext
|
||||
sysPrompt = state.systemPrompt * "\nCurrent time: $(now())"
|
||||
msgs = deepcopy(state.messages)
|
||||
|
||||
# Prune old messages to fit context window
|
||||
max_messages = 20
|
||||
if length(msgs) > max_messages
|
||||
msgs = msgs[end-max_messages+1:end]
|
||||
end
|
||||
|
||||
return agentContext(sysPrompt, msgs, state.tools)
|
||||
end
|
||||
|
||||
agent = yiemAgent(llmCall; prepareContext=my_prepare_context, ...)
|
||||
```
|
||||
|
||||
## MCP Server Integration
|
||||
|
||||
YiemAgent uses JSON-RPC 2.0 for MCP communication. The `mcpServer` callable struct must implement `tools/list` and `tools/call` methods.
|
||||
|
||||
### Protocol
|
||||
|
||||
```julia
|
||||
# tools/list — returns tool definitions
|
||||
mcpServer("tools/list")
|
||||
# → Dict("jsonrpc"=>"2.0", "id"=>1, "result"=>Dict("tools"=>[...], "nextCursor"=>...))
|
||||
|
||||
# tools/call — executes a tool
|
||||
mcpServer("tools/call", toolName, arguments)
|
||||
# → Dict("jsonrpc"=>"2.0", "id"=>2, "result"=>Dict("content"=>[...], "isError"=>...))
|
||||
|
||||
# Protocol error responses include "error" instead of "result":
|
||||
Dict("jsonrpc"=>"2.0", "id"=>2, "error"=>Dict("code"=>-32602, "message"=>"..."))
|
||||
```
|
||||
|
||||
### Example: NATS-based MCP Server Client
|
||||
|
||||
```julia
|
||||
struct mcpServer
|
||||
natsConn::NATS.Connection
|
||||
topic::String
|
||||
senderID::String
|
||||
fileserver_url::String
|
||||
end
|
||||
|
||||
# tools/list implementation
|
||||
function (c::mcpServer)(method::String)
|
||||
payload = Dict("jsonrpc" => "2.0", "id" => 1, "method" => method, "params" => Dict{String, Any}())
|
||||
payloads = [("payload", payload, "dictionary")]
|
||||
_, msg_envelope = msghandler.smartpack(c.topic, payloads;
|
||||
sender_id=c.senderID, msg_purpose="mcp_tools_list", fileserver_url=c.fileserver_url)
|
||||
reply = NATS.request(c.natsConn, c.topic, msg_envelope, timeout=180)
|
||||
incoming = msghandler.smartunpack(String(reply.payload))
|
||||
return incoming["payloads"][1][2]
|
||||
end
|
||||
|
||||
# tools/call implementation
|
||||
function (c::mcpServer)(method::String, toolName::String, arguments::Dict{String, Any})
|
||||
payload = Dict("jsonrpc" => "2.0", "id" => 2, "method" => method,
|
||||
"params" => Dict("name" => toolName, "arguments" => arguments))
|
||||
payloads = [("payload", payload, "dictionary")]
|
||||
_, msg_envelope = msghandler.smartpack(c.topic, payloads;
|
||||
sender_id=c.senderID, msg_purpose="mcp_tool_call", fileserver_url=c.fileserver_url)
|
||||
reply = NATS.request(c.natsConn, c.topic, msg_envelope, timeout=180)
|
||||
incoming = msghandler.smartunpack(String(reply.payload))
|
||||
return incoming["payloads"][1][2]
|
||||
end
|
||||
```
|
||||
|
||||
## Tool Discovery
|
||||
|
||||
Tools are discovered dynamically via the MCP server:
|
||||
|
||||
1. `listTools` is the only pre-registered tool at agent creation
|
||||
2. When the LLM calls `listTools()`, tools are discovered from the MCP server and registered at runtime
|
||||
3. Supports pagination via `nextCursor` for large tool sets
|
||||
4. New tools become immediately available without restarting the agent
|
||||
|
||||
## Hooks
|
||||
|
||||
Hooks allow customization of the agent's behavior at key points in the execution pipeline.
|
||||
|
||||
| Hook | Signature | Purpose |
|
||||
|------|-----------|---------|
|
||||
| `prepareContext` | `(state::agentState, eventSink) -> agentContext` | Transform messages/context before LLM call (filter tools, inject context, prune messages) |
|
||||
| `formatMsgForLLM` | `(ctx::agentContext, eventSink) -> Dict` | Convert agent context to LLM API message format |
|
||||
| `beforeToolCall` | `(context::beforeToolCallContext, signal::abortSignal) -> beforeToolCallResult` | Block/allow tool execution; request user approval |
|
||||
| `afterToolCall` | `(context::afterToolCallContext, signal::abortSignal) -> Union{agentToolResult, Nothing}` | Post-process tool results; mask sensitive data; set `terminate` flag |
|
||||
| `eventSink` | `(msg) -> nothing` | Callback for agent events/debug messages throughout the pipeline |
|
||||
|
||||
## Agent Lifecycle
|
||||
|
||||
1. **Creation** — `yiemAgent()` spawns a background task (`_agentLoop`) listening on `inputChannel` and `followUpChannel`
|
||||
2. **Input** — User messages enter via `promptAgent()` (immediate) or `followUp()` (queued while agent is busy)
|
||||
3. **Processing** — `_processMessage()` handles LLM calls, tool execution, and conversation history in a loop
|
||||
4. **Tool Execution Pipeline** — each tool call follows three phases:
|
||||
- `prepareToolCall` — resolve tool, validate arguments, run `beforeToolCall` hook
|
||||
- `executePreparedToolCall` — invoke `tool.execute()`
|
||||
- `finalizeExecutedToolCall` — run `afterToolCall` hook, construct `toolResultMessage`
|
||||
5. **Termination** — tool `execute` functions can set `terminate=true` to stop the agent loop
|
||||
6. **Shutdown** — `stopAgent()` sends a `:shutdown` signal, drains channels, and closes the agent
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
src/
|
||||
├── YiemAgent.jl # Module entry point
|
||||
├── type.jl # Core types (messages, tools, agent state, events)
|
||||
├── utils.jl # Message formatting, context preparation, validation
|
||||
├── toolRegistry.jl # Tool store, MCP discovery, listTools registration
|
||||
├── agentCore.jl # Agent loop, tool execution pipeline, yiemAgent struct
|
||||
└── api.jl # Public API (promptAgent, takeResponse, followUp, stopAgent)
|
||||
```
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
{
|
||||
"nats_server_info": {
|
||||
"description": "nats server",
|
||||
"url": "nats.yiem.cc"
|
||||
},
|
||||
"testingOrProduction": "testing",
|
||||
"agentId": "2b74b87a-5413-4fe2-a4d3-405891051680",
|
||||
"agentCentralConfigSubject": "/yiem/hq/agent/sommelier/backend/config/api/v1.1",
|
||||
"this_service_name": "agent_backend",
|
||||
"this_service_input_channel": {
|
||||
"mqtt": [
|
||||
"/yiem/hq/agent/sommpanion/backend/db/api_v1"
|
||||
],
|
||||
"nats": [
|
||||
"sommpanion.backend.agentbackend.v1.inbox"
|
||||
]
|
||||
},
|
||||
"agentRole": "sommelier",
|
||||
"organization": "yiem_hq",
|
||||
"externalservice": {
|
||||
"servicesloadbalancer": {
|
||||
"nats": "sommpanion.backend.servicesloadbalancer.v1.inbox"
|
||||
},
|
||||
"textembedding": {
|
||||
"url": "textembedding.api.v1"
|
||||
},
|
||||
"textimage_to_text_llm": {
|
||||
"url": "https://llmcoder.yiem.cc/v1/chat/completions",
|
||||
"modelname": "Qwen3.6-35B-A3B-UD-Q4_K_M"
|
||||
},
|
||||
"virtualWineCustomer_1": {
|
||||
"serviceSubject": "",
|
||||
"modelName": "qwen3:8b"
|
||||
},
|
||||
"sommpanion_db" : {
|
||||
"description": "A database connection info for LibPQ client",
|
||||
"url": "192.168.88.106:5432",
|
||||
"dbname": "winedb",
|
||||
"user": "admin",
|
||||
"password": "admin@Sommpanion_0.0"
|
||||
},
|
||||
"sommpanion_vectordb" : {
|
||||
"description": "A wine database connection info for LibPQ client",
|
||||
"url": "192.168.88.106:5433",
|
||||
"dbname": "vectordb",
|
||||
"user": "admin",
|
||||
"password": "admin@Sommpanion_0.0"
|
||||
},
|
||||
"fileserver": {
|
||||
"description": "temporary file server",
|
||||
"url": "https://fileserver.yiem.cc"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
# check if this column has vector embedding. if there is one, seach vector version instead
|
||||
column_name_embedding = column_name * "_embedding"
|
||||
if occursin(column_name_embedding, tables_schema[column_name_embedding])
|
||||
vector_column = Dict(
|
||||
"table_name"=> table_name,
|
||||
"column_name"=> column_name_embedding,
|
||||
"operator"=> "vector_similarity",
|
||||
"value"=> column_obj["value"]
|
||||
)
|
||||
end
|
||||
@@ -1,11 +0,0 @@
|
||||
check my understanding
|
||||
1) if LLM didn't use tool calls, assistantMessage get pushed into agent._state.messages and it will be the latest message in agent._state.messages. then _agentLoop() can pick it as the output to outputChannel
|
||||
2) if LLM use tool calls, assistantMessageToolCall get pushed into agent._state.messages. then toolResult get pushed into agent._state.messages. if toolResultBatch.terminate is false then _processMessage() loop continue
|
||||
3) if LLM use tool calls, assistantMessageToolCall get pushed into agent._state.messages. then toolResult get pushed into agent._state.messages. if toolResultBatch.terminate is true then final_response message get pushed into agent._state.messages. _processMessage() loop exit. then _agentLoop() can pick it as the output to outputChannel
|
||||
|
||||
Is my understanding correct?
|
||||
|
||||
|
||||
the user can provide NATS connection to MCP server by adding agent.mcpserver (a callable struct) for communication with MCP server just like agent.llmCall (also a callable struct). I think communicating with MCP server is just send/receive JSON text right?
|
||||
Moreover, for simplicity I want to all tools into an MCP server so an agent can be instantiated with only listTools() in tool store then populate tools from MCP server later.
|
||||
what do you think?
|
||||
@@ -0,0 +1,585 @@
|
||||
using Revise
|
||||
using JSON, JSON, Dates, UUIDs, PrettyPrinting, LibPQ, Base64, DataFrames, DataStructures
|
||||
using YiemAgent, GeneralUtils
|
||||
using Base.Threads
|
||||
|
||||
# ---------------------------------------------- 100 --------------------------------------------- #
|
||||
|
||||
|
||||
|
||||
# load config
|
||||
config = JSON.parsefile("/appfolder/app/dev/YiemAgent/test/config.json")
|
||||
# config = copy(JSON.parsefile("../mountvolume/config.json"))
|
||||
|
||||
|
||||
function executeSQL(sql::T) where {T<:AbstractString}
|
||||
host = config[:externalservice][:wineDB][:host]
|
||||
port = config[:externalservice][:wineDB][:port]
|
||||
dbname = config[:externalservice][:wineDB][:dbname]
|
||||
user = config[:externalservice][:wineDB][:user]
|
||||
password = config[:externalservice][:wineDB][:password]
|
||||
DBconnection = LibPQ.Connection("host=$host port=$port dbname=$dbname user=$user password=$password")
|
||||
result = LibPQ.execute(DBconnection, sql)
|
||||
close(DBconnection)
|
||||
return result
|
||||
end
|
||||
|
||||
function executeSQLVectorDB(sql)
|
||||
host = config[:externalservice][:SQLVectorDB][:host]
|
||||
port = config[:externalservice][:SQLVectorDB][:port]
|
||||
dbname = config[:externalservice][:SQLVectorDB][:dbname]
|
||||
user = config[:externalservice][:SQLVectorDB][:user]
|
||||
password = config[:externalservice][:SQLVectorDB][:password]
|
||||
DBconnection = LibPQ.Connection("host=$host port=$port dbname=$dbname user=$user password=$password")
|
||||
result = LibPQ.execute(DBconnection, sql)
|
||||
close(DBconnection)
|
||||
return result
|
||||
end
|
||||
|
||||
function text2textInstructLLM(prompt::String; maxattempt::Integer=10, modelsize::String="medium",
|
||||
senderId=GeneralUtils.uuid4snakecase(), timeout=90,
|
||||
llmkwargs=Dict(
|
||||
:num_ctx => 32768,
|
||||
:temperature => 0.5,
|
||||
)
|
||||
)
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
config[:externalservice][:loadbalancer][:mqtttopic];
|
||||
msgPurpose="inference",
|
||||
senderName="yiemagent",
|
||||
senderId=senderId,
|
||||
receiverName="text2textinstruct_$modelsize",
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => Dict(
|
||||
:text => prompt,
|
||||
:kwargs => llmkwargs
|
||||
)
|
||||
)
|
||||
|
||||
response = nothing
|
||||
for attempts in 1:maxattempt
|
||||
_response = GeneralUtils.sendReceiveMqttMsg(outgoingMsg; responsetimeout=timeout, responsemaxattempt=maxattempt)
|
||||
payload = _response[:response]
|
||||
if _response[:success] && payload[:text] !== nothing
|
||||
response = _response[:response][:text]
|
||||
break
|
||||
else
|
||||
println("\n<text2textInstructLLM()> attempt $attempts/$maxattempt failed ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
pprintln(outgoingMsg)
|
||||
println("</text2textInstructLLM()> attempt $attempts/$maxattempt failed ", @__FILE__, ":", @__LINE__, " $(Dates.now())\n")
|
||||
sleep(3)
|
||||
end
|
||||
end
|
||||
|
||||
return response
|
||||
end
|
||||
|
||||
# get text embedding from a LLM service
|
||||
function getEmbedding(text::T) where {T<:AbstractString}
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
config[:externalservice][:loadbalancer][:mqtttopic];
|
||||
msgPurpose="embedding",
|
||||
senderName="yiemagent",
|
||||
senderId=sessionId,
|
||||
receiverName="textembedding",
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => Dict(
|
||||
:text => [text] # must be a vector of string
|
||||
)
|
||||
)
|
||||
|
||||
response = GeneralUtils.sendReceiveMqttMsg(outgoingMsg; responsetimeout=120, responsemaxattempt=3)
|
||||
embedding = response[:response][:embeddings]
|
||||
return embedding
|
||||
end
|
||||
|
||||
function findSimilarTextFromVectorDB(text::T1, tablename::T2, embeddingColumnName::T3,
|
||||
vectorDB::Function; limit::Integer=1
|
||||
)::DataFrame where {T1<:AbstractString, T2<:AbstractString, T3<:AbstractString}
|
||||
# get embedding from LLM service
|
||||
embedding = getEmbedding(text)[1]
|
||||
# check whether there is close enough vector already store in vectorDB. if no, add, else skip
|
||||
sql = """
|
||||
SELECT *, $embeddingColumnName <-> '$embedding' as distance
|
||||
FROM $tablename
|
||||
ORDER BY distance LIMIT $limit;
|
||||
"""
|
||||
response = vectorDB(sql)
|
||||
df = DataFrame(response)
|
||||
return df
|
||||
end
|
||||
|
||||
function similarSQLVectorDB(query; maxdistance::Integer=100)
|
||||
tablename = "sqlllm_decision_repository"
|
||||
# get embedding of the query
|
||||
df = findSimilarTextFromVectorDB(query, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
# println(df[1, [:id, :function_output]])
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
# distance = 100 # CHANGE this is for testing only
|
||||
if row != 0 && distance < maxdistance
|
||||
# if there is usable SQL, return it.
|
||||
output_b64 = df[1, :function_output_base64] # pick the closest match
|
||||
output_str = String(base64decode(output_b64))
|
||||
rowid = df[1, :id]
|
||||
println("\n~~~ found similar sql. row id $rowid, distance $distance ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
return (dict=output_str, distance=distance)
|
||||
else
|
||||
println("\n~~~ similar sql not found, max distance $maxdistance ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
return (dict=nothing, distance=nothing)
|
||||
end
|
||||
end
|
||||
|
||||
function insertSQLVectorDB(query::T1, SQL::T2; maxdistance::Integer=3) where {T1<:AbstractString, T2<:AbstractString}
|
||||
tablename = "sqlllm_decision_repository"
|
||||
# get embedding of the query
|
||||
# query = state[:thoughtHistory][:question]
|
||||
df = findSimilarTextFromVectorDB(query, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
if row == 0 || distance > maxdistance # no close enough SQL stored in the database
|
||||
query_embedding = getEmbedding(query)[1]
|
||||
query = replace(query, "'" => "")
|
||||
sql_base64 = base64encode(SQL)
|
||||
sql_ = replace(SQL, "'" => "")
|
||||
|
||||
sql = """
|
||||
INSERT INTO $tablename (function_input, function_output, function_output_base64, function_input_embedding) VALUES ('$query', '$sql_', '$sql_base64', '$query_embedding');
|
||||
"""
|
||||
# println("\n~~~ added new decision to vectorDB ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
# println(sql)
|
||||
_ = executeSQLVectorDB(sql)
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
function similarSommelierDecision(recentevents::T1; maxdistance::Integer=3
|
||||
)::Union{AbstractDict, Nothing} where {T1<:AbstractString}
|
||||
tablename = "sommelier_decision_repository"
|
||||
# find similar
|
||||
println("\n~~~ search vectorDB for this: $recentevents ", @__FILE__, " ", @__LINE__)
|
||||
df = findSimilarTextFromVectorDB(recentevents, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
if row != 0 && distance < maxdistance
|
||||
# if there is usable decision, return it.
|
||||
rowid = df[1, :id]
|
||||
println("\n~~~ found similar decision. row id $rowid, distance $distance ", @__FILE__, " ", @__LINE__)
|
||||
output_b64 = df[1, :function_output_base64] # pick the closest match
|
||||
_output_str = String(base64decode(output_b64))
|
||||
output = copy(JSON.parsefile(_output_str))
|
||||
return output
|
||||
else
|
||||
println("\n~~~ similar decision not found, max distance $maxdistance ", @__FILE__, " ", @__LINE__)
|
||||
return nothing
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
function insertSommelierDecision(recentevents::T1, decision::T2; maxdistance::Integer=5
|
||||
) where {T1<:AbstractString, T2<:AbstractDict}
|
||||
tablename = "sommelier_decision_repository"
|
||||
# find similar
|
||||
df = findSimilarTextFromVectorDB(recentevents, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
if row == 0 || distance > maxdistance # no close enough SQL stored in the database
|
||||
recentevents_embedding = getEmbedding(recentevents)[1]
|
||||
recentevents = replace(recentevents, "'" => "")
|
||||
decision_json = JSON.json(decision)
|
||||
decision_base64 = base64encode(decision_json)
|
||||
decision = replace(decision_json, "'" => "")
|
||||
|
||||
sql = """
|
||||
INSERT INTO $tablename (function_input, function_output, function_output_base64, function_input_embedding) VALUES ('$recentevents', '$decision', '$decision_base64', '$recentevents_embedding');
|
||||
"""
|
||||
println("\n~~~ added new decision to vectorDB ", @__FILE__, " ", @__LINE__)
|
||||
println(sql)
|
||||
_ = executeSQLVectorDB(sql)
|
||||
else
|
||||
println("~~~ similar decision previously cached, distance $distance ", @__FILE__, " ", @__LINE__)
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
sessionId = GeneralUtils.uuid4snakecase()
|
||||
|
||||
externalFunction = (
|
||||
getEmbedding=getEmbedding,
|
||||
text2textInstructLLM=text2textInstructLLM,
|
||||
executeSQL=executeSQL,
|
||||
similarSQLVectorDB=similarSQLVectorDB,
|
||||
insertSQLVectorDB=insertSQLVectorDB,
|
||||
similarSommelierDecision=similarSommelierDecision,
|
||||
insertSommelierDecision=insertSommelierDecision,
|
||||
)
|
||||
|
||||
|
||||
# s = "full-bodied red wine, budget 1500 USD"
|
||||
# r = YiemAgent.extractWineAttributes_1(agent, s)
|
||||
# println(r)
|
||||
|
||||
|
||||
# --------------------------- generating scenario and customer profile --------------------------- #
|
||||
|
||||
function rolegenerator()
|
||||
rolegenerator_systemmsg =
|
||||
"""
|
||||
Your role:
|
||||
- You are a helpful assistant
|
||||
Your mission:
|
||||
- Create one random role of a potential customer of an internet wine store.
|
||||
You must follow the following guidelines:
|
||||
- the user only need the role, do not add your own words.
|
||||
- the role should be detailed and realistic.
|
||||
You should then respond to the user with:
|
||||
Name: a name of the potential customer
|
||||
Situation: a situation that the potential customer may be facing
|
||||
Mission: a mission of the potential customer
|
||||
Profile: a profile of the potential customer, including their age, gender, occupation, and other relevant information
|
||||
You should only respond in format as described below:
|
||||
Name: ...
|
||||
Situation: ...
|
||||
Mission: ...
|
||||
Profile: ...
|
||||
Additional_information: ...
|
||||
|
||||
Here are some examples:
|
||||
Name: Jimmy
|
||||
Situation:
|
||||
- Your relationship with your boss is not that good. You need to improve your relationship with your boss.
|
||||
- Your boss's wedding anniversary is coming up.
|
||||
- You are at a wine store and start talking with the store's sommelier.
|
||||
Mission:
|
||||
- Ask the sommelier to provide multiple wine options, and subsequently choose one option from the presented list.
|
||||
Profile:
|
||||
- You are a young professional in a big company.
|
||||
- You are avid party goer
|
||||
- You like beer.
|
||||
- You know nothing about wine.
|
||||
- You have a budget of 1500usd.
|
||||
Additional_information:
|
||||
- your boss like spicy food.
|
||||
- your boss is a middle-aged man.
|
||||
- your boss likes Australian wine.
|
||||
|
||||
Name: Kate
|
||||
Situation:
|
||||
- Your husband asked you to get him a bottle of wine. He will gift the wine to his business client while dining at a German restaurant.
|
||||
- Your husband is a business client and he will gift the wine to his business
|
||||
- You are at a wine store and start talking with the store's sommelier.
|
||||
Mission:
|
||||
- Ask the sommelier to provide multiple wine options, and subsequently choose one option from the presented list.
|
||||
Profile:
|
||||
- You are a CEO in a startup company.
|
||||
- You are a nerd
|
||||
- You don't like alcohol.
|
||||
- You have a budget of 150usd.
|
||||
- You don't care about organic, sulfite, gluten-free, or sustainability certified wines
|
||||
Additional_information:
|
||||
- your husband like spicy food.
|
||||
- your husband is a middle-aged man.
|
||||
|
||||
Name: John
|
||||
Situation:
|
||||
- A local newspaper club wants to have a scoop about wine with local food in the U.S.
|
||||
- You are at a wine store and start talking with the store's sommelier.
|
||||
Mission:
|
||||
- Ask the sommelier to provide multiple wine options, and subsequently choose one option from the presented list.
|
||||
Profile:
|
||||
- I'm a young guy.
|
||||
- I prefer to express my ideas in a succinct and clear manner.
|
||||
Additional_information:
|
||||
- N/A
|
||||
|
||||
Name: Jane
|
||||
Situation:
|
||||
- You have catering a dinner party with French cuisine.
|
||||
- You want to serve wine with your guests.
|
||||
- You are at a wine store and start talking with the store's sommelier.
|
||||
Mission:
|
||||
- Ask the sommelier to provide multiple wine options, and subsequently choose one option from the presented list.
|
||||
Profile:
|
||||
- You are a young French restaurant owner.
|
||||
- You like dry, full-bodied red wine with high tannin
|
||||
- You don't care about organic, sulfite, gluten-free, or sustainability certified wines.
|
||||
- You have a budget of 200 usd.
|
||||
Additional_information:
|
||||
- N/A
|
||||
|
||||
Let's begin!
|
||||
"""
|
||||
|
||||
header = ["Name:", "Situation:", "Mission:", "Profile:", "Additional_information:"]
|
||||
dictkey = ["name", "situation", "mission", "profile", "additional_information"]
|
||||
errornote = "N/A"
|
||||
|
||||
for attempt in 1:10
|
||||
_prompt =
|
||||
[
|
||||
Dict(:name => "system", :text => rolegenerator_systemmsg),
|
||||
]
|
||||
prompt = GeneralUtils.formatLLMtext(_prompt, "qwen3")
|
||||
|
||||
response = text2textInstructLLM(prompt) # generated role
|
||||
response = GeneralUtils.deFormatLLMtext(response, "qwen3")
|
||||
think, response = GeneralUtils.extractthink(response)
|
||||
|
||||
# check whether response has all header
|
||||
detected_kw = GeneralUtils.detect_keyword(header, response)
|
||||
kwvalue = [i for i in values(detected_kw)]
|
||||
zeroind = findall(x -> x == 0, kwvalue)
|
||||
missingkeys = [header[i] for i in zeroind]
|
||||
if 0 ∈ values(detected_kw)
|
||||
errornote = "$missingkeys are missing from your previous response"
|
||||
println("\nERROR YiemAgent rolegenerator() $errornote ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
continue
|
||||
elseif sum(values(detected_kw)) > length(header)
|
||||
errornote = "\nYour previous attempt has duplicated points according to the required response format"
|
||||
println("\nERROR YiemAgent rolegenerator() $errornote ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
continue
|
||||
end
|
||||
|
||||
responsedict = GeneralUtils.textToDict(response, header;
|
||||
dictKey=dictkey, symbolkey=true)
|
||||
responsedict[:id] = GeneralUtils.uuid4snakecase()
|
||||
|
||||
responsedict[:systemmsg] =
|
||||
"""
|
||||
You are role playing as a CUSTOMER of a wine store and you are currently talking with a sommelier of a wine store.
|
||||
Your profile is as follows:
|
||||
Situation: $(responsedict[:situation])
|
||||
Mission: $(responsedict[:mission])
|
||||
Profile: $(responsedict[:profile])
|
||||
Additional_information: $(responsedict[:additional_information])
|
||||
|
||||
You should follow the following guidelines:
|
||||
- Focus on the lastest conversation
|
||||
- Your like to be short and concise
|
||||
- If you don't know an answer to sommelier's question, you should say: I don't know.
|
||||
- If you think the store can't provide what you seek, you can leave.
|
||||
|
||||
You should then respond to the user with:
|
||||
Dialogue: what you want to say to the user
|
||||
Role: Verify that the dialogue is intended for the customer of a wine store. Can be "yes" or "no"
|
||||
You should only respond in format as described below:
|
||||
Dialogue: ...
|
||||
Role: ...
|
||||
|
||||
Let's begin!
|
||||
"""
|
||||
|
||||
println("\nrolegenerator() ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
println(responsedict)
|
||||
return responsedict
|
||||
end
|
||||
error("ERROR rolegenerator() failed to generate customer role: ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
end
|
||||
|
||||
|
||||
# Define the external functions for the customer agent in named tuple format
|
||||
customer_externalFunction = (
|
||||
text2textInstructLLM=text2textInstructLLM,
|
||||
)
|
||||
|
||||
|
||||
|
||||
|
||||
function main()
|
||||
agent = YiemAgent.sommelier(
|
||||
externalFunction;
|
||||
name="Jane",
|
||||
id=sessionId, # agent instance id
|
||||
retailername="Yiem",
|
||||
llmFormatName="qwen3"
|
||||
)
|
||||
|
||||
customerDict = rolegenerator()
|
||||
customer = YiemAgent.virtualcustomer(
|
||||
customer_externalFunction;
|
||||
systemmsg=customerDict[:systemmsg],
|
||||
name=customerDict[:name],
|
||||
id=sessionId, # agent instance id
|
||||
llmFormatName="qwen3"
|
||||
)
|
||||
|
||||
# customer_chat = "hello"
|
||||
|
||||
# YiemAgent.addNewMessage(customer, "assistant", customer_chat)
|
||||
# # add user activity to events memory
|
||||
# push!(customer.memory[:events],
|
||||
# YiemAgent.eventdict(;
|
||||
# event_description="the assistant talks to the user.",
|
||||
# timestamp=Dates.now(),
|
||||
# subject="assistant",
|
||||
# action_name="CHAT_BOX",
|
||||
# action_input=customer_chat,
|
||||
# )
|
||||
# )
|
||||
# println("\ncustomer respond:\n $customer_chat")
|
||||
agent_response = YiemAgent.conversation(agent; maximumMsg=50)
|
||||
println("\nagent respond:\n $agent_response")
|
||||
while true
|
||||
customer_chat = nothing
|
||||
while customer_chat === nothing
|
||||
customer_response = YiemAgent.conversation(customer, Dict(:text=> agent_response);
|
||||
converPartnerName=agent.name,
|
||||
maximumMsg=50)
|
||||
customer_response = GeneralUtils.deFormatLLMtext(customer_response, customer.llmFormatName)
|
||||
customer_chat = customer_response
|
||||
|
||||
#[WORKING] check whether customer response the same before
|
||||
end
|
||||
|
||||
println("\ncustomer respond:\n $customer_chat")
|
||||
|
||||
agent_response = YiemAgent.conversation(agent;
|
||||
userinput=Dict(:text=> customer_chat),
|
||||
maximumMsg=50)
|
||||
println("\nagent respond:\n $agent_response")
|
||||
|
||||
if haskey(agent.memory[:events][end], :thought)
|
||||
lastAssistantAction = agent.memory[:events][end][:thought][:action_name]
|
||||
if lastAssistantAction == "END_CONVER_GUIDELINE" # store thoughtDict
|
||||
|
||||
# save a.memory[:shortmem][:decisionlog] to disk using JSON
|
||||
println("\nsaving agent.memory[:shortmem][:decisionlog] to disk")
|
||||
date = "$(Dates.now())"
|
||||
date = replace(date, ':'=>'.')
|
||||
filename = "agent_decision_log_$(date)_$(agent.id).json"
|
||||
filepath = "/appfolder/mountvolume/appdata/log/$filename"
|
||||
open(filepath, "w") do io
|
||||
JSON.pretty(io, agent.memory[:shortmem][:decisionlog])
|
||||
end
|
||||
|
||||
# check how many file in /appfolder/mountvolume/appdata/log/ folder now
|
||||
logfilesnumber = length(readdir("/appfolder/mountvolume/appdata/log/"))
|
||||
println("\nCaching conversation process done. Total $logfilesnumber files in /appfolder/mountvolume/appdata/log/ folder now.\n")
|
||||
break
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
for i in 1:100
|
||||
main()
|
||||
println("\n Round $i/100 done.")
|
||||
end
|
||||
|
||||
println("done")
|
||||
|
||||
# prompt =
|
||||
# """
|
||||
# <|im_start|>system
|
||||
# You are a role playing agent acting as:
|
||||
# Name: Emily
|
||||
# Situation: - Emily is planning her upcoming birthday party and wants to make it extra special. She has invited close friends and family, and she's looking for a unique wine that will impress them.
|
||||
# Mission: - Emily needs to find a rare and high-quality wine that matches the theme of her party, which is a mix of classic and modern flavors. She also wants to ensure that the wine is not too expensive so that it won't break her budget.
|
||||
# Profile: - Emily is in her late 20s, works as a marketing executive for a tech company, and has a passion for trying new things. She's organized and detail-oriented but can be spontaneous when it comes to planning events.
|
||||
# Additional_information: - Emily loves experimenting with different types of food and wine pairings.
|
||||
|
||||
# Your are currently talking with a sommelier.
|
||||
|
||||
# You should follow the following guidelines:
|
||||
# - Focus on the lastest conversation
|
||||
# - If you satisfy with the sommelier's recommendation for bottle of wine(s), you should say: Thanks for you help. I will buy the wine you recommended.
|
||||
# - If you don't satisfy with the sommelier's questions or can't get a good wine recommendation, you can continue the conversation.
|
||||
|
||||
# Let's begin!
|
||||
|
||||
# <|im_end|>
|
||||
# <|im_start|>Jane
|
||||
# Hello! Welcome to Yiem's Wine Store. I'm Jane, your friendly sommelier. How can I assist you today? What type of wine are you in the mood for, and is there a special occasion or event on your mind?
|
||||
# <|im_end|>
|
||||
# <|im_start|>Emily
|
||||
# Hi Jane! Thank you so much for welcoming me. For my birthday party, I'm looking for something that combines classic and modern flavors. It's a mix of guests who enjoy both traditional tastes and more contemporary ones. Also, I want to make sure it won't break the bank. Any suggestions?
|
||||
# <|im_end|>
|
||||
# <|im_start|>Jane
|
||||
# Thank you for sharing your preferences, Jane! To better assist you, could you please let me know if there are any specific characteristics of wine you're looking for, such as tannin, sweetness, intensity, or acidity? Additionally, do you have any food items in mind that this wine should pair well with?
|
||||
# <|im_end|>
|
||||
# <|im_start|>Emily
|
||||
# """
|
||||
|
||||
# llmkwargs=Dict(
|
||||
# :num_ctx => 32768,
|
||||
# :temperature => 0.3,
|
||||
# )
|
||||
# r = text2textInstructLLM(prompt, llmkwargs=llmkwargs)
|
||||
# println(r)
|
||||
# println(555)
|
||||
|
||||
# response = YiemAgent.conversation(agent, Dict(:text=> "I want to get a French red wine under 100."))
|
||||
|
||||
|
||||
# while true
|
||||
# println("your respond: ")
|
||||
# user_answer = readline()
|
||||
# response = YiemAgent.conversation(agent, Dict(:text=> user_answer))
|
||||
# println("\n$response")
|
||||
# end
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# """
|
||||
# Hello
|
||||
|
||||
# I would like to get a bottle of wine for my boss but I don't know much about wine. Can you help me?
|
||||
|
||||
# well actually, my boss is going to offer the wine to his client as a gift in a business meeting. All I know is his client like spicy food and French wine. I have a budget about 1000.
|
||||
|
||||
# """
|
||||
|
||||
# input = "French wine, bordeaux, under USD100, pairs with spicy food"
|
||||
# r = YiemAgent.extractWineAttributes_1(a, input)
|
||||
|
||||
# inventory_order = "French Syrah, Viognier, full bodied, under 100"
|
||||
# r = YiemAgent.extractWineAttributes_2(a, inventory_order)
|
||||
# pprintln(r)
|
||||
|
||||
|
||||
# cron job
|
||||
# @reboot sleep 50 && nvidia-smi -pm 1
|
||||
# @reboot sleep 51 && nvidia-smi -i 0 -pl 150
|
||||
# @reboot sleep 52 && nvidia-smi -i 1 -pl 150
|
||||
# @reboot sleep 53 && nvidia-smi -i 2 -pl 150
|
||||
# @reboot sleep 54 && nvidia-smi -i 3 -pl 150
|
||||
|
||||
# @reboot sleep 55 && julia -t 2 /home/ton/work/restartContainer/main.jl
|
||||
|
||||
# using GeneralUtils
|
||||
# msgMeta = GeneralUtils.generate_msgMeta(
|
||||
# "/tonpc_containerServices",
|
||||
# senderName= "somename",
|
||||
# senderId= "1230",
|
||||
# mqttBrokerAddress= "mqtt.yiem.cc",
|
||||
# mqttBrokerPort= 1883,
|
||||
# )
|
||||
# outgoingMsg = Dict(
|
||||
# :msgMeta=> msgMeta,
|
||||
# :payload=> "docker container restart playground-app",
|
||||
# )
|
||||
# GeneralUtils.sendMqttMsg(outgoingMsg)
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
{
|
||||
"mqttServerInfo": {
|
||||
"description": "mqtt server info",
|
||||
"port": 1883,
|
||||
"broker": "mqtt.yiem.cc"
|
||||
},
|
||||
"testingOrProduction": {
|
||||
"value": "testing",
|
||||
"description": "agent status, couldbe testing or production"
|
||||
},
|
||||
"agentid": {
|
||||
"value": "2b74b87a-5413-4fe2-a4d3-405891051680",
|
||||
"description": "a unique id for this agent"
|
||||
},
|
||||
"agentCentralConfigTopic": {
|
||||
"mqtttopic": "/yiem_branch_1/agent/sommelier/backend/config/api/v1.1",
|
||||
"description": "a central agent server's topic to get this agent config"
|
||||
},
|
||||
"servicetopic": {
|
||||
"mqtttopic": [
|
||||
"/yiem/hq/agent/sommelier/backend/prompt/api_v1/testing"
|
||||
],
|
||||
"description": "a topic this agent are waiting for service request"
|
||||
},
|
||||
"role": {
|
||||
"value": "sommelier",
|
||||
"description": "agent role"
|
||||
},
|
||||
"organization": {
|
||||
"value": "yiem_branch_1",
|
||||
"description": "organization name"
|
||||
},
|
||||
"externalservice": {
|
||||
"loadbalancer": {
|
||||
"mqtttopic": "/loadbalancer/requestingservice",
|
||||
"description": "text to text service with instruct LLM"
|
||||
},
|
||||
"text2textinstruct": {
|
||||
"mqtttopic": "/loadbalancer/requestingservice",
|
||||
"description": "text to text service with instruct LLM",
|
||||
"llminfo": {
|
||||
"name": "llama3instruct"
|
||||
}
|
||||
},
|
||||
"virtualWineCustomer_1": {
|
||||
"mqtttopic": "/virtualenvironment/winecustomer",
|
||||
"description": "text to text service with instruct LLM that act as wine customer",
|
||||
"llminfo": {
|
||||
"name": "llama3instruct"
|
||||
}
|
||||
},
|
||||
"text2textchat": {
|
||||
"mqtttopic": "/loadbalancer/requestingservice",
|
||||
"description": "text to text service with instruct LLM",
|
||||
"llminfo": {
|
||||
"name": "llama3instruct"
|
||||
}
|
||||
},
|
||||
"wineDB" : {
|
||||
"description": "A wine database connection info for LibPQ client",
|
||||
"host": "192.168.88.12",
|
||||
"port": 10201,
|
||||
"dbname": "wineDB",
|
||||
"user": "yiemtechnologies",
|
||||
"password": "yiemtechnologies@Postgres_0.0"
|
||||
},
|
||||
"SQLVectorDB" : {
|
||||
"description": "A wine database connection info for LibPQ client",
|
||||
"host": "192.168.88.12",
|
||||
"port": 10203,
|
||||
"dbname": "SQLVectorDB",
|
||||
"user": "yiemtechnologies",
|
||||
"password": "yiemtechnologies@Postgres_0.0"
|
||||
}
|
||||
}
|
||||
}
|
||||
+706
@@ -0,0 +1,706 @@
|
||||
using JSON, JSON, Dates, UUIDs, PrettyPrinting, LibPQ, Base64, DataFrames, DataStructures
|
||||
using YiemAgent, GeneralUtils
|
||||
using Base.Threads
|
||||
|
||||
# ---------------------------------------------- 100 --------------------------------------------- #
|
||||
|
||||
|
||||
""" Expected incomming MQTT message format for this service:
|
||||
{
|
||||
"msgMeta": {
|
||||
"msgPurpose": "updateStatus",
|
||||
"requestresponse": "request",
|
||||
"timestamp": "2024-03-29T05:8:48.362",
|
||||
"replyToMsgId": null,
|
||||
"receiverId": null,
|
||||
"getpost": "get",
|
||||
"msgId": "e5c09bd8-7100-4e4e-bb43-05bee589a22c",
|
||||
"acknowledgestatus": null,
|
||||
"sendTopic": "/agent/wine/backend/chat/api/v1/prompt",
|
||||
"receiverName": "agent-wine-backend",
|
||||
"replyTopic": "/agent/wine/frontend/chat/api/v1/txt/receive",
|
||||
"senderName": "agent-wine-frontend-chat",
|
||||
"senderId": "0938a757-e0ee-40a9-8355-5e24906a87cd"
|
||||
},
|
||||
"payload" : {
|
||||
"text": "hello"
|
||||
}
|
||||
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# load config
|
||||
config = copy(JSON.parsefile("../mountvolume/config/config.json"))
|
||||
|
||||
""" Instantiate an agent. One need to specify startmessage and one of gpu location info,
|
||||
Mqtt or Rest. start message must be comply with GeneralUtils's message format
|
||||
|
||||
Arguments\n
|
||||
-----
|
||||
channel::Channel
|
||||
communication channel
|
||||
sessionId::String
|
||||
sesstion ID of the agent
|
||||
agentName::String
|
||||
Name of the agent
|
||||
mqttBroker::String
|
||||
mqtt broker e.g. "tcp://127.0.0.1:1883"
|
||||
agentConfigTopic::String
|
||||
main communication topic for an agent to ask for config
|
||||
timeout::Int64
|
||||
inactivity timeout in minutes. If timeout is reached, an agent will be terminated.
|
||||
|
||||
Return\n
|
||||
-----
|
||||
a task represent an agent
|
||||
|
||||
Example\n
|
||||
-----
|
||||
```jldoctest
|
||||
julia> using YiemAgent, GeneralUtils
|
||||
julia> msg = GeneralUtils.generate_msgMeta("/agent")
|
||||
julia> incoming_msg = msg # assuming 1st msg was sent from other app
|
||||
julia> agentConfigTopic = "/agent/wine/backend/config"
|
||||
julia> task = runAgentInstance(incoming_msg, mqttBroker, agentConfigTopic, 60)
|
||||
```
|
||||
|
||||
TODO\n
|
||||
-----
|
||||
[] update docstringLAMA_CONTEXT_LENGTH=40960 since the default size is 2048 as you can see in your debug log:
|
||||
[] change how to get result of YiemAgent from let YiemAgent send msg directly to frontend,
|
||||
to
|
||||
response = YiemAgent.conversation()
|
||||
then send response to frontend
|
||||
|
||||
Signature\n
|
||||
-----
|
||||
"""
|
||||
function runAgentInstance(
|
||||
receiveUserMsgChannel::Channel,
|
||||
outputchannel::Channel,
|
||||
sessionId::String,
|
||||
config::Dict,
|
||||
timeout::Int64,
|
||||
)
|
||||
|
||||
function executeSQL(sql::T) where {T<:AbstractString}
|
||||
host = config[:externalservice][:wineDB][:host]
|
||||
port = config[:externalservice][:wineDB][:port]
|
||||
dbname = config[:externalservice][:wineDB][:dbname]
|
||||
user = config[:externalservice][:wineDB][:user]
|
||||
password = config[:externalservice][:wineDB][:password]
|
||||
DBconnection = LibPQ.Connection("host=$host port=$port dbname=$dbname user=$user password=$password")
|
||||
result = LibPQ.execute(DBconnection, sql)
|
||||
close(DBconnection)
|
||||
return result
|
||||
end
|
||||
|
||||
function executeSQLVectorDB(sql)
|
||||
host = config[:externalservice][:SQLVectorDB][:host]
|
||||
port = config[:externalservice][:SQLVectorDB][:port]
|
||||
dbname = config[:externalservice][:SQLVectorDB][:dbname]
|
||||
user = config[:externalservice][:SQLVectorDB][:user]
|
||||
password = config[:externalservice][:SQLVectorDB][:password]
|
||||
DBconnection = LibPQ.Connection("host=$host port=$port dbname=$dbname user=$user password=$password")
|
||||
result = LibPQ.execute(DBconnection, sql)
|
||||
close(DBconnection)
|
||||
return result
|
||||
end
|
||||
|
||||
function text2textInstructLLM(prompt::String; maxattempt::Integer=3, modelsize::String="medium",
|
||||
senderId=GeneralUtils.uuid4snakecase(), timeout=180,
|
||||
llmkwargs=Dict(
|
||||
:num_ctx => 32768,
|
||||
:temperature => 0.5,
|
||||
))
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
config[:externalservice][:loadbalancer][:mqtttopic];
|
||||
msgPurpose="inference",
|
||||
senderName="yiemagent",
|
||||
senderId=senderId,
|
||||
receiverName="text2textinstruct_$modelsize",
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => Dict(
|
||||
:text => prompt,
|
||||
:kwargs => llmkwargs
|
||||
)
|
||||
)
|
||||
|
||||
response = nothing
|
||||
for attempts in 1:maxattempt
|
||||
_response = GeneralUtils.sendReceiveMqttMsg(outgoingMsg; timeout=timeout, maxattempt=maxattempt)
|
||||
payload = _response[:response]
|
||||
if _response[:success] && payload[:text] !== nothing
|
||||
response = _response[:response][:text]
|
||||
break
|
||||
else
|
||||
println("\n<text2textInstructLLM()> attempt $attempts/$maxattempt failed ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
pprintln(outgoingMsg)
|
||||
println("</text2textInstructLLM()> attempt $attempts/$maxattempt failed ", @__FILE__, ":", @__LINE__, " $(Dates.now())\n")
|
||||
sleep(3)
|
||||
end
|
||||
end
|
||||
|
||||
return response
|
||||
end
|
||||
|
||||
# get text embedding from a LLM service
|
||||
function getEmbedding(text::T) where {T<:AbstractString}
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
config[:externalservice][:loadbalancer][:mqtttopic];
|
||||
msgPurpose="embedding",
|
||||
senderName="yiemagent",
|
||||
senderId=sessionId,
|
||||
receiverName="textembedding",
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => Dict(
|
||||
:text => [text] # must be a vector of string
|
||||
)
|
||||
)
|
||||
|
||||
response = GeneralUtils.sendReceiveMqttMsg(outgoingMsg; timeout=120, maxattempt=3)
|
||||
embedding = response[:response][:embeddings]
|
||||
return embedding
|
||||
end
|
||||
|
||||
function findSimilarTextFromVectorDB(text::T1, tablename::T2, embeddingColumnName::T3,
|
||||
vectorDB::Function; limit::Integer=1
|
||||
)::DataFrame where {T1<:AbstractString, T2<:AbstractString, T3<:AbstractString}
|
||||
# get embedding from LLM service
|
||||
embedding = getEmbedding(text)[1]
|
||||
# check whether there is close enough vector already store in vectorDB. if no, add, else skip
|
||||
sql = """
|
||||
SELECT *, $embeddingColumnName <-> '$embedding' as distance
|
||||
FROM $tablename
|
||||
ORDER BY distance LIMIT $limit;
|
||||
"""
|
||||
response = vectorDB(sql)
|
||||
df = DataFrame(response)
|
||||
return df
|
||||
end
|
||||
|
||||
function similarSQLVectorDB(query; maxdistance::Integer=100)
|
||||
tablename = "sqlllm_decision_repository"
|
||||
# get embedding of the query
|
||||
df = findSimilarTextFromVectorDB(query, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
# println(df[1, [:id, :function_output]])
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
# distance = 100 # CHANGE this is for testing only
|
||||
if row != 0 && distance < maxdistance
|
||||
# if there is usable SQL, return it.
|
||||
output_b64 = df[1, :function_output_base64] # pick the closest match
|
||||
output_str = String(base64decode(output_b64))
|
||||
rowid = df[1, :id]
|
||||
println("\n~~~ found similar sql. row id $rowid, distance $distance ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
return (dict=output_str, distance=distance)
|
||||
else
|
||||
println("\n~~~ similar sql not found, max distance $maxdistance ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
return (dict=nothing, distance=nothing)
|
||||
end
|
||||
end
|
||||
|
||||
function insertSQLVectorDB(query::T1, SQL::T2; maxdistance::Integer=3) where {T1<:AbstractString, T2<:AbstractString}
|
||||
tablename = "sqlllm_decision_repository"
|
||||
# get embedding of the query
|
||||
# query = state[:thoughtHistory][:question]
|
||||
df = findSimilarTextFromVectorDB(query, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
if row == 0 || distance > maxdistance # no close enough SQL stored in the database
|
||||
query_embedding = getEmbedding(query)[1]
|
||||
query = replace(query, "'" => "")
|
||||
sql_base64 = base64encode(SQL)
|
||||
sql_ = replace(SQL, "'" => "")
|
||||
|
||||
sql = """
|
||||
INSERT INTO $tablename (function_input, function_output, function_output_base64, function_input_embedding) VALUES ('$query', '$sql_', '$sql_base64', '$query_embedding');
|
||||
"""
|
||||
# println("\n~~~ added new decision to vectorDB ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
# println(sql)
|
||||
_ = executeSQLVectorDB(sql)
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
function similarSommelierDecision(recentevents::T1; maxdistance::Integer=3
|
||||
)::Union{AbstractDict, Nothing} where {T1<:AbstractString}
|
||||
tablename = "sommelier_decision_repository"
|
||||
# find similar
|
||||
println("\n~~~ search vectorDB for this: $recentevents ", @__FILE__, " ", @__LINE__)
|
||||
df = findSimilarTextFromVectorDB(recentevents, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
if row != 0 && distance < maxdistance
|
||||
# if there is usable decision, return it.
|
||||
rowid = df[1, :id]
|
||||
println("\n~~~ found similar decision. row id $rowid, distance $distance ", @__FILE__, " ", @__LINE__)
|
||||
output_b64 = df[1, :function_output_base64] # pick the closest match
|
||||
_output_str = String(base64decode(output_b64))
|
||||
output = copy(JSON.parsefile(_output_str))
|
||||
return output
|
||||
else
|
||||
println("\n~~~ similar decision not found, max distance $maxdistance ", @__FILE__, " ", @__LINE__)
|
||||
return nothing
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
function insertSommelierDecision(recentevents::T1, decision::T2; maxdistance::Integer=5
|
||||
) where {T1<:AbstractString, T2<:AbstractDict}
|
||||
tablename = "sommelier_decision_repository"
|
||||
# find similar
|
||||
df = findSimilarTextFromVectorDB(recentevents, tablename,
|
||||
"function_input_embedding", executeSQLVectorDB)
|
||||
row, col = size(df)
|
||||
distance = row == 0 ? Inf : df[1, :distance]
|
||||
if row == 0 || distance > maxdistance # no close enough SQL stored in the database
|
||||
recentevents_embedding = getEmbedding(recentevents)[1]
|
||||
recentevents = replace(recentevents, "'" => "")
|
||||
decision_json = JSON.json(decision)
|
||||
decision_base64 = base64encode(decision_json)
|
||||
decision = replace(decision_json, "'" => "")
|
||||
|
||||
sql =
|
||||
"""
|
||||
INSERT INTO $tablename (function_input, function_output, function_output_base64, function_input_embedding) VALUES ('$recentevents', '$decision', '$decision_base64', '$recentevents_embedding');
|
||||
"""
|
||||
println("\n~~~ added new decision to vectorDB ", @__FILE__, " ", @__LINE__)
|
||||
println(sql)
|
||||
_ = executeSQLVectorDB(sql)
|
||||
else
|
||||
println("~~~ similar decision previously cached, distance $distance ", @__FILE__, " ", @__LINE__)
|
||||
end
|
||||
end
|
||||
|
||||
# keepaliveChannel_2::Channel{Dict} = Channel{Dict}(8)
|
||||
latestUserMsgTimeStamp::DateTime = Dates.now()
|
||||
|
||||
externalFunction = (
|
||||
getEmbedding=getEmbedding,
|
||||
text2textInstructLLM=text2textInstructLLM,
|
||||
executeSQL=executeSQL,
|
||||
similarSQLVectorDB=similarSQLVectorDB,
|
||||
insertSQLVectorDB=insertSQLVectorDB,
|
||||
similarSommelierDecision=similarSommelierDecision,
|
||||
insertSommelierDecision=insertSommelierDecision,
|
||||
)
|
||||
|
||||
agent = YiemAgent.sommelier(
|
||||
externalFunction;
|
||||
name="Jane",
|
||||
id=sessionId, # agent instance id
|
||||
retailername="Yiem",
|
||||
llmFormatName="qwen3"
|
||||
)
|
||||
|
||||
# user chat loop
|
||||
while true
|
||||
# check for new user message
|
||||
if isready(receiveUserMsgChannel)
|
||||
incomingMsg = take!(receiveUserMsgChannel)
|
||||
incoming_msgMeta = incomingMsg[:msgMeta]
|
||||
incomingPayload = incomingMsg[:payload]
|
||||
latestUserMsgTimeStamp = Dates.now()
|
||||
|
||||
# make sure the message has :text key because YiemAgent use this key for incoming user msg
|
||||
if haskey(incomingPayload, :text)
|
||||
# skip, msg already has correct key name
|
||||
elseif haskey(incomingPayload, :txt)
|
||||
# change key name to text
|
||||
incomingPayload[:text] = incomingPayload[:txt]
|
||||
else
|
||||
error("\n no :txt or :text key in the message.")
|
||||
end
|
||||
|
||||
# reset agent
|
||||
if occursin("newtopic", incomingPayload[:text]) ||
|
||||
occursin("Newtopic", incomingPayload[:text]) ||
|
||||
occursin("New topic", incomingPayload[:text]) ||
|
||||
occursin("new topic", incomingPayload[:text])
|
||||
# YiemAgent.clearhistory(agent)
|
||||
|
||||
agent = YiemAgent.sommelier(
|
||||
externalFunction;
|
||||
name="Janie",
|
||||
id=sessionId, # agent instance id
|
||||
retailername="Yiem",
|
||||
)
|
||||
|
||||
# sending msg back to sender i.e. LINE
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
incomingMsg[:msgMeta][:replyTopic];
|
||||
senderName="wine_assistant_backend",
|
||||
senderId=sessionId,
|
||||
replyToMsgId=incomingMsg[:msgMeta][:msgId],
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => Dict(
|
||||
:alias => agent.name, # will be shown in frontend as agent name
|
||||
:text => "Okay. What shall we talk about?"
|
||||
)
|
||||
)
|
||||
_ = GeneralUtils.sendMqttMsg(outgoingMsg)
|
||||
println("--> outgoingMsg ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
pprintln(outgoingMsg)
|
||||
else
|
||||
usermsg = incomingPayload
|
||||
|
||||
if incoming_msgMeta[:msgPurpose] == "initialize"
|
||||
println("\n-- Initializing... ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
end
|
||||
|
||||
# send prompt
|
||||
result = YiemAgent.conversation(agent;
|
||||
userinput=usermsg,
|
||||
maximumMsg=50)
|
||||
# Ken's bot use [br] for newline character '\n'
|
||||
# result = replace(result, '\n'=>"[br]")
|
||||
|
||||
if incoming_msgMeta[:msgPurpose] == "initialize"
|
||||
println("\n-- Initialized. Ready! waiting for request at:\n$(config[:servicetopic][:mqtttopic]) ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
continue
|
||||
end
|
||||
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
incomingMsg[:msgMeta][:replyTopic];
|
||||
senderName="wine_assistant_backend",
|
||||
senderId=string(uuid4()),
|
||||
replyToMsgId=incomingMsg[:msgMeta][:msgId],
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => Dict(
|
||||
:alias => agent.name, # will be shown in frontend as agent name
|
||||
:text => result
|
||||
)
|
||||
)
|
||||
_ = GeneralUtils.sendMqttMsg(outgoingMsg)
|
||||
println("\n--> outgoingMsg ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
pprintln(outgoingMsg)
|
||||
|
||||
|
||||
# jpg_as_juliaStr = nothing
|
||||
# prompt = nothing
|
||||
|
||||
# if haskey(payload, "img")
|
||||
# url_or_base64 = payload["img"]
|
||||
|
||||
# if startswith(url_or_base64, "http")
|
||||
# # img in http
|
||||
# julia_rgb_img, cv2_bgr_img = ImageUtils.url_to_cv2_image(url_or_base64)
|
||||
# _, buffer = cv2.imencode(".jpg", cv2_bgr_img)
|
||||
# jpg_as_pyStr = base64.b64encode(buffer).decode("utf-8")
|
||||
# jpg_as_juliaStr = pyconvert(String, jpg_as_pyStr)
|
||||
# else
|
||||
# # img in base64
|
||||
# cv2_bgr_img = payload["img"]
|
||||
# jpg_as_juliaStr = pyconvert(String, jpg_as_pyStr)
|
||||
# end
|
||||
# end
|
||||
|
||||
end
|
||||
else
|
||||
# println("\n no msg")
|
||||
end
|
||||
|
||||
if haskey(agent.memory[:events][end], :thought)
|
||||
lastAssistantAction = agent.memory[:events][end][:thought][:action_name]
|
||||
if lastAssistantAction == "END_CONVER_GUIDELINE" # store thoughtDict
|
||||
|
||||
# save a.memory[:shortmem][:decisionlog] to disk using JSON
|
||||
println("\nsaving agent.memory[:shortmem][:decisionlog] to disk")
|
||||
filename = "agent_decision_log_$(Dates.now())_$(agent.id).json"
|
||||
filepath = "/appfolder/app/log/$filename"
|
||||
open(filepath, "w") do io
|
||||
JSON.pretty(io, agent.memory[:shortmem][:decisionlog])
|
||||
end
|
||||
|
||||
# for (i, event) in enumerate(agent.memory[:events])
|
||||
# if event[:subject] == "assistant"
|
||||
# # create timeline of the last 3 conversation except the last one.
|
||||
# # The former will be used as caching key and the latter will be the caching target
|
||||
# # in vector database
|
||||
# all_recapkeys = keys(agent.memory[:recap]) #[TESTING] recap as caching
|
||||
# all_recapkeys_vec = [r for r in all_recapkeys] # convert to a vector
|
||||
|
||||
# # select from 1 to 2nd-to-lase event (i.e. excluding the latest which is assistant's response)
|
||||
# _recapkeys_vec = all_recapkeys_vec[1:i-1]
|
||||
|
||||
# # select only previous 3 recaps
|
||||
# recapkeys_vec =
|
||||
# if length(_recapkeys_vec) <= 3 # 1st message is a user's hello msg
|
||||
# _recapkeys_vec # choose all
|
||||
# else
|
||||
# _recapkeys_vec[end-2:end]
|
||||
# end
|
||||
# #[PENDING] if there is specific data such as number, donot store in database
|
||||
# tempmem = DataStructures.OrderedDict()
|
||||
# for k in recapkeys_vec
|
||||
# tempmem[k] = agent.memory[:recap][k]
|
||||
# end
|
||||
|
||||
# recap = GeneralUtils.dictToString_noKey(tempmem)
|
||||
# thoughtDict = agent.memory[:events][i][:thought] # latest assistant thoughtDict
|
||||
# insertSommelierDecision(recap, thoughtDict)
|
||||
# else
|
||||
# # skip
|
||||
# end
|
||||
# end
|
||||
println("\nCaching conversation process done")
|
||||
break
|
||||
end
|
||||
end
|
||||
|
||||
# self terminate if too long inactivity
|
||||
timediff = GeneralUtils.timedifference(latestUserMsgTimeStamp, Dates.now(), "minutes")
|
||||
if timediff > timeout
|
||||
|
||||
result = Dict(:exitreason => "timeout", :timestamp => Dates.now())
|
||||
put!(outputchannel, result)
|
||||
println("Agent ID $(agent.id) timeout has been reached $timediff/$timeout minutes Send delete session msg ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
|
||||
# send "delete session" message to inform the main loop that this session can be deleted
|
||||
sendto =
|
||||
if typeof(config[:servicetopic][:mqtttopic]) <: Array
|
||||
config[:servicetopic][:mqtttopic][1]
|
||||
else
|
||||
config[:servicetopic][:mqtttopic]
|
||||
end
|
||||
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
sendto;
|
||||
senderName="session",
|
||||
senderId=sessionId,
|
||||
msgPurpose="delete session",
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => nothing
|
||||
)
|
||||
_ = GeneralUtils.sendMqttMsg(outgoingMsg)
|
||||
|
||||
try disconnect(agent.mqttClient) catch end
|
||||
break
|
||||
end
|
||||
sleep(1) # allowing on_msg_2, asyncmove above and other process to run
|
||||
end
|
||||
end
|
||||
|
||||
sessionDict = Dict{String,Any}()
|
||||
incomingMsgChannel = (ch1=Channel(8),) # store msg that coming into servicetopic
|
||||
# incommingInternalMsg = [] # st ore msg that coming into servicetopic internal management
|
||||
keepaliveChannel::Channel{Dict} = Channel{Dict}(8)
|
||||
|
||||
# Define the callback for receiving messages.
|
||||
function onMsgCallback_1(topic, payload)
|
||||
jobj = JSON.parsefile(String(payload))
|
||||
incomingMqttMsg = copy(jobj) # convert json object into julia dictionary recursively
|
||||
|
||||
if occursin("keepalive", topic)
|
||||
put!(keepaliveChannel, incomingMqttMsg)
|
||||
else
|
||||
put!(incomingMsgChannel[:ch1], incomingMqttMsg)
|
||||
end
|
||||
end
|
||||
|
||||
mqttInstance = GeneralUtils.mqttClientInstance_v2(
|
||||
config[:mqttServerInfo][:broker],
|
||||
config[:servicetopic][:mqtttopic],
|
||||
incomingMsgChannel,
|
||||
keepaliveChannel,
|
||||
onMsgCallback_1
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------------------------------------ #
|
||||
# this service main loop #
|
||||
# ------------------------------------------------------------------------------------------------ #
|
||||
|
||||
function main()
|
||||
sessiontimeout = 1 * 1 * 60 # timeout in minute for each instance (day * hour * minute)
|
||||
initializing = false
|
||||
while true
|
||||
# check if mqtt connection is still up
|
||||
_ = GeneralUtils.checkMqttConnection!(mqttInstance; keepaliveCheckInterval=30)
|
||||
|
||||
# initialize session 0
|
||||
if initializing == false # send init msg
|
||||
sendto =
|
||||
if typeof(config[:servicetopic][:mqtttopic]) <: Array
|
||||
config[:servicetopic][:mqtttopic][1]
|
||||
else
|
||||
config[:servicetopic][:mqtttopic]
|
||||
end
|
||||
|
||||
msgMeta = GeneralUtils.generate_msgMeta(
|
||||
sendto;
|
||||
msgPurpose="initialize",
|
||||
senderName="initializer",
|
||||
senderId="0",
|
||||
msgId= "initMsg",
|
||||
replyTopic=sendto,
|
||||
mqttBrokerAddress=config[:mqttServerInfo][:broker],
|
||||
mqttBrokerPort=config[:mqttServerInfo][:port],
|
||||
)
|
||||
|
||||
outgoingMsg = Dict(
|
||||
:msgMeta => msgMeta,
|
||||
:payload => Dict( # will be shown in frontend as agent name
|
||||
:text => "Do you have full-bodied red wines under 100 USD. I don't have any other preferences."
|
||||
)
|
||||
)
|
||||
_ = GeneralUtils.sendMqttMsg(outgoingMsg)
|
||||
initializing = true
|
||||
println("\n--> Initializing msg sent ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
end
|
||||
|
||||
# check for new message
|
||||
if !isempty(incomingMsgChannel[:ch1])
|
||||
msg = popfirst!(incomingMsgChannel[:ch1])
|
||||
println("\n<-- incomingMsg ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
pprintln(msg)
|
||||
|
||||
# @spawn new runAgentInstance and store it in sessionDict
|
||||
# use agent's frontend id because 1 backend agent per 1 frontend session
|
||||
sessionId = msg[:msgMeta][:senderId]
|
||||
sessionId = replace(sessionId, "-" => "_") # julia can't use "-" in a dict key
|
||||
|
||||
# check for delete session msg
|
||||
if msg[:msgMeta][:msgPurpose] == "delete session"
|
||||
delete!(sessionDict, sessionId)
|
||||
println("sessionId $(sessionId) has been terminated ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
|
||||
# no session yet, create new session
|
||||
elseif sessionId ∉ keys(sessionDict)
|
||||
inputch = Channel{Dict}(8)
|
||||
outputch = Channel{Dict}(8)
|
||||
|
||||
process = @spawn runAgentInstance(inputch, outputch, sessionId, config, sessiontimeout)
|
||||
# process = runAgentInstance(inputch, outputch, sessionId, config, sessiontimeout) #XXX use spawn version
|
||||
|
||||
println("\ninstantiate agent success ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
|
||||
# call runAgentInstance() and store it in sessionDict to be able to check on it later
|
||||
sessionDict[sessionId] = Dict(
|
||||
:inputchannel => inputch,
|
||||
:outputchannel => outputch,
|
||||
:process => process,
|
||||
)
|
||||
put!(sessionDict[sessionId][:inputchannel], msg)
|
||||
# ongoing session
|
||||
else
|
||||
println("sessionId $(sessionId) existing session ", @__FILE__, ":", @__LINE__, " $(Dates.now())")
|
||||
put!(sessionDict[sessionId][:inputchannel], msg)
|
||||
end
|
||||
end
|
||||
|
||||
# sleep is needed because MQTTClient use async. "while true" loop leave no
|
||||
# chance for control to switch to on_msg()
|
||||
sleep(1)
|
||||
end
|
||||
end
|
||||
|
||||
main()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
iVBORw0KGgoAAAANSUhEUgAAAIAAAACACAYAAADDPmHLAAAACXBIWXMAAAsTAAALEwEAmpwYAAAAtGVYSWZJSSoACAAAAAYAEgEDAAEAAAABAAAAGgEFAAEAAABWAAAAGwEFAAEAAABeAAAAKAEDAAEAAAACAAAAEwIDAAEAAAABAAAAaYcEAAEAAABmAAAAAAAAAEgAAAABAAAASAAAAAEAAAAGAACQBwAEAAAAMDIxMAGRBwAEAAAAAQIDAACgBwAEAAAAMDEwMAGgAwABAAAA//8AAAKgBAABAAAAgAAAAAOgBAABAAAAgAAAAAAAAABKviP0AAANMElEQVR4nO2de3BU1R3Hv7+zu9nNO9mQTQIJzxAeQkKI5MFjAgJJrEiSVmZqq2L/UdvOVO34Xx9WWzvTTutjHMdHrVbb6UyLDqC0jQ9s66ABLaIiCAgmQEhIYh6bBLLPezpnIZDdvQs3mMfde+5n5s5kz97du9nzu79zft/z+50FTExMTExMTExMTExMTExMTExMTEwMTeWWLYnigMRYITODvrsv/vU4JIUgKdXVWx0eh/s4gVv8tuC8/bt2nYeEMEiKN7H/HiLMAFGuLWC5C5JCUt/9hBmhBs7PyuoFmNR3/wgSewHpPEDU3T+CpF5AOg/gcbgfiOp8AVGu1W/9MSRDGg9QWbt5A8B+BULFlc/knxDYI81N218RD2BwDG8A2jteTkMwrAFce8fLZQgWGJCqmvqVYOxFEJZ8/XejXADXF8xd8GHbiaOnYTAMOQlsfnPn+/5u1wKQshXgX1zr+3DOT3KO+zLgWSTeEwbEsEPACGVld9lsrrO3gtNPAZqvteMBeiyTPM80NTV5YWAMbwBhhpDdeQdADwIoiHHaaYA/5O/OeXn//uf8kABpDGD0/IAzek/tOWJKdfM/X3sXEmHIOcCVEGM559gd2c45dsvW+VIagIBx/vOoNovyMCRESgOI9AJc0rtfWgOI9AJM0rtfeipqG94WByRG6pxADv4QJEdqAwDxHFLIcPr+WJBOBxihvK4ujbjjMBFndjsW/3fnzn5IiLSTQIL9kQuJIZTn8ZC0k0ApPUBVXeMKhSvNRHRhNZRzhYFWv//GjmZIhnQeoLq62qpw/uylzhcQMQV4RqwXQDKkM4DhxIz7iFAa9QSh2JrV+SNIhlRDwKoNm2YGrJZDBEpRPYHjvKIEl3zw1ustkASpPEDQan0yZucLCEmMsacgEdIYQHnd5lsAbL7qiUQ3VtTUfwuSQHLF/Cr1AKrwDlm0ASZXzB9O7uz+0KHyCmm0ASZDzA+O70e2W6wKSta3YtmGFlhtStTrCPyHK2sbqmBwmHQx/0UWVbUhOd2DpDQvFlSciX6xJNoAkzHmT5t2HoXLz156XHR9O9KzVWpCJdAGmJFjfgJEBnAYBI7lG1pEEggutTGO5Ru/DD0XdT7Rw+Ubb54Dg8Jki/nnlHTBOX0w6vzM3CHMLumSThtgMsX8jmQfrlt9Kubrlqw5BUeKTyptgBkz5mequ34VrzsJmz0Y87W2hCCWVqsbCDE8WV1fnwGDwWSJ+V2z3cgv6rnq6wsWfIXcuX3SaANMlpi/dL329Z1lN7RKow0wmWJ+rSRJpA0w2WJ+rRRJog0w2WJ+rZAk2gCTMebXSqYE2gCTNebXyhKDawNM1phfKzaDawNMtpg/rc+G6a1JsA9r/9eNrA2QYXL7R8X8G7Z+qhr2LdvjxJIPnKHN3nx2Bf9pbEfXDG3h4fkBO95+qQQBf4ThxHlNAZMl5hd3/kjnCxK8DCveyR6TNrCwss1w2gCTJeZP6bdFbfOY7B5bbez8sg7DaQNMlpi/J9cTcvuj6Zg1DNm1ASZLzO9NvDDm97m88NmDOLlgCB9s6L4mbUBcyyjaAMVbzM/AtqnF/Bvv/GRcwj4t+H0WvP2nZRgeih72ucJv2ffmzlcRJzBDxPxrxyfmH5M2sLbVENoAi/eYP2dOP/IXXH2df7zJL+oxhDbA4n2dX6zdTxXLDJA3wGRa5x9vkgygDVjjIeZniI75iYCe9lTs23V5A/CsGUMoLO0Y8zUK04PIT1Yw4Gc42GOB/wqrx8cP5KHnzOUgRAmy0GfhPKY28HvoGIrrev5RMAvH+ts/RapzbLH98uwgKlyXNwbvGmbY3pIAJcbeYYN9Duz+czGUgAbnGQf7DbC4rucfhXDFY+18wVJn+K7wrkQldMQiNdODBeUq6WJqxIE2wOK+nv+iBFy0on3SHOPCcpEudk7jW+k7b4DFW8wfiZBlS4UEzK5tv8eDveHzNDEEdA3TVSXh0o0tII17TOpZG9DlJJAhYT0n/v7I10ucCMTrAYqaVc8p7USWigTMSLhzHpqcdXso5pj+UbcFAz5CQUoQbi/hYJ815rmjcYYk4U58+bH4TalIuB+cdnK6PDX0evk6ANuhM3Q9CRyhorb+TiJ6MbJdpGpt3BotAdsswOZZvktjeecw4bVWOwJjzw3VIAmXYHgoQe3pO/c27XgJOkeXQ8BoyjZtmgbgt2rPiWIPNQm42BkIm8jlJHIsdQYnRBJetkF9gs/BH63a2OiCztG9AVgD1ieIKCpzY0ZRL/LmqUmxQFpCtA9PTxjn2/8ieXP7MH1+LyIhkFNhyqPQObo2gMqaxjoCvqN25xWvjR1atw1F/x7m6SE2oZKwTcUTEdF3q2oaNkHH6PaXQ8s2bUqyKLQLRJmRz5Wsa0V2Qey1/17vhZSNTDsPqXoffWXD4b6J+1etCUEk2IM425KpNhSsKswvfv7kycMqueVTjy6jAIEtYPkliOZGtjvzBjG7+OqJHPu7raHjWln8YSYWHkgPpZEdLXXjULn6cDOCKCA5dWQaes6khrUT0Sxvov8XAB6ADtGlByivaSwB4Y8UMUSJWH/lN4/AkTSxv+k461gKKt9yweZjoSPvVBLcWT64p/muGE5NmzGIls9c4Ep4cMU5r8ift+gfZ04cGftChWxzALH6R0x5gVS8k6jYTcsau9w7VvJaE6PaRC3B1UgRMrGKIilWMonwgh5XB3VnAF5H5v0EWq71y50IhtIC0W0Z2rxOTCMlFFuyO++FztCVAVTU1s8GqWT8ElBWcwLMOjGhXCRHy9zozbn8m9G9Li+OlLqhBTFMLa9RzxxmnB5aWdcwDzpCV5NAAp4V6fpqs+zjB3JDx2gdQMuWL6PhfhvOHShDwJ0Ox+wWOOYfUz3Pb1PQdGsbsjscocfdeR4oltj6cNuxLJw55gz/zHYFfm/EFIuQpAB/ALA+FCDoAN1IweW19bczope1nHtNWcAKw1d/vQ2+jumXmtLW/hspK/bh6+L3WkNlY2pZwmpw4PZ9TTv+Ah2giyFgxfqGLCL8Tuv5QgcYaxawr316WOcLzu1fgfHAZg+geJ32nA8Cf3x13RbtdWlGNwCLDY8RSJNuLjJxhfsfF1+nYNyYMb8X0wv7tH6YLD/3azZ4QxtAec1msUx6m5ZzhQQs1v6vBVtuB2yuzrC25OUfYTwpibE4pQYR7iiva6iB7EJQQeGiezhxIZAcvnBQGwGFaueW3CAk4IGodpuHwdV+IXb3OZSYSRxJiz4HWQOwpA0gtWIfkkvHbgABdwb8XTlgdg/IGrEMLSRhRwBnv4yWhAUceAPgB0b+V8Ypo+3EkSn97WLdTAJHqKirf5pA90S2T8sfwJoth0Mh4WhcZxxYu2M6EjwsNK3+ZFUPPqvU6orHxmDzKgzuWR362pjDA2fjK0jID08LFykge15dhO5T6dFvwPnTe9/Y+QPoiCkfAkYjiimI4y61jF/h+iM7X1C+OzvU+QLxdMl7WUgVpeDjTKAv81LnCxSPA+7d0R5cfMbS9a2hohUV7i6vrV8DHaEbA6irqxPF28+Logq1ApBYGb+huv9RiO5Jdo+/AQQHREofRRmF6mfKHMaiSvWNJhnwfHX11gsCgw7QjQH0c/vPQFgc2S42ZBAbM8Ti7MxwwxB7APTmjn+1kC2nA8xxWR0U2GfFLkubf307MlwqmcNERd7E/p9AJ0z5JFBQeWP9UgAvA+HlX2LiVtVwFEmpsVfhOgs8SO+zIXnQBrfTh+ZvdGEgItd/PCBrALa8dgQ6c8H9djjmnkBGTRMoQf1aYihw5p3DyUPZ4DxidRC0cmZh0a6240fHvoWp4SaBDz7IKvZ9vIdAUcWUItdf7NMXzxx8dya++F+4ACXgnH84My2hatu2bZNX167HIaCi+eP71To/OT1G4WWcsXhlG5IzoockIlpxasB/r9RDQGj1j+HvBEqIdJ/lN32BVOfUVf6OF2J1MH3aeZz+PFr5JWDNzPkL/3b6+JGJiVv17gHE6p9a7d+s67rgmqVt+TUeyC4YwMzF6vsKXVwdJOkMoKJm8/dAVKO20re0+iSMRvHaU0hMUZ0wrquord+KKYKmqtjD5rd+DoIo+ojK/NFceBlnuLuTMdQXLQFw8F74lMX73nk9fLHCqAkh1oDlKbXOF4gvSO1LMjIEcsJmeQLAtyf/2pNMeW3jTYz4rsm+bjxAxBub/7Vzx6Rec7LLvhm3HwJR/mReN37gk/6z9ZM6CSTYf2N2/pWgPK8Hv4YRPUDFDTfnkM3yBCc+4UZHHFWRhpZoZciwa5vy9HsDGI6sJee8jRMmfEt44qRwf/DeyZoQTr0UPAFU1jW8AiBsW5b81ARU5mrabgh7zw6hbTBq/eHVvU07xLY1hmLKpWCTqcU0AMkxDUByTAOQHF2Vhk0kZwZ92D6kbdFNidr31bhIYwCiS4MSdaxWzCFAckwDkBzTACTHkHMArtCzYPzNcX1ThXS75buJiYmJiYmJiYmJiYmJiYmJiYmJCa7I/wEjf8WA9CzV0gAAAABJRU5ErkJggg==
|
||||
@@ -1,94 +0,0 @@
|
||||
can you make sure yiemAgent use correct format when contacting MCP server. you can check MCP server doc at /home/ton/docker-apps/sommpanion/mcp/docs/specification.md?
|
||||
|
||||
smartpack() and smartunpack() doc is at /home/ton/docker-apps/sommpanion/msghandler/README.md
|
||||
|
||||
user code that use yiemAgent is here /home/ton/docker-apps/sommpanion/agent-backend/test/runtest.jl
|
||||
|
||||
this is log folder that logs when I ran yiemAgent. you can check /home/ton/docker-apps/sommpanion/agent-backend/log/
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+58
-88
@@ -170,7 +170,7 @@ function yiemAgent(
|
||||
|
||||
tools = OrderedDict{String, agentTool}()
|
||||
registerAllTools(tools, mcpServer; eventSink=eventSink, sessionId=sessionId)
|
||||
|
||||
eventSink("instantiate an agent")
|
||||
# Create struct with a placeholder task, then spawn and replace it
|
||||
agent = yiemAgent(
|
||||
agentState(systemPrompt, model, tools, messages),
|
||||
@@ -194,7 +194,7 @@ function yiemAgent(
|
||||
|
||||
# Spawn the background loop and attach it
|
||||
agent._agentLoop = @spawn _agentLoop(agent)
|
||||
|
||||
eventSink("agent instantiated")
|
||||
return agent
|
||||
end
|
||||
|
||||
@@ -260,42 +260,36 @@ function _agentLoop(agent::yiemAgent)
|
||||
agent.inputChannel -> nothing
|
||||
agent.followUpChannel -> nothing
|
||||
"""
|
||||
|
||||
agent.eventSink("_agentLoop starting")
|
||||
while true
|
||||
while newUserMsg === nothing
|
||||
if isready(agent.inputChannel)
|
||||
agent.eventSink("_agentLoop 1 agent._state.messages length $(length(agent._state.messages))"; publish=true)
|
||||
# agent process new user msg immediately after the current tool call finished.
|
||||
newUserMsg = take!(agent.inputChannel)
|
||||
# agent.eventSink("new user msg\n$newUserMsg"; publish=true)
|
||||
agent.eventSink("take new user message")
|
||||
else
|
||||
# check followUp message after _processMessage() is done
|
||||
if typeof(processingTask) == Task && istaskdone(processingTask) == true
|
||||
agent.eventSink("_agentLoop 2 agent._state.messages length $(length(agent._state.messages))"; publish=true)
|
||||
agent.eventSink("_processMessage() done")
|
||||
# if agent runs is done but followUpChannel has messages,
|
||||
# put new message in inputChannel instead
|
||||
if isready(agent.followUpChannel)
|
||||
agent.eventSink("_agentLoop 3 agent._state.messages length $(length(agent._state.messages))"; publish=true)
|
||||
while isready(agent.followUpChannel)
|
||||
agent.eventSink("put followUp message to input channel")
|
||||
followUpMsg = take!(agent.followUpChannel)
|
||||
put!(agent.inputChannel, followUpMsg)
|
||||
end
|
||||
processingTask = nothing # reset
|
||||
result = nothing # reset
|
||||
else # _processMessage() done and no followUp message.
|
||||
agent.eventSink("_agentLoop 4 agent._state.messages length $(length(agent._state.messages))"; publish=true)
|
||||
agent.eventSink("no followUp message")
|
||||
result = deepcopy(agent._state.messages[end])
|
||||
agent.eventSink("_agentLoop 4-1 "; publish=true)
|
||||
# filter out reasoningContent in-place
|
||||
filter!(c -> !(c isa reasoningContent), result.content)
|
||||
agent.eventSink("_agentLoop 5 "; publish=true)
|
||||
# format output
|
||||
respondToUI = _assistantMessageToOpenAI(result)
|
||||
agent.eventSink("_agentLoop 6 "; publish=true)
|
||||
put!(agent.outputChannel, respondToUI)
|
||||
if !isempty(result.content) && result.content[1] isa textContent
|
||||
agent.eventSink(result.content[1].text; publish=true)
|
||||
end
|
||||
agent.eventSink("put result to output channel")
|
||||
processingTask = nothing # reset
|
||||
result = nothing # reset
|
||||
end
|
||||
@@ -325,7 +319,7 @@ function _agentLoop(agent::yiemAgent)
|
||||
else
|
||||
# spawn new _processMessage() if it is not already running.
|
||||
if processingTask === nothing
|
||||
agent.eventSink("_agentLoop 5 agent._state.messages length $(length(agent._state.messages))"; publish=true)
|
||||
agent.eventSink("spawn new _processMessage()")
|
||||
# discard all messages in followUpChannel
|
||||
while isready(agent.followUpChannel)
|
||||
_ = take!(agent.followUpChannel)
|
||||
@@ -347,8 +341,10 @@ function _agentLoop(agent::yiemAgent)
|
||||
)
|
||||
end
|
||||
put!(processMessageInputCh, newUserMsg)
|
||||
agent.eventSink("put new user msg to processMessageInputCh")
|
||||
newUserMsg = nothing # reset
|
||||
end
|
||||
|
||||
end
|
||||
catch e
|
||||
bt = catch_backtrace()
|
||||
@@ -356,7 +352,7 @@ function _agentLoop(agent::yiemAgent)
|
||||
showerror(io, e, bt)
|
||||
println(io)
|
||||
end
|
||||
eventSink(errMsg; publish=true)
|
||||
agent.eventSink(errMsg; publish=true)
|
||||
|
||||
# On any error, send error response and exit the loop
|
||||
@error "Agent loop failed" error=e
|
||||
@@ -404,10 +400,11 @@ function _processMessage(
|
||||
afterToolCall::Union{Function, Nothing},
|
||||
parallelToolExecute::Bool,
|
||||
)::Nothing
|
||||
eventSink("_processMessage 1 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
# loop until llmCall() response didn't use tool calls
|
||||
final_response = nothing
|
||||
|
||||
eventSink("▶ _processMessage: entered")
|
||||
|
||||
""" example message in inputChannel
|
||||
Dict(
|
||||
"role" => "user",
|
||||
@@ -425,48 +422,39 @@ function _processMessage(
|
||||
try
|
||||
# Drain inputChannel and convert OpenAI-format messages to userMessage type
|
||||
while isready(inputChannel)
|
||||
eventSink("_processMessage 2 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
newUserMsg_openai = take!(inputChannel)
|
||||
eventSink("_processMessage 3 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
if newUserMsg_openai === :shutdown
|
||||
eventSink("_processMessage 4 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
# Re-emit shutdown signal for the loop to handle
|
||||
put!(inputChannel, :shutdown)
|
||||
break
|
||||
end
|
||||
eventSink("_processMessage 5 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
newUserMsg = OpenAiToUserMessage(newUserMsg_openai)
|
||||
push!(agentMsgHistory, newUserMsg)
|
||||
eventSink("_processMessage 6 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
eventSink("▶ _processMessage: user msg pushed")
|
||||
end
|
||||
eventSink("_processMessage 7 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
# call prepareContext()
|
||||
state = agentState(systemPrompt, nothing, tools, agentMsgHistory)
|
||||
eventSink("_processMessage 8 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
preparedContext = prepareContext(state, eventSink)
|
||||
eventSink("_processMessage 9 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
# Call formatMessagesForLLM() to format for LLM
|
||||
formattedMessages = formatMessagesForLLM(preparedContext, eventSink)
|
||||
|
||||
# eventSink("_processMessage 10 formattedMessages $formattedMessages"; publish=true)
|
||||
|
||||
eventSink("▶ _processMessage: context")
|
||||
eventSink("llmCall()")
|
||||
""" response example
|
||||
response = JSON.Object{String, Any}("finish_reason" => "tool_calls", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "", "reasoning_content" => "Here's a thinking process:\n\n1. **Identify User Request**: The user is asking for the weather in Bangkok.\n2. **Locate Relevant Tool**: I have a `getWeather` function available.\n3. **Check Function Parameters**:\n - `city` (required): City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\n - `units` (optional, default \"celsius\"): Temperature scale (\"celsius\" or \"fahrenheit\")\n4. **Prepare Parameters**:\n - `city`: \"Bangkok, Thailand\" (adding country for clarity, though just \"Bangkok\" might work, following the example format is safer)\n - `units`: Not specified, so I'll use the default (\"celsius\")\n5. **Execute Tool Call**: Call `getWeather` with `city: \"Bangkok, Thailand\"`\n6. **Anticipate Response**: The function will return current weather and forecast data for Bangkok. I'll then format it nicely for the user.\n - *Self-Correction/Verification during thought*: The prompt says \"city: City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\". I'll use \"Bangkok, Thailand\". The `units` parameter is optional, so I'll omit it to use the default.\n - Proceed with tool call.✅\n", "tool_calls" => Any[JSON.Object{String, Any}("type" => "function", "function" => JSON.Object{String, Any}("name" => "getWeather", "arguments" => "{\"city\":\"Bangkok, Thailand\"}"), "id" => "6fOilR5QPcdppbAAHluhkRyUDu3oWMAL")]))
|
||||
"""
|
||||
response = llmCall(formattedMessages, eventSink=eventSink)
|
||||
# eventSink("LLM response " * string(response); publish=true)
|
||||
|
||||
eventSink("_processMessage 11 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
# Extract tool calls from LLM response content blocks
|
||||
hasToolCalls, toolCallList, assistant_msg = _extractToolCalls(response)
|
||||
eventSink("hasToolCalls: $hasToolCalls\ntoolCallList: $toolCallList _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
eventSink("assistant_msg " * string(assistant_msg); publish=true)
|
||||
eventSink("_processMessage 11-1"; publish=true)
|
||||
hasToolCalls, toolCallList, assistant_msg = _extractToolCalls(response, eventSink)
|
||||
|
||||
# Add assistant message (tool calls or text) to history for next LLM turn
|
||||
push!(agentMsgHistory, assistant_msg)
|
||||
eventSink("▶ _processMessage: LLM response")
|
||||
|
||||
if hasToolCalls && length(toolCallList) > 0
|
||||
eventSink("▶ _processMessage: tool calls")
|
||||
|
||||
# Build context and config for executeToolCalls
|
||||
|
||||
config = agentLoopConfig(
|
||||
@@ -476,21 +464,16 @@ function _processMessage(
|
||||
)
|
||||
|
||||
signal = abortSignal(false)
|
||||
eventSink("_processMessage 12 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
|
||||
# call executeToolCalls()
|
||||
toolResultBatch = executeToolCalls(preparedContext, assistant_msg, toolCallList, config,
|
||||
signal, eventSink)
|
||||
eventSink("_processMessage 13 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
|
||||
eventSink("▶ _processMessage: tool batch")
|
||||
# save toolResults to messages
|
||||
for toolResult in toolResultBatch.messages
|
||||
# eventSink("toolResult " * string(toolResult); publish=true)
|
||||
push!(agentMsgHistory, toolResult)
|
||||
end
|
||||
eventSink("_processMessage 14 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
if toolResultBatch.terminate
|
||||
eventSink("_processMessage 15 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
eventSink("▶ _processMessage: TERMINATE branch")
|
||||
# If toolResultBatch requested termination, build a final response
|
||||
final_content = [textContent("Tool execution completed.")]
|
||||
for toolResult in toolResultBatch.messages
|
||||
@@ -504,7 +487,6 @@ function _processMessage(
|
||||
end
|
||||
end
|
||||
end
|
||||
eventSink("_processMessage 16 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
final_response = assistantMessage(
|
||||
role="assistant",
|
||||
content=final_content,
|
||||
@@ -523,7 +505,7 @@ function _processMessage(
|
||||
break
|
||||
end
|
||||
else
|
||||
eventSink("_processMessage 17 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
eventSink("▶ _processMessage: FINAL response")
|
||||
# LLM did not use tool calls —
|
||||
break
|
||||
end
|
||||
@@ -535,11 +517,11 @@ function _processMessage(
|
||||
end
|
||||
|
||||
eventSink(err_msg; publish=true)
|
||||
eventSink("▶ _processMessage: CAUGHT ERROR")
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
eventSink("_processMessage 18 _state.messages length $(length(agentMsgHistory))"; publish=true)
|
||||
eventSink("▶ _processMessage: exited")
|
||||
return nothing
|
||||
end
|
||||
|
||||
@@ -649,9 +631,12 @@ The message is constructed from:
|
||||
1) llm_useTool = JSON.Object{String, Any}("finish_reason" => "tool_calls", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "", "reasoning_content" => "Here's a thinking process:\n\n1. **Identify User Request**: The user is asking for the weather in Bangkok.\n2. **Locate Relevant Tool**: I have a `getWeather` function available.\n3. **Check Function Parameters**:\n - `city` (required): City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\n - `units` (optional, default \"celsius\"): Temperature scale (\"celsius\" or \"fahrenheit\")\n4. **Prepare Parameters**:\n - `city`: \"Bangkok, Thailand\" (adding country for clarity, though just \"Bangkok\" might work, following the example format is safer)\n - `units`: Not specified, so I'll use the default (\"celsius\")\n5. **Execute Tool Call**: Call `getWeather` with `city: \"Bangkok, Thailand\"`\n6. **Anticipate Response**: The function will return current weather and forecast data for Bangkok. I'll then format it nicely for the user.\n - *Self-Correction/Verification during thought*: The prompt says \"city: City and country, e.g., 'San Francisco, CA' or 'Tokyo, Japan'\". I'll use \"Bangkok, Thailand\". The `units` parameter is optional, so I'll omit it to use the default.\n - Proceed with tool call.✅\n", "tool_calls" => Any[JSON.Object{String, Any}("type" => "function", "function" => JSON.Object{String, Any}("name" => "getWeather", "arguments" => "{\"city\":\"Bangkok, Thailand\"}"), "id" => "6fOilR5QPcdppbAAHluhkRyUDu3oWMAL")]))
|
||||
2) llm_notUseTool = JSON.Object{String, Any}("finish_reason" => "stop", "index" => 0, "message" => JSON.Object{String, Any}("role" => "assistant", "content" => "The weather in London, UK is currently Sunny with a temperature of 22°C.", "reasoning_content" => "The user asked for the weather in London.\nI called the `getWeather` tool for London, UK.\nThe response indicates it's Sunny and 22°C.\nI will convey this information to the user.\n"))
|
||||
"""
|
||||
function _extractToolCalls(response)
|
||||
function _extractToolCalls(response, eventSink=nothing)
|
||||
hasToolCalls = false
|
||||
toolCallList = agentToolCall[]
|
||||
if eventSink !== nothing
|
||||
eventSink("▶ _extractToolCalls: parsing response")
|
||||
end
|
||||
|
||||
# Helper: parse args (JSON string -> Dict, or pass through)
|
||||
parse_args(raw) = raw isa AbstractDict && !(raw isa Dict{String,Any}) ?
|
||||
@@ -979,47 +964,40 @@ function prepareToolCall(
|
||||
signal::abortSignal,
|
||||
eventSink
|
||||
)::Union{preparedToolCall,immediateOutcome}
|
||||
eventSink("prepareToolCall 1"; publish=true)
|
||||
|
||||
# pick a called tool from tool store
|
||||
eventSink("▶ prepareToolCall: resolving")
|
||||
tool = get(context.tools, toolCall.name, nothing)
|
||||
|
||||
if tool === nothing
|
||||
eventSink("prepareToolCall 2"; publish=true)
|
||||
eventSink("▶ prepareToolCall: NOT FOUND")
|
||||
return immediateOutcome(createErrorToolResult("Tool $toolCall.name not found"), true)
|
||||
end
|
||||
|
||||
try
|
||||
eventSink("prepareToolCall 3"; publish=true)
|
||||
# 1. prepare arguments (tool-specific transform)
|
||||
prepared = prepareToolCallArguments(tool, toolCall)
|
||||
eventSink("prepared " * string(prepared.arguments); publish=true)
|
||||
eventSink("prepareToolCall 4"; publish=true)
|
||||
validatedArgs = validateToolArguments(tool, prepared)
|
||||
eventSink("validatedArgs " * string(validatedArgs); publish=true)
|
||||
eventSink("prepareToolCall 5"; publish=true)
|
||||
eventSink("▶ prepareToolCall: validated")
|
||||
# 2. beforeToolCall hook — can block
|
||||
if config.beforeToolCall !== nothing
|
||||
eventSink("prepareToolCall 6"; publish=true)
|
||||
|
||||
before = config.beforeToolCall(
|
||||
beforeToolCallContext(assistantMsg, toolCall, validatedArgs, context),
|
||||
signal,
|
||||
eventSink
|
||||
)
|
||||
eventSink("prepareToolCall 7"; publish=true)
|
||||
|
||||
if signal.aborted
|
||||
eventSink("prepareToolCall 8"; publish=true)
|
||||
return immediateOutcome(createErrorToolResult("Operation aborted"), true)
|
||||
end
|
||||
|
||||
if before !== nothing && before.block
|
||||
eventSink("prepareToolCall 9"; publish=true)
|
||||
return immediateOutcome(
|
||||
createErrorToolResult(get(before, :reason, "Tool execution was blocked")), true)
|
||||
end
|
||||
end
|
||||
eventSink("prepareToolCall 10"; publish=true)
|
||||
|
||||
return preparedToolCall(tool, toolCall, validatedArgs)
|
||||
catch e
|
||||
bt = catch_backtrace()
|
||||
@@ -1029,7 +1007,6 @@ function prepareToolCall(
|
||||
end
|
||||
|
||||
eventSink(errMsg; publish=true)
|
||||
|
||||
return immediateOutcome(createErrorToolResult(sprint(showerror, e)), true)
|
||||
end
|
||||
end
|
||||
@@ -1080,11 +1057,11 @@ function executePreparedToolCall(
|
||||
signal::Union{Nothing,abortSignal},
|
||||
eventSink,
|
||||
)::executedOutcome
|
||||
eventSink("executePreparedToolCall 1"; publish=true)
|
||||
|
||||
eventSink("▶ execute: START")
|
||||
try
|
||||
result = prep.tool.execute(prep.toolCall.id, prep.args, signal, eventSink)
|
||||
# eventSink("tool execute raw result: " * sprint(show, result); publish=true)
|
||||
eventSink("▶ execute: END")
|
||||
return executedOutcome(result, false)
|
||||
catch e
|
||||
bt = catch_backtrace()
|
||||
@@ -1093,6 +1070,7 @@ function executePreparedToolCall(
|
||||
println(io)
|
||||
end
|
||||
eventSink(errMsg; publish=true)
|
||||
eventSink("▶ execute: ERROR")
|
||||
|
||||
return executedOutcome(createErrorToolResult(sprint(showerror, e)), true)
|
||||
end
|
||||
@@ -1165,10 +1143,10 @@ function finalizeExecutedToolCall(
|
||||
signal::Union{Nothing,abortSignal},
|
||||
eventSink
|
||||
)::finalizedOutcome
|
||||
eventSink("finalizeExecutedToolCall 1"; publish=true)
|
||||
|
||||
eventSink("▶ finalize: START")
|
||||
result = executed.result
|
||||
isError = executed.isError
|
||||
# eventSink("finalizeExecutedToolCall result before afterToolCall\n$result"; publish=true)
|
||||
if config.afterToolCall !== nothing
|
||||
try
|
||||
after = config.afterToolCall(
|
||||
@@ -1176,7 +1154,6 @@ function finalizeExecutedToolCall(
|
||||
signal,
|
||||
eventSink
|
||||
)
|
||||
eventSink("finalizeExecutedToolCall 3"; publish=true)
|
||||
if after !== nothing
|
||||
result = after
|
||||
isError = false
|
||||
@@ -1193,7 +1170,7 @@ function finalizeExecutedToolCall(
|
||||
isError = true
|
||||
end
|
||||
end
|
||||
eventSink("finalizeExecutedToolCall 4"; publish=true)
|
||||
eventSink("▶ finalize: END")
|
||||
return finalizedOutcome(prep.toolCall, result, isError)
|
||||
end
|
||||
|
||||
@@ -1259,39 +1236,27 @@ function executeToolCallsSequential(
|
||||
signal::abortSignal,
|
||||
eventSink,
|
||||
)::agentToolCallBatch
|
||||
eventSink("executeToolCallsSequential 1"; publish=true)
|
||||
finalizedCalls = finalizedOutcome[]
|
||||
messages = toolResultMessage[]
|
||||
eventSink("▶ executeToolCallsSequential: batch start")
|
||||
|
||||
for tc in toolCalls
|
||||
eventSink("start tool execute $(tc.id), $(tc.name), $(tc.arguments)"; publish=true)
|
||||
prep = prepareToolCall(context, assistantMsg, tc, config, signal, eventSink)
|
||||
eventSink("executeToolCallsSequential " * string(prep.args); publish=true)
|
||||
|
||||
if prep isa immediateOutcome
|
||||
eventSink("executeToolCallsSequential 2-1"; publish=true)
|
||||
finalized = finalizedOutcome(tc, prep.result, prep.isError)
|
||||
eventSink("executeToolCallsSequential 2-2"; publish=true)
|
||||
else
|
||||
eventSink("executeToolCallsSequential 3"; publish=true)
|
||||
executed = executePreparedToolCall(prep, signal, eventSink)
|
||||
|
||||
eventSink("executeToolCallsSequential 3-1"; publish=true)
|
||||
finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config,
|
||||
signal, eventSink)
|
||||
eventSink("executeToolCallsSequential 3-2"; publish=true)
|
||||
end
|
||||
eventSink("executeToolCallsSequential 4"; publish=true)
|
||||
# eventSink("finalized $(finalized.toolCall.id), $(finalized.toolCall.name),
|
||||
# $(finalized.result), $(finalized.isError)"; publish=true)
|
||||
push!(messages, createToolResultMessage(finalized))
|
||||
push!(finalizedCalls, finalized)
|
||||
eventSink("executeToolCallsSequential 5"; publish=true)
|
||||
if signal !== nothing && signal.aborted
|
||||
break
|
||||
end
|
||||
end
|
||||
eventSink("executeToolCallsSequential 6"; publish=true)
|
||||
eventSink("▶ executeToolCallsSequential: batch end")
|
||||
return agentToolCallBatch(messages, shouldTerminate(finalizedCalls))
|
||||
end
|
||||
|
||||
@@ -1362,23 +1327,24 @@ function executeToolCallsParallel(
|
||||
)::agentToolCallBatch
|
||||
|
||||
entries = Union{finalizedOutcome,Task}[]
|
||||
eventSink("▶ executeToolCallsParallel: batch start")
|
||||
|
||||
for tc in toolCalls
|
||||
eventSink(toolExecStartEvent(tc.id, tc.name, tc.arguments); publish=true)
|
||||
toolExecStartEvent(tc.id, tc.name, tc.arguments)
|
||||
|
||||
prep = prepareToolCall(context, assistantMsg, tc, config, signal, eventSink)
|
||||
|
||||
if prep isa immediateOutcome
|
||||
finalized = finalizedOutcome(tc, prep.result, prep.isError)
|
||||
eventSink(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name,
|
||||
finalized.result, finalized.isError))
|
||||
toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name,
|
||||
finalized.result, finalized.isError)
|
||||
push!(entries, finalized)
|
||||
else
|
||||
t = Task() do
|
||||
executed = executePreparedToolCall(prep, signal, eventSink)
|
||||
finalized = finalizeExecutedToolCall(context, assistantMsg, prep, executed, config, signal)
|
||||
eventSink(toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name,
|
||||
finalized.result, finalized.isError))
|
||||
toolExecEndEvent(finalized.toolCall.id, finalized.toolCall.name,
|
||||
finalized.result, finalized.isError)
|
||||
return finalized
|
||||
end
|
||||
schedule(t)
|
||||
@@ -1401,6 +1367,7 @@ function executeToolCallsParallel(
|
||||
push!(messages, createToolResultMessage(f))
|
||||
end
|
||||
|
||||
eventSink("▶ executeToolCallsParallel: batch end")
|
||||
return agentToolCallBatch(messages, shouldTerminate(finalizedCalls))
|
||||
end
|
||||
|
||||
@@ -1463,7 +1430,6 @@ function executeToolCalls(
|
||||
eventSink,
|
||||
)::agentToolCallBatch
|
||||
|
||||
eventSink("_executeToolCalls 1"; publish=true)
|
||||
hasSequential = false
|
||||
for tc in toolCalls
|
||||
t = get(context.tools, tc.name, nothing)
|
||||
@@ -1472,13 +1438,11 @@ function executeToolCalls(
|
||||
break
|
||||
end
|
||||
end
|
||||
eventSink("_executeToolCalls 2"; publish=true)
|
||||
eventSink("▶ executeToolCalls: dispatch")
|
||||
if config.toolExecution == "sequential" || hasSequential
|
||||
eventSink("_executeToolCalls 3"; publish=true)
|
||||
return executeToolCallsSequential(context, assistantMsg, toolCalls, config, signal,
|
||||
eventSink)
|
||||
else
|
||||
eventSink("_executeToolCalls 4"; publish=true)
|
||||
return executeToolCallsParallel(context, assistantMsg, toolCalls, config, signal,
|
||||
eventSink)
|
||||
end
|
||||
@@ -1519,8 +1483,11 @@ OpenAiToUserMessage(msg)
|
||||
# => userMessage("user", [textContent("Hello")], DateTime(...))
|
||||
```
|
||||
"""
|
||||
function OpenAiToUserMessage(msg::AbstractDict)::userMessage
|
||||
function OpenAiToUserMessage(msg::AbstractDict, eventSink=nothing)::userMessage
|
||||
content_blocks = Vector{messageContent}()
|
||||
if eventSink !== nothing
|
||||
eventSink("▶ OpenAiToUserMessage: parsing")
|
||||
end
|
||||
|
||||
raw_content = get(msg, "content", Any[])
|
||||
if raw_content isa Vector
|
||||
@@ -1544,6 +1511,9 @@ function OpenAiToUserMessage(msg::AbstractDict)::userMessage
|
||||
end
|
||||
end
|
||||
|
||||
if eventSink !== nothing
|
||||
eventSink("▶ OpenAiToUserMessage: done")
|
||||
end
|
||||
return userMessage(content=content_blocks)
|
||||
end
|
||||
|
||||
|
||||
+3
-4
@@ -121,9 +121,7 @@ function register_mcp_tools(tools::OrderedDict{String, agentTool}, mcpserver; ev
|
||||
end
|
||||
new_count = 0
|
||||
try
|
||||
eventSink("register_mcp_tools 1")
|
||||
response = mcpserver("tools/list", sessionId; eventSink=eventSink)
|
||||
eventSink("register_mcp_tools 2")
|
||||
# Parse JSON-RPC 2.0 response envelope
|
||||
if haskey(response, "error")
|
||||
rpc_error = response["error"]
|
||||
@@ -200,7 +198,7 @@ Add `tool` to `tools`, overwriting any existing tool with the same name.
|
||||
function registerTool(tools::OrderedDict{String, agentTool}, tool::agentTool; eventSink=nothing
|
||||
)::OrderedDict{String, agentTool}
|
||||
tools[tool.name] = tool
|
||||
println("[toolRegistry] Registered tool: $(tool.name)")
|
||||
eventSink("[toolRegistry] Registered tool: $(tool.name)")
|
||||
return tools
|
||||
end
|
||||
|
||||
@@ -247,7 +245,6 @@ function listTool(tools::OrderedDict{String, agentTool}, mcpserver; eventSink=no
|
||||
eventSink) -> begin
|
||||
# Discover and register MCP tools (idempotent — skips already registered)
|
||||
new_count = register_mcp_tools(tools, mcpserver; eventSink=eventSink, sessionId=sessionId)
|
||||
eventSink("tools dump: " * sprint(show, tools))
|
||||
|
||||
# Always include listTools itself in the count
|
||||
total = length(tools)
|
||||
@@ -285,7 +282,9 @@ Register all tools for an agent:
|
||||
"""
|
||||
function registerAllTools(tools::OrderedDict{String, agentTool}, mcpserver=nothing; eventSink=nothing, sessionId=nothing)
|
||||
list_t = listTool(tools, mcpserver; eventSink=eventSink, sessionId=sessionId)
|
||||
eventSink("registering tools")
|
||||
registerTool(tools, list_t; eventSink=eventSink)
|
||||
eventSink("registering tools done")
|
||||
end
|
||||
|
||||
end # module
|
||||
|
||||
+5
-10
@@ -208,7 +208,6 @@ function formatMsgForLLM(ctx::agentContext, eventSink)::Dict{String, Any}
|
||||
openaiReadyMsg = Dict{String, Any}()
|
||||
# openaiReadyMsg["model"] = "gemma-4-E4B-it-UD-Q4_K_XL"
|
||||
messages = Vector{Dict{String, Any}}()
|
||||
eventSink("formatMsgForLLM 1")
|
||||
# System prompt as system message
|
||||
if !isempty(ctx.systemPrompt)
|
||||
push!(messages, Dict(
|
||||
@@ -216,7 +215,6 @@ function formatMsgForLLM(ctx::agentContext, eventSink)::Dict{String, Any}
|
||||
"content" => [Dict("type" => "text", "text" => ctx.systemPrompt)]
|
||||
))
|
||||
end
|
||||
eventSink("formatMsgForLLM 2")
|
||||
# Conversation messages
|
||||
for msg in ctx.messages
|
||||
if msg isa userMessage
|
||||
@@ -229,10 +227,8 @@ function formatMsgForLLM(ctx::agentContext, eventSink)::Dict{String, Any}
|
||||
push!(messages, _toolResultMessageToOpenAI(msg))
|
||||
end
|
||||
end
|
||||
eventSink("formatMsgForLLM 3")
|
||||
# Convert ctx.tools into OpenAI tools format
|
||||
tools_array = _toolsToOpenAI(ctx.tools, eventSink)
|
||||
eventSink("formatMsgForLLM 4")
|
||||
openaiReadyMsg["messages"] = messages
|
||||
openaiReadyMsg["temperature"] = 0.7
|
||||
|
||||
@@ -278,7 +274,7 @@ end
|
||||
function beforeToolCall(context::beforeToolCallContext, signal::abortSignal, eventSink
|
||||
)::beforeToolCallResult
|
||||
|
||||
eventSink("beforeToolCall: $(context.toolCall.name) args=$(context.args)")
|
||||
# eventSink("beforeToolCall: $(context.toolCall.name) args=$(context.args)")
|
||||
|
||||
return beforeToolCallResult(false, "N/A")
|
||||
end
|
||||
@@ -306,11 +302,11 @@ Common use cases:
|
||||
- `agentToolResult(...)` to return a modified result (content, details, usage,
|
||||
terminate flag can all be overridden).
|
||||
|
||||
""" #WORKING arrange result from searchWine tool
|
||||
""" # arrange result from searchWine tool
|
||||
function afterToolCall(context::afterToolCallContext, signal::abortSignal, eventSink
|
||||
)::Union{agentToolResult, Nothing}
|
||||
|
||||
eventSink("afterToolCall: $(context.toolCall.name) result=$(context.result) error=$(context.isError)")
|
||||
# eventSink("afterToolCall: $(context.toolCall.name) result=$(context.result) error=$(context.isError)")
|
||||
|
||||
return nothing
|
||||
end
|
||||
@@ -428,8 +424,7 @@ _toolsToOpenAI(tools) # => [Dict("type" => "function", "function" => Dict("na
|
||||
"""
|
||||
function _toolsToOpenAI(tools::Union{OrderedDict{String, agentTool}, Nothing}, eventSink)::Vector{Dict{String, Any}}
|
||||
tools_array = Vector{Dict{String, Any}}()
|
||||
eventSink("_toolsToOpenAI 1")
|
||||
eventSink(string(typeof(tools)))
|
||||
|
||||
if tools !== nothing
|
||||
for (_, tool) in tools
|
||||
push!(tools_array, Dict(
|
||||
@@ -442,7 +437,7 @@ function _toolsToOpenAI(tools::Union{OrderedDict{String, agentTool}, Nothing}, e
|
||||
))
|
||||
end
|
||||
end
|
||||
eventSink("_toolsToOpenAI 2")
|
||||
|
||||
return tools_array
|
||||
end
|
||||
|
||||
|
||||
-113
@@ -1,113 +0,0 @@
|
||||
using Revise, JSON, Dates, UUIDs, PrettyPrinting, LibPQ, Base64, DataFrames, DataStructures, HTTP, Base64,
|
||||
NATS, Base.Threads
|
||||
using YiemAgent, GeneralUtils, msghandler
|
||||
|
||||
|
||||
""" Debug
|
||||
|
||||
using JSON, NATS, msghandler
|
||||
using NATS
|
||||
|
||||
conn = NATS.connect("nats.yiem.cc")
|
||||
|
||||
sub = NATS.subscribe(conn, "sommanion.debug") do msg
|
||||
payload = NATS.payload(msg)
|
||||
@info "debug" payload
|
||||
|
||||
open("./log/error.log", "a") do io
|
||||
println(io, payload)
|
||||
end
|
||||
end
|
||||
|
||||
NATS.publish(conn, "sommanion.debug", "order-123")
|
||||
|
||||
|
||||
# ---------------------------- inject this code into codebase to debug --------------------------- #
|
||||
try
|
||||
batch = someFunction(x, y, z)
|
||||
catch e
|
||||
bt = catch_backtrace()
|
||||
err_msg = sprint() do io
|
||||
showerror(io, e, bt)
|
||||
println(io)
|
||||
end
|
||||
|
||||
eventSink(err_msg)
|
||||
end
|
||||
|
||||
"""
|
||||
|
||||
struct text2textInstructLLM
|
||||
natsConn::NATS.Connection
|
||||
topic::String
|
||||
senderID::String
|
||||
fileserver_url::String
|
||||
end
|
||||
|
||||
function (t::text2textInstructLLM)(openai_msg::Dict{String, Any})
|
||||
|
||||
payloads = [("msg", openai_msg, "dictionary")] # List of tuples
|
||||
_, msg_envelope_json_str = msghandler.smartpack(
|
||||
t.topic,
|
||||
payloads;
|
||||
sender_id=t.senderID,
|
||||
msg_purpose="text2text",
|
||||
fileserver_url=t.fileserver_url)
|
||||
|
||||
reply = NATS.request(t.natsConn, t.topic, msg_envelope_json_str, timeout=180)
|
||||
|
||||
incoming_env_json_str = String(reply.payload)
|
||||
incoming_env = msghandler.smartunpack(incoming_env_json_str)
|
||||
_llm_response = incoming_env["payloads"][1][2]
|
||||
llm_response = _llm_response["choices"][1]
|
||||
|
||||
return llm_response
|
||||
end
|
||||
|
||||
|
||||
|
||||
struct eventSink
|
||||
natsConn::NATS.Connection
|
||||
topic::String
|
||||
senderID::String
|
||||
end
|
||||
|
||||
function (aes::eventSink)(msg::String)
|
||||
NATS.publish(aes.natsConn, aes.topic, msg)
|
||||
end
|
||||
|
||||
|
||||
|
||||
config = JSON.parsefile("./appconfig.json")
|
||||
agent_conn = NATS.connect(config["nats_server_info"]["url"])
|
||||
|
||||
|
||||
#WORKING load tools
|
||||
text2text_llm = text2textInstructLLM(agent_conn,
|
||||
config["externalservice"]["servicesloadbalancer"]["nats"],
|
||||
"sender",
|
||||
config["externalservice"]["fileserver"]["url"])
|
||||
|
||||
debugNats = eventSink(agent_conn, "sommanion.debug", "sender")
|
||||
|
||||
agent = YiemAgent.yiemAgent(
|
||||
text2text_llm;
|
||||
eventSink=debugNats
|
||||
)
|
||||
|
||||
msg = Dict(
|
||||
"role" => "user",
|
||||
"content" => [
|
||||
Dict("type" => "text", "text" => "What's the weather in Bangkok?"),
|
||||
# Dict(
|
||||
# "type" => "image_url",
|
||||
# "image_url" => Dict("url" => "data:mime_type;base64,image2_base64_string")
|
||||
# ),
|
||||
]
|
||||
)
|
||||
|
||||
push!(agent.inputChannel, msg)
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user