Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 35 additions & 29 deletions packages/core/src/agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -53,24 +53,24 @@ export class Agent {
readonly #model: ModelProvider;
readonly #session: SessionLog;
readonly #tools: ToolSet;
readonly #ids: AgentIdGenerator;
readonly #idGenerator: AgentIdGenerator;
readonly #maxSteps: number;
readonly #settlementSignals: SettlementSignalFactory;
readonly #settlementSignalFactory: SettlementSignalFactory;
readonly #clock: Clock | undefined;
readonly #prompts: PromptAssembler;
readonly #promptAssembler: PromptAssembler;
readonly #runTimeoutMs: number | undefined;
readonly #modelAttemptTimeoutMs: number;
readonly #toolTimeoutMs: number;
readonly #maxModelOutputBytes: number | undefined;
readonly #maxToolResultBytes: number;
readonly #maxQueuedRuns: number;
readonly #observers: readonly AgentObserver[];
readonly #reservations: AgentIdReservations; // 标识符预留池,保证同一个 Agent 里,turn / step / attempt / message / toolCall 的 ID 不重复
readonly #idReservations: AgentIdReservations;
readonly #inbox = new AgentInbox();
#phase: AgentPhase = { kind: 'idle' };
#controller: AbortController | undefined;
#activeRunAbortController: AbortController | undefined;
#accepting = true;
#draining: Promise<void> | undefined; // 正在 draining 的 promise,用于保证只有一个 drain 在执行
#inboxDrain: Promise<void> | undefined;

constructor(options: AgentOptions) {
// 参数校验
Expand All @@ -94,19 +94,19 @@ export class Agent {
this.#model = options.model;
this.#session = options.session;
this.#tools = options.tools;
this.#ids = options.ids;
this.#idGenerator = options.ids;
this.#maxSteps = options.maxSteps;
this.#settlementSignals = options.settlementSignals;
this.#settlementSignalFactory = options.settlementSignals;
this.#clock = options.clock;
this.#prompts = options.prompts;
this.#promptAssembler = options.prompts;
this.#runTimeoutMs = options.runTimeoutMs;
this.#modelAttemptTimeoutMs = options.modelAttemptTimeoutMs ?? DEFAULT_MODEL_ATTEMPT_TIMEOUT_MS;
this.#toolTimeoutMs = options.toolTimeoutMs ?? DEFAULT_TOOL_TIMEOUT_MS;
this.#maxModelOutputBytes = options.maxModelOutputBytes;
this.#maxToolResultBytes = options.maxToolResultBytes ?? DEFAULT_MAX_TOOL_RESULT_BYTES;
this.#maxQueuedRuns = options.maxQueuedRuns ?? DEFAULT_MAX_QUEUED_RUNS;
this.#observers = options.observers ?? [];
this.#reservations = new AgentIdReservations(options.reservedIds);
this.#idReservations = new AgentIdReservations(options.reservedIds);
}

get phase(): AgentPhase {
Expand All @@ -115,7 +115,7 @@ export class Agent {

/** 取消当前活动 run;不影响是否继续接纳新的 run。 */
cancel(reason: unknown = new Error('agent run cancelled')): void {
this.#controller?.abort(reason);
this.#activeRunAbortController?.abort(reason);
}

/** 永久停止接纳新 run,并取消当前活动。重复调用安全。 */
Expand All @@ -128,12 +128,12 @@ export class Agent {
/** 等待当前 drain 完成,其中包含活动 turn 与已接纳的排队输入。 */
async whenIdle(): Promise<void> {
for (;;) {
const draining = this.#draining;
if (draining === undefined) {
const inboxDrain = this.#inboxDrain;
if (inboxDrain === undefined) {
return;
}
await draining;
if (this.#draining === undefined) {
await inboxDrain;
if (this.#inboxDrain === undefined) {
return;
}
}
Expand Down Expand Up @@ -186,11 +186,11 @@ export class Agent {

/** 确保只有一个 FIFO drain;单个输入失败不会阻断后续输入。 */
#ensureInboxDraining(): void {
if (this.#draining !== undefined) {
if (this.#inboxDrain !== undefined) {
return;
}
this.#draining = this.#drainInbox().finally(() => {
this.#draining = undefined;
this.#inboxDrain = this.#drainInbox().finally(() => {
this.#inboxDrain = undefined;
if (this.#inbox.hasPending) {
this.#ensureInboxDraining();
}
Expand All @@ -200,7 +200,7 @@ export class Agent {
/** 串行领取已接纳输入,并仅结算各自调用,不让单项失败中断后续领取。 */
async #drainInbox(): Promise<void> {
for (;;) {
const queued = this.#inbox.claim();
const queued = this.#inbox.claim(); // 领取最早尚未执行的输入
if (queued === undefined) {
return;
}
Expand All @@ -225,12 +225,12 @@ export class Agent {
isAbortSignal(options) ? undefined : options.execution,
);
const turn: ActiveTurn = {
turnId: this.#reservations.turn(this.#ids.turn()),
turnId: this.#idReservations.turn(this.#idGenerator.turn()),
started: false,
ended: false,
};
const controller = new AbortController();
this.#controller = controller;
this.#activeRunAbortController = controller;
const timeoutSignal =
this.#runTimeoutMs === undefined ? undefined : AbortSignal.timeout(this.#runTimeoutMs);
const signal = AbortSignal.any([
Expand All @@ -255,15 +255,15 @@ export class Agent {
turn.started = true;

// 提交用户消息事件
this.#reservations.message(message.id);
this.#idReservations.message(message.id);
await this.#append([{ type: 'user.message', message }], signal);

// 循环执行 step,直至无工具调用或触及步数上限
for (let stepNumber = 1; stepNumber <= this.#maxSteps; stepNumber += 1) {
throwIfAborted(signal);

// 创建 step 生命周期
const stepId = this.#reservations.step(this.#ids.step());
const stepId = this.#idReservations.step(this.#idGenerator.step());
const step = createStep(stepId);
turn.activeStep = step;
this.#phase = { kind: 'running', turnId: turn.turnId, stepId };
Expand Down Expand Up @@ -300,15 +300,15 @@ export class Agent {
// 如果 terminal settlement 事件不为空,则提交结算事件
if (terminalEvents.length > 0) {
try {
const settlementSignal = this.#settlementSignals.create();
const settlementSignal = this.#settlementSignalFactory.create();
await this.#append(terminalEvents, settlementSignal);
} catch (persistenceError) {
throw new SettlementPersistenceError(error, persistenceError, outcome);
}
}
throw error;
} finally {
this.#controller = undefined;
this.#activeRunAbortController = undefined;
this.#phase = { kind: 'idle' };
}
}
Expand All @@ -332,7 +332,11 @@ export class Agent {

// 组装请求
const request = await waitForSignal(
this.#prompts.assemble(await this.#session.read({ signal }), this.#tools.list(), signal),
this.#promptAssembler.assemble(
await this.#session.read({ signal }),
this.#tools.list(),
signal,
),
signal,
);

Expand All @@ -341,7 +345,7 @@ export class Agent {
const prepared = await this.#model.prepare(modelSignal);
throwIfAborted(modelSignal);
const attempt: Attempt = {
attemptId: this.#reservations.attempt(this.#ids.attempt()),
attemptId: this.#idReservations.attempt(this.#idGenerator.attempt()),
partialText: '',
observedToolCalls: [],
settlementPersisted: false,
Expand All @@ -354,6 +358,8 @@ export class Agent {
let outputTokens = 0;
let usageSequence = 0;
const modelStartedAt = this.#now();

// 提交模型开始事件
if (modelStartedAt !== undefined) {
this.#notifyExecution({
type: 'model.started',
Expand Down Expand Up @@ -383,7 +389,7 @@ export class Agent {
case 'tool.call':
outputBytes += utf8ByteLength(JSON.stringify(event.call));
this.#assertOutputBudget(outputBytes);
this.#reservations.toolCall(event.call.id);
this.#idReservations.toolCall(event.call.id);
attempt.observedToolCalls.push(event.call);
break;

Expand Down Expand Up @@ -439,7 +445,7 @@ export class Agent {

// 构建 assistant 消息
const assistantMessage: AssistantMessage = {
id: this.#reservations.message(this.#ids.message()),
id: this.#idReservations.message(this.#idGenerator.message()),
role: 'assistant',
content: attempt.partialText,
toolCalls: attempt.observedToolCalls,
Expand Down
4 changes: 2 additions & 2 deletions packages/core/src/inbox.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@ import type { AgentRunOptions, AssistantMessage, UserMessage } from '@agent-harn
export interface QueuedRun {
readonly message: UserMessage;
readonly options: AbortSignal | AgentRunOptions;
readonly resolve: (message: AssistantMessage) => void;
readonly reject: (error: unknown) => void;
readonly resolve: (message: AssistantMessage) => void; // 执行成功时,调用 resolve 通知调用者
readonly reject: (error: unknown) => void; // 执行失败时,调用 reject 通知调用者
readonly clearAbortListener: () => void;
}

Expand Down
2 changes: 1 addition & 1 deletion packages/core/src/lifecycle.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import {

type RuntimeFailureOutcome = Extract<TurnOutcome, 'failed' | 'cancelled'>;

/** 跟踪单次模型 attempt 的流式累积、结算与工具结果,供失败路径补写事件。 */
/** 一次 LLM 调用 */
export interface Attempt {
readonly attemptId: AttemptId;
partialText: string; // 模型生成时的中间输出
Expand Down
Loading
Loading