PoC: CAP aligned API and data model

[BobdenOs]

API proposal

LLM

At the core of the Agent story is the LLM provider. The easier it is to interact with the LLM provider the simpler the rest of the implementation becomes. Therefor I focused on keeping the LLMService API as straight forward as it really is. We have to send an array of Messages to the LLM provider in a provider specific structure. Therefor the LLMService accepts an array of Messages. Additionally you might not really care about the details and just want to get a prompt answered. For that use case it is also possible to just send your prompt to the LLMService. Behind the scenes the prompt will be put into an object and wrapped into the array.

const llm = cds.connect.to('llm')

const response = await llm.send('Who are you ?')
const response = await llm.send({data: [{role:'user',type:'text',content:'who are you?'}]})

// Result streaming
const stream = await llm.send({data: [{role:'user',type:'text',content:'who are you?'}], iterator: true})
// Two way streaming
const stream = await llm.send({data: Readable.from([{role:'user',type:'text',content:'who are you?'}]), iterator: true})
for await(const message of stream) {
  process.stdout.write(message.content)
}

Agent

For the majority of the time people will only interact with @agent and be satisfied with only doing AGENT.md and SKILL.md setups, but for CAP itself and for more complex agent configurations it will be required to control in more detail what happens around the Agent. Therefor the AgentService can be extended and used as the basis of any CAP service to provide a fully customized Agent environment. Like defining their own true tools as all actions currently will be proxied through the default action tool. As it is becoming a standard to group tools.

export default class CustomAgentService extend AgentService {
  async init() {
    await super.init()
    this.on('custom_tool', async() => {...})
  }
}

The fully custom tools require only CAP knowledge as the CAP types are automatically translated to the proper json_schema or whatever the LLM provider requires for tool definitions their input and output types. Additionally the type conversion enforces the same requirements that cds.assert does by default as long as it is possible to express it in the LLM providers their tool schema definition.

service CustomAgentService {
  action custom_tool(arg1: String, arg2: Uint8 not null) returns String;
}

Data Model

As the LLM providers have a hard limitation to what can go to their endpoints we know that everything that we will ever be sending to their endpoints are arrays. So while the exact contents of the array will differ between the LLM providers it comes down to a simple construct Messages.

Additionally as all the Messages are inherently part of a Session or Conversation. There is no real need to have a secondary persistence to track what Sessions we have available to us. Therefor The Sessions entity was turned into a view.

entity Messages : cuid, managed {
  session  : UUID;
  sequence : Integer;
  prev     : Association to Messages; // first message of a forked session points here
  role     : RoleType;
  type     : MessageType;
  content  : LargeString;
  query    : Map;
}

view Sessions as
  select from Messages {
    key session         as ID,
        min(createdAt)  as createdAt,
        max(modifiedAt) as modifiedAt,
  }
  group by
    session;

Harness

Probably the most important API for customization is the actual harness. As this allows for clever tricks for truly engineering the context. This is often called workflows where the harness takes on certain responsibilities that the LLM would require extra round trips. Additionally the LLM has the chance that it will not use the tool correctly and derail in trying to debug how to call the tool.

Some examples of what the harness might take over:

• embedding search
• manipulate the prompt (using the llm service)
• extract information from the prompt (using the llm service or RegExp 🤷)
• simply query the tables for live data inside the context
• search through the Knowledge Graph for relevant context

Additionally it is possible to use the harness to create sessions programmatically. Which enables future features like even more sub agents. Handing over specific tools, skills or system prompts.

const agents = await cds.connect.to('agents')
const session = await agents.send('start', {
  ID,
  options: {
    system: '...',
    tools: {...},
    skills: [...],
  },
})

session.write(message) // Send and persist a message
for await(const message of session) { ... } // React to responses

The session is a Duplex stream, because the LLM providers are limited to having a linear array of context. All implementation on top are effectively just chats and as everyone knows that has written an single working chat program. They are all simply a by directional socket which is also just a Duplex stream.

As the session is just another stream. All the Message processing is done through a pipeline. Which allows all the complex logic to be split into their own individual functions. As an example lets implement a rate limit. That will prevent the LLM from sending more then 100 messages or call more then 10 tools.

async function* rate_limit(stream) {
  let messages = 0
  let tool_calls = 0
  for await (const message of stream) {
    if (message.type === 'tool_call') if(tool_calls++ > 10) return
    else if (messages++ > 100) return
    yield message
  }
}

Or if someone wants to disable reasoning from going further through the stream.

async function* internal_thinking(stream) {
  for await (const message of stream) {
    // prevent anyone else from seeing the reasoning tokens
    if (message.type === 'reasoning') continue
    yield message
  }
}

A2A

It is just a protocol adapter now that all the required functionalities have stable APIs.

Agent Flow

It is important to understand the primary difference between the LLMService and the AgentService is that the AgentSession persists the session context to the DatabaseService. Therefor it is required for every iteration to send the whole session context to the LLMService. To enable the CAP service to do this for many sessions in parallel the whole context is never loaded into memory. Instead the DatabaseService is queried to stream the Messages directly to the LLMService. This means that no matter how many sessions or messages there are in any given session the CAP service can process them. Additionally all the responses from the LLMService are streamed back into the DatabaseService to ensure that the LargeString contents of the responses never accumelate in the application memory. This also applies to all tool calls who respond with stream responses.

The following diagram describes the flow of sending a prompt to an AgentSession. Where the first response of the LLMService is a tool_call which will trigger another cycle in the AgentSession. For the LLMService to respond with the final result for the prompt.

sequenceDiagram
  User->>A2A: prompt
  A2A->>AgentService: prompt
  AgentService->>Session: prompt
  Session->>Database: store prompt
  Session->>LLMService: tools, system, agent, skills
  Session->>Database: query messages
  Database->>LLMService: messages
  LLMService-->>Session: tool_call, responses
  Session->>Database: store tool_call
  Session->>CAP: execute tool_call
  CAP-->>Session: return tool_result
  Session->>Database: store tool_result
  Session->>LLMService: tools, system, agent, skills
  Session->>Database: query messages
  Database->>LLMService: messages
  LLMService-->>Session: responses
  Session-->>A2A: responses
  A2A-->>User: responses

Loading

The Messages inside the AgentSession look as follows. Depending on the type of messages the AgentSession will either start / pause / resume. Every time the AgentSession either starts or resumes the whole session context is generated and directly streamed into the LLMService. Therefor as more Messages are stored into the DatabaseService the session context is automatically extended. Keeping the AgentSession implementation itself stateless. This also ensures that when an A2A task request is received by any application instance it is able to directly interact with the DatabaseService to either check / cancel / list the tasks. Where as a langchain or pi implementation would require the A2A request to hit the exact application instance that happens to be processing the target task.

[
  // source: AgentSession
  { role: 'systemm', type:'tool', content: '{tools, args: {types...}}'}, 
  { role: 'system', type: 'text', content: 'you are a agent...'},
  { role: 'system', type: 'text', content: 'you have the following skills: ...'},
  // source: User / Database (starts session)
  { role: 'user', type: 'text', content: 'the prompt...'},
  // source: LLMService / Database (pauses session)
  { role: 'assistant', type: 'tool_call', query: {tool: 'name', ID:'abc', args: {...}} },
  // source: CAP / Database (resumes session)
  { role: 'assistant', type: 'tool', content: 'result', query: {tool:'name', ID: 'abc'}},
  // source: LLMService / Database (pauses session)
  { role: 'assistant', type: 'text', content: 'reply'},
]