feat: add agent config

plugin/langchain/app.ts

@@ -27,7 +27,7 @@ export default class ModuleLangChainHook implements IBoot {
   configWillLoad() {
     this.#app.eggObjectLifecycleUtil.registerLifecycle(this.#graphObjectHook);
     this.#app.eggObjectLifecycleUtil.registerLifecycle(this.#boundModelObjectHook);
-    this.#app.eggObjectFactory.registerEggObjectCreateMethod(CompiledStateGraphProto, CompiledStateGraphObject.createObject);
+    this.#app.eggObjectFactory.registerEggObjectCreateMethod(CompiledStateGraphProto, CompiledStateGraphObject.createObject(this.#app));
     this.#app.eggPrototypeLifecycleUtil.registerLifecycle(this.#graphPrototypeHook);
   }
 

plugin/langchain/lib/graph/CompiledStateGraphObject.ts

@@ -5,12 +5,18 @@ import {
   EggObject,
   EggObjectStatus,
 } from '@eggjs/tegg-runtime';
-import { EggObjectName, EggPrototypeName, Id, IdenticalUtil } from '@eggjs/tegg';
+import { EggObjectName, EggPrototypeName, Id, IdenticalUtil, ModuleConfig } from '@eggjs/tegg';
 import { CompiledStateGraphProto } from './CompiledStateGraphProto';
 import { EggPrototype } from '@eggjs/tegg-metadata';
 import { ChatCheckpointSaverInjectName, ChatCheckpointSaverQualifierAttribute, GRAPH_EDGE_METADATA, GRAPH_NODE_METADATA, GraphEdgeMetadata, GraphMetadata, GraphNodeMetadata, IGraph, IGraphEdge, IGraphNode, TeggToolNode } from '@eggjs/tegg-langchain-decorator';
 import { LangGraphTracer } from '../tracing/LangGraphTracer';
 import { BaseCheckpointSaver, CompiledStateGraph } from '@langchain/langgraph';
+import { Application, Context } from 'egg';
+import { ModuleConfigUtil, TimerUtil } from '@eggjs/tegg-common-util';
+import { LangChainConfigSchemaType } from 'typings';
+import pathToRegexp from 'path-to-regexp';
+import { AIMessage, HumanMessage, SystemMessage, ToolMessage } from '@langchain/core/messages';
+import { Readable, Transform } from 'node:stream';
 
 export class CompiledStateGraphObject implements EggObject {
   private status: EggObjectStatus = EggObjectStatus.PENDING;
@@ -19,17 +25,19 @@ export class CompiledStateGraphObject implements EggObject {
   readonly proto: CompiledStateGraphProto;
   readonly ctx: EggContext;
   readonly daoName: string;
-  private _obj: object;
+  _obj: object;
   readonly graphMetadata: GraphMetadata;
   readonly graphName: string;
+  readonly app: Application;
 
-  constructor(name: EggObjectName, proto: CompiledStateGraphProto) {
+  constructor(name: EggObjectName, proto: CompiledStateGraphProto, app: Application) {
     this.name = name;
     this.proto = proto;
     this.ctx = ContextHandler.getContext()!;
     this.id = IdenticalUtil.createObjectId(this.proto.id, this.ctx?.id);
     this.graphMetadata = proto.graphMetadata;
     this.graphName = proto.graphName;
+    this.app = app;
   }
 
   async init() {
@@ -47,6 +55,8 @@ export class CompiledStateGraphObject implements EggObject {
       return await originalInvoke.call(graph, input, config);
     };
 
+    await this.boundByConfig();
+
     this.status = EggObjectStatus.READY;
   }
 
@@ -76,6 +86,125 @@ export class CompiledStateGraphObject implements EggObject {
     return compileGraph;
   }
 
+  async boundByConfig() {
+    const config = ModuleConfigUtil.loadModuleConfigSync(this.proto.unitPath) as ModuleConfig | undefined;
+    const agents: LangChainConfigSchemaType['agents'] = config?.langchain?.agents ?? [];
+    const configName = `${this.graphName.slice(0, 1).toUpperCase()}${this.graphName.slice(1)}`;
+    if (Object.keys(agents).includes(configName)) {
+      const { path: methodRealPath, type, stream, timeout } = agents[configName];
+
+      if ((type ?? '').toLocaleLowerCase() === 'http') {
+        const router = this.app.router;
+        const regExp = pathToRegexp(methodRealPath!, {
+          sensitive: true,
+        });
+        const handler = this.createHttpHandler(stream, timeout);
+        Reflect.apply(router.post, router,
+          [ `${this.graphName}.Invoke`, methodRealPath, ...[], handler ]);
+        this.app.rootProtoManager.registerRootProto('AgentControllerInvoke', (ctx: Context) => {
+          if (regExp.test(ctx.path)) {
+            return this.proto;
+          }
+        }, '');
+      }
+    }
+  }
+
+  createHttpHandler(stream?: boolean, timeout?: number) {
+    const self = this;
+    return async function(ctx: Context) {
+      const invokeFunc = (self._obj as CompiledStateGraph<any, any>).invoke;
+      const streamFunc = (self._obj as CompiledStateGraph<any, any>).stream;
+      const args = ctx.request.body;
+      const genArgs = Object.entries(args).reduce((acc, [ key, value ]) => {
+        if (Array.isArray(value) && typeof value[0] === 'object') {
+          acc[key] = value.map(obj => {
+            switch (obj.role) {
+              case 'human':
+                return new HumanMessage(obj);
+              case 'ai':
+                return new AIMessage(obj);
+              case 'system':
+                return new SystemMessage(obj);
+              case 'tool':
+                return new ToolMessage(obj);
+              default:
+                throw new Error('unknown message type');
+            }
+          });
+        } else {
+          acc[key] = value;
+        }
+        return acc;
+      }, {});
+
+      const defaultConfig = {
+        configurable: {
+          thread_id: process.pid.toString(),
+        },
+      };
+
+      let body: unknown;
+      try {
+        body = await TimerUtil.timeout<unknown>(() => Reflect.apply(stream ? streamFunc : invokeFunc, (self._obj as CompiledStateGraph<any, any>), [ genArgs, defaultConfig ]), timeout);
+      } catch (e: any) {
+        if (e instanceof TimerUtil.TimeoutError) {
+          ctx.logger.error(`timeout after ${timeout}ms`);
+          ctx.throw(500, 'timeout');
+        }
+        throw e;
+      }
+
+      // https://github.com/koajs/koa/blob/master/lib/response.js#L88
+      // ctx.status is set
+      const explicitStatus = (ctx.response as any)._explicitStatus;
+
+      if (
+        // has body
+        body != null ||
+        // status is not set and has no body
+        // code should by 204
+        // https://github.com/koajs/koa/blob/master/lib/response.js#L140
+        !explicitStatus
+      ) {
+        ctx.body = body;
+        if (stream) {
+          ctx.set({
+            'content-type': 'text/event-stream',
+            'cache-control': 'no-cache',
+            'transfer-encoding': 'chunked',
+            'X-Accel-Buffering': 'no',
+          });
+          const transformStream = new Transform({
+            objectMode: true,
+            transform(chunk: any, _encoding: string, callback) {
+              try {
+                // 如果 chunk 是对象，转换为 JSON
+                let data: string;
+                if (typeof chunk === 'string') {
+                  data = chunk;
+                } else if (typeof chunk === 'object') {
+                  data = JSON.stringify(chunk);
+                } else {
+                  data = String(chunk);
+                }
+
+                // 格式化为 SSE 格式
+                const sseFormatted = `data: ${data}\n\n`;
+                callback(null, sseFormatted);
+              } catch (error) {
+                callback(error);
+              }
+            },
+          });
+          ctx.body = Readable.fromWeb(body as any, { objectMode: true }).pipe(transformStream);
+        } else {
+          ctx.body = body;
+        }
+      }
+    };
+  }
+
   async boundNodes(stateGraph: EggObject) {
     const graphObj = stateGraph.obj as IGraph<any, any>;
     const nodes = this.graphMetadata.nodes ?? [];
@@ -122,9 +251,11 @@ export class CompiledStateGraphObject implements EggObject {
     return this._obj;
   }
 
-  static async createObject(name: EggObjectName, proto: EggPrototype): Promise<CompiledStateGraphObject> {
-    const compiledStateGraphObject = new CompiledStateGraphObject(name, proto as CompiledStateGraphProto);
-    await compiledStateGraphObject.init();
-    return compiledStateGraphObject;
+  static createObject(app: Application) {
+    return async function(name: EggObjectName, proto: EggPrototype): Promise<CompiledStateGraphObject> {
+      const compiledStateGraphObject = new CompiledStateGraphObject(name, proto as CompiledStateGraphProto, app);
+      await compiledStateGraphObject.init();
+      return compiledStateGraphObject;
+    };
   }
 }

plugin/langchain/lib/graph/CompiledStateGraphProto.ts

@@ -24,13 +24,15 @@ export class CompiledStateGraphProto implements EggPrototype {
   readonly name: EggPrototypeName;
   readonly graphMetadata: GraphMetadata;
   readonly graphName: string;
+  readonly unitPath: string;
 
   constructor(loadUnit: LoadUnit, protoName: string, graphName: string, graphMetadata: GraphMetadata) {
     this.loadUnitId = loadUnit.id;
     this.qualifiers = [];
     this.name = protoName;
     this.graphMetadata = graphMetadata;
     this.graphName = graphName;
+    this.unitPath = loadUnit.unitPath;
     this.id = IdenticalUtil.createProtoId(loadUnit.id, protoName);
   }
 

plugin/langchain/package.json

@@ -52,7 +52,7 @@
     "typescript": true
   },
   "engines": {
-    "node": ">=18.0.0"
+    "node": ">=20.0.0"
   },
   "dependencies": {
     "@eggjs/egg-module-common": "^3.64.2",

plugin/langchain/test/fixtures/apps/langchain/app/modules/bar/module.yml

@@ -16,4 +16,11 @@ mcp:
       clientName: barSse
       version: 1.0.0
       transportType: SSE
-      type: http
+      type: http
+
+langchain:
+  agents:
+    FooGraph:
+      path: /graph/stream
+      type: http
+      stream: true

plugin/langchain/test/fixtures/apps/langchain/config/config.default.js

@@ -8,9 +8,6 @@ module.exports = function() {
         enable: false,
       },
     },
-    bodyParser: {
-      enable: false,
-    },
   };
   return config;
 };

plugin/langchain/test/llm.test.ts

@@ -76,5 +76,60 @@ describe('plugin/langchain/test/llm.test.ts', () => {
         .get('/llm/graph')
         .expect(200, { value: 'hello graph toolhello world' });
     });
+
+    it('should agent controller work', async () => {
+      const url = await app.httpRequest()
+        .post('/graph/stream').url;
+      const response = await fetch(url, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({ messages: [{ role: 'human', content: 'hello world' }] }),
+      });
+
+
+      if (!response.ok) {
+        throw new Error(`HTTP ${response.status}`);
+      }
+
+      if (!response.body) {
+        throw new Error('Response body is null');
+      }
+
+      const reader = response.body.getReader();
+      const decoder = new TextDecoder();
+      let buffer = '';
+
+      const messages: object[] = [];
+
+      try {
+        // eslint-disable-next-line no-constant-condition
+        while (true) {
+          const { done, value } = await reader.read();
+
+          if (done) break;
+
+          buffer += decoder.decode(value, { stream: true });
+          const lines = buffer.split('\n');
+          buffer = lines.pop() || '';
+
+          lines.forEach(line => {
+            if (line.startsWith('data: ')) {
+              const data = line.slice(6);
+              try {
+                const parsed = JSON.parse(data);
+                messages.push(parsed);
+              } catch (e) {
+                throw e;
+              }
+            }
+          });
+        }
+      } finally {
+        reader.releaseLock();
+      }
+      assert(messages.length === 3);
+    });
   }
 });

plugin/langchain/typings/index.d.ts

@@ -108,11 +108,34 @@ export const ChatModelConfigModuleConfigSchema = Type.Object({
   name: 'ChatModel',
 });
 
+
+export const LangChainConfigSchema = Type.Object({
+  agents: Type.Record(Type.String(), Type.Object({
+    path: Type.Optional(Type.String({
+      description: 'http path',
+    })),
+    stream: Type.Optional(Type.Boolean({
+      description: '是否流式返回',
+    })),
+    type: Type.Optional(Type.String({
+      description: 'Http',
+    })),
+    timeout: Type.Optional(Type.Number({
+      description: '接口超时时间',
+    })),
+  })),
+}, {
+  title: 'langchain 设置',
+  name: 'langchain',
+});
+
 export type ChatModelConfigModuleConfigType = Static<typeof ChatModelConfigModuleConfigSchema>;
+export type LangChainConfigSchemaType = Static<typeof LangChainConfigSchema>;
 
 declare module '@eggjs/tegg' {
   export type LangChainModuleConfig = {
     ChatModel?: ChatModelConfigModuleConfigType;
+    langchain?: LangChainConfigSchema;
   };
 
   export interface ModuleConfig extends LangChainModuleConfig {