agents/user-agent

Classes#UserAgent<I, O>#Agent is the base class for all agents. It provides a mechanism for defining input/output schemas and implementing processing logic, serving as the foundation of the entire agent system.
By extending the Agent class and implementing the process method, you can create custom agents with various capabilities:
Process structured input and output data
Validate data formats using schemas
Communicate between agents through contexts
Support streaming or non-streaming responses
Maintain memory of past interactions
Output in multiple formats (JSON/text)
Forward tasks to other agents
Example#Here's an example of how to create a custom agent:
class MyAgent extends Agent {
  process(input: Message): Message {
    console.log(input);

    return {
      text: "Hello, How can I assist you today?",
    };
  }
}

const agent = new MyAgent();

const result = await agent.invoke({ message: "hello" });

console.log(result); // { text: "Hello, How can I assist you today?" }

Extends#`Agent`<I, O>
Type Parameters#Type Parameter
Default type
Description
I extends `Message`
`Message`
The input message type the agent accepts
O extends `Message`
`Message`
The output message type the agent returns
Indexable#[key: symbol]: () => string | () => Promise<void>
Constructors#Constructor#new UserAgent<I, O>(options): UserAgent<I, O>
Parameters#Parameter
Type
options
UserAgentOptions<I, O>
Returns#UserAgent<I, O>
Overrides#`Agent`.`constructor`
Properties#tag#tag: string = "UserAgent"
Overrides#`Agent`.`tag`
context#context: Context
invoke()#invoke: {(input, options?): Promise<O>; (input, options): Promise<`AgentResponseStream`<O>>; (input, options?): Promise<`AgentResponse`<O>>; }
Invoke the agent with regular (non-streaming) response
Regular mode waits for the agent to complete processing and return the final result, suitable for scenarios where a complete result is needed at once.
Call Signature#(input, options?): Promise<O>
Parameters#Parameter
Type
input
I & `Message`
options?
Partial<`AgentInvokeOptions`<UserContext>> & { streaming?: false; }
Returns#Promise<O>
Call Signature#(input, options): Promise<`AgentResponseStream`<O>>
Invoke the agent with streaming response
Streaming responses allow the agent to return results incrementally, suitable for scenarios requiring real-time progress updates, such as chat bot typing effects.
Parameters#Parameter
Type
Description
input
I & `Message`
Input message to the agent
options
Partial<`AgentInvokeOptions`<UserContext>> & { streaming: true; }
Invocation options, must set streaming to true for this overload
Returns#Promise<`AgentResponseStream`<O>>
Streaming response object
Example#Here's an example of invoking an agent with streaming response:
const model = new OpenAIChatModel();

// AIGNE: Main execution engine of AIGNE Framework.
const aigne = new AIGNE({
  model,
});

// Create an Agent instance
const agent = AIAgent.from({
  name: "chat",
  description: "A chat agent",
  inputKey: "message",
});

// Invoke the agent with streaming enabled
const stream = await aigne.invoke(
  agent,
  { message: "hello" },
  { streaming: true },
);

const chunks: string[] = [];

// Read the stream using an async iterator
for await (const chunk of stream) {
  if (isAgentResponseDelta(chunk)) {
    const text = chunk.delta.text?.message;
    if (text) {
      chunks.push(text);
    }
  }
}

console.log(chunks); // Output: ["Hello", ",", " ", "How", " ", "can", " ", "I", " ", "assist", " ", "you", " ", "today", "?"]

Call Signature#(input, options?): Promise<`AgentResponse`<O>>
General signature for invoking the agent
Returns either streaming or regular response based on the streaming parameter in options
Parameters#Parameter
Type
Description
input
I & `Message`
Input message to the agent
options?
Partial<`AgentInvokeOptions`<UserContext>>
Invocation options
Returns#Promise<`AgentResponse`<O>>
Agent response (streaming or regular)
Param#Input message to the agent
Param#Invocation options, must set streaming to false or leave unset
Returns#Final JSON response
Example#Here's an example of invoking an agent with regular mode:
const model = new OpenAIChatModel();

// AIGNE: Main execution engine of AIGNE Framework.
const aigne = new AIGNE({
  model,
});

// Create an Agent instance
const agent = AIAgent.from({
  name: "chat",
  description: "A chat agent",
  inputKey: "message",
});

// Invoke the agent
const result = await aigne.invoke(agent, { message: "hello" });

console.log(result); // Output: { message: "Hello, How can I assist you today?" }

Overrides#`Agent`.`invoke`
publish()#publish: (topic, payload, options?) => void
Publish a message to a topic, the aigne will invoke the listeners of the topic
Parameters#Parameter
Type
Description
topic
string | string[]
topic name, or an array of topic names
payload
`Message` | Omit<MessagePayload, "context">
message to publish
options?
InvokeOptions<UserContext>
-
Returns#void
subscribe()#subscribe: {(topic, listener?): Promise<MessagePayload>; (topic, listener): Unsubscribe; (topic, listener?): Unsubscribe | Promise<MessagePayload>; (topic, listener?): Unsubscribe | Promise<MessagePayload>; }
Call Signature#(topic, listener?): Promise<MessagePayload>
Parameters#Parameter
Type
topic
string | string[]
listener?
undefined
Returns#Promise<MessagePayload>
Call Signature#(topic, listener): Unsubscribe
Parameters#Parameter
Type
topic
string | string[]
listener
MessageQueueListener
Returns#Unsubscribe
Call Signature#(topic, listener?): Unsubscribe | Promise<MessagePayload>
Parameters#Parameter
Type
topic
string | string[]
listener?
MessageQueueListener
Returns#Unsubscribe | Promise<MessagePayload>
Call Signature#(topic, listener?): Unsubscribe | Promise<MessagePayload>
Parameters#Parameter
Type
topic
string | string[]
listener?
MessageQueueListener
Returns#Unsubscribe | Promise<MessagePayload>
unsubscribe()#unsubscribe: (topic, listener) => void
Parameters#Parameter
Type
topic
string | string[]
listener
MessageQueueListener
Returns#void
Accessors#stream#Get Signature#get stream(): ReadableStream<MessagePayload & { topic: string; }>
Returns#ReadableStream<MessagePayload & { topic: string; }>
Methods#from()#static from<I, O>(options): UserAgent<I, O>
Type Parameters#Type Parameter
I extends `Message`
O extends `Message`
Parameters#Parameter
Type
options
UserAgentOptions<I, O>
Returns#UserAgent<I, O>
subscribeToTopics()#protected subscribeToTopics(context): void
Parameters#Parameter
Type
context
Pick<Context, "subscribe">
Returns#void
Overrides#`Agent`.`subscribeToTopics`
publishToTopics()#protected publishToTopics(output, options): Promise<void>
Parameters#Parameter
Type
output
O
options
`AgentInvokeOptions`
Returns#Promise<void>
Overrides#`Agent`.`publishToTopics`
process()#process(input, options): Promise<`AgentProcessResult`<O>>
Core processing method of the agent, must be implemented in subclasses
This is the main functionality implementation of the agent, processing input and generating output. Can return various types of results:
Regular object response
Streaming response
Async generator
Another agent instance (transfer agent)
Parameters#Parameter
Type
Description
input
I
Input message
options
`AgentInvokeOptions`
Options for agent invocation
Returns#Promise<`AgentProcessResult`<O>>
Processing result
Examples#Example of returning a direct object:
class DirectResponseAgent extends Agent {
  process(input: Message): Message {
    // Process input and return a direct object response
    return {
      text: `Hello, I received your message: ${JSON.stringify(input)}`,
      confidence: 0.95,
      timestamp: new Date().toISOString(),
    };
  }
}

const agent = new DirectResponseAgent();

const result = await agent.invoke({ message: "Hello" });

console.log(result); // { text: "Hello, I received your message: { message: 'Hello' }", confidence: 0.95, timestamp: "2023-10-01T12:00:00Z" }

Example of returning a streaming response:
class StreamResponseAgent extends Agent {
  process(_input: Message): AgentResponseStream<Message> {
    // Return a ReadableStream as a streaming response
    return new ReadableStream({
      start(controller) {
        controller.enqueue(textDelta({ text: "Hello" }));
        controller.enqueue(textDelta({ text: ", " }));
        controller.enqueue(textDelta({ text: "This" }));
        controller.enqueue(textDelta({ text: " is" }));
        controller.enqueue(textDelta({ text: "..." }));
        controller.close();
      },
    });
  }
}

const agent = new StreamResponseAgent();
const stream = await agent.invoke({ message: "Hello" }, { streaming: true });

let fullText = "";
for await (const chunk of stream) {
  if (isAgentResponseDelta(chunk)) {
    const text = chunk.delta.text?.text;
    if (text) fullText += text;
  }
}

console.log(fullText); // Output: "Hello, This is..."

Example of using an async generator:
class AsyncGeneratorAgent extends Agent {
  async *process(
    _input: Message,
    _options: AgentInvokeOptions,
  ): AgentProcessAsyncGenerator<Message> {
    // Use async generator to produce streaming results
    yield textDelta({ message: "This" });
    yield textDelta({ message: "," });
    yield textDelta({ message: " " });
    yield textDelta({ message: "This" });
    yield textDelta({ message: " " });
    yield textDelta({ message: "is" });
    yield textDelta({ message: "..." });

    // Optional return a JSON object at the end
    return { time: new Date().toISOString() };
  }
}

const agent = new AsyncGeneratorAgent();
const stream = await agent.invoke({ message: "Hello" }, { streaming: true });

const message: string[] = [];
let json: Message | undefined;

for await (const chunk of stream) {
  if (isAgentResponseDelta(chunk)) {
    const text = chunk.delta.text?.message;
    if (text) message.push(text);
    if (chunk.delta.json) json = chunk.delta.json;
  }
}

console.log(message); // Output: ["This", ",", " ", "This", " ", "is", "..."]
console.log(json); // Output: { time: "2023-10-01T12:00:00Z" }

Example of transfer to another agent:
class SpecialistAgent extends Agent {
  process(_input: Message): Message {
    return {
      response: "This is a specialist response",
      expertise: "technical",
    };
  }
}

class MainAgent extends Agent {
  process(_input: Message): Agent {
    // Create a specialized agent for handling technical issues
    return new SpecialistAgent();
  }
}

const aigne = new AIGNE({});
const mainAgent = new MainAgent();

const result = await aigne.invoke(mainAgent, { message: "technical question" });
console.log(result); // { response: "This is a specialist response", expertise: "technical" }

Overrides#`Agent`.`process`
checkAgentInvokesUsage()#protected checkAgentInvokesUsage(_options): void
Check agent invocation usage to prevent exceeding limits
If the context has a maximum invocation limit set, checks if the limit has been exceeded and increments the invocation counter
Parameters#Parameter
Type
Description
_options
`AgentInvokeOptions`
Invocation options containing context and limits
Returns#void
Throws#Error if maximum invocation limit is exceeded
Overrides#`Agent`.`checkAgentInvokesUsage`
Interfaces#UserAgentOptions<I, O>#Configuration options for an agent
Extends#`AgentOptions`<I, O>
Type Parameters#Type Parameter
Default type
Description
I extends `Message`
`Message`
The agent input message type
O extends `Message`
`Message`
The agent output message type
Properties#Property
Type
context
Context
process?
`FunctionAgentFn`<I, O>
activeAgent?
`Agent`<any, any>

Parameter	Type
`input`	`I` & `Message`
`options?`	`Partial`<`AgentInvokeOptions`<`UserContext`>> & { `streaming?`: `false`; }

Parameter	Type	Description
`topic`	`string` \| `string`[]	topic name, or an array of topic names
`payload`	`Message` \| `Omit`<`MessagePayload`, `"context"`>	message to publish
`options?`	`InvokeOptions`<`UserContext`>	-

Parameter	Type
`topic`	`string` \| `string`[]
`listener?`	`undefined`

Parameter	Type
`topic`	`string` \| `string`[]
`listener`	`MessageQueueListener`

Parameter	Type
`topic`	`string` \| `string`[]
`listener?`	`MessageQueueListener`

Type Parameter	Default type	Description
`I` extends `Message`	`Message`	The input message type the agent accepts
`O` extends `Message`	`Message`	The output message type the agent returns

Parameter	Type	Description
`input`	`I` & `Message`	Input message to the agent
`options`	`Partial`<`AgentInvokeOptions`<`UserContext`>> & { `streaming`: `true`; }	Invocation options, must set streaming to true for this overload

Parameter	Type	Description
`input`	`I`	Input message
`options`	`AgentInvokeOptions`	Options for agent invocation

Type Parameter	Default type	Description
`I` extends `Message`	`Message`	The agent input message type
`O` extends `Message`	`Message`	The agent output message type

Property	Type
`context`	`Context`
`process?`	`FunctionAgentFn`<`I`, `O`>
`activeAgent?`	`Agent`<`any`, `any`>

Getting Started

AIGNE CLI

AIGNE Observability

How to guides

Concepts

API Reference

Contributing

agents/user-agent

Classes#

UserAgent<I, O>#

Example#

Extends#

Type Parameters#

Indexable#

Constructors#

Constructor#

Parameters#

Returns#

Overrides#

Properties#

tag#

Overrides#

context#

invoke()#

Call Signature#

Parameters#

Returns#

Call Signature#

Parameters#

Returns#

Example#

Call Signature#

Parameters#

Returns#

Param#

Param#

Returns#

Example#

Overrides#

publish()#

Parameters#

Returns#

subscribe()#

Call Signature#

Parameters#

Returns#

Call Signature#

Parameters#

Returns#

Call Signature#

Parameters#

Returns#

Call Signature#

Parameters#

Returns#

unsubscribe()#

Parameters#

Returns#

Accessors#

stream#

Get Signature#

Returns#

Methods#

from()#

Type Parameters#

Parameters#

Returns#

subscribeToTopics()#

Parameters#

Returns#

Overrides#

publishToTopics()#

Parameters#

Returns#

Overrides#

process()#

Parameters#

Returns#

Examples#

Overrides#