一小时基于pi-mono实现类似openclaw的agent系统，pi-agent上手实践（云端agent系列 二）

1
app.post('/api/chat', async (req, res) => {
2
  const { input, llmConfig, sandboxId, sandboxApiKey } = req.body;
3
  if (!input) {
4
    return res.status(400).json({ error: 'Input is required' });
5
  }
6

7
  // Handle setting API keys for PI runtime
8
  if (llmConfig) {
9
    if (llmConfig.type === 'anthropic') {
10
      process.env.ANTHROPIC_API_KEY = llmConfig.apiKey;
11
    } else {
12
      process.env.OPENAI_API_KEY = llmConfig.apiKey;
13
      if (llmConfig.baseUrl) process.env.OPENAI_BASE_URL = llmConfig.baseUrl;
14
    }
15
  }
16

17
  // Re-connect to the existing Sandbox
18
  let sandbox: Sandbox | null = null;
19
  if (sandboxId && sandboxApiKey) {
20
    try {
21
      sandbox = await Sandbox.connect(sandboxId, { apiKey: sandboxApiKey });
22
      setActiveSandbox(sandbox);
23
    } catch (e: any) {
24
      console.warn("Could not connect to sandbox:", e.message);
25
    }
26
  }
27

28
  try {
29
    const loader = new DefaultResourceLoader({
30
      cwd: __dirname,
31
      agentDir: getAgentDir(),
32
      // Use the extension factory to dynamically register our 'computer' tool
33
      extensionFactories: [registerComputerUseExtension]
34
    });
35
    await loader.reload();
36

37
    const { session } = await createAgentSession({
38
      resourceLoader: loader,
39
      sessionManager: SessionManager.inMemory()
40
    });
41

42
    // Start processing input
43
    console.log(`Processing user input through pi-agent: ${input}`);
44

45
    // Provide initial screenshot to agent if sandbox exists
46
    let initialPrompt: any = input;
47
    if (sandbox) {
48
      try {
49
        const initialScreenshot = await sandbox.screenshot();
50
        const initialScreenshotBase64 = Buffer.from(initialScreenshot).toString('base64');
51
        initialPrompt = [
52
          { type: 'text', text: input },
53
          { type: 'image_url', image_url: { url: `data:image/png;base64,${initialScreenshotBase64}` } }
54
        ];
55
      } catch (e: any) {
56
        console.warn("Failed to get initial screenshot:", e.message);
57
      }
58
    }
59

60
    // Subscribe to event stream and stream back as SSE
61
    res.setHeader('Content-Type', 'text/event-stream');
62
    res.setHeader('Cache-Control', 'no-cache');
63
    res.setHeader('Connection', 'keep-alive');
64

65
    const sendEvent = (event: string, data: any) => {
66
      res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
67
    };
68

69
    let responseText = '';
70

71
    session.subscribe((event) => {
72
      if (event.type === 'message_update') {
73
        const msgEvent = event.assistantMessageEvent;
74
        if (msgEvent.type === 'text_delta') {
75
          responseText += msgEvent.delta;
76
          sendEvent('status', { message: `pi-agent 正在思考... (${responseText.length} chars)` });
77
        } else if (msgEvent.type === 'tool_call') {
78
          sendEvent('action', { description: `pi-agent 正在调用工具: ${msgEvent.toolCall.function.name}` });
79
        }
80
      } else if (event.type === 'tool_result') {
81
        sendEvent('status', { message: `pi-agent 工具执行完毕: ${event.toolCall.function.name}` });
82
      }
83
    });
84

85
    try {
86
      await session.prompt(initialPrompt);
87
      sendEvent('done', { response: responseText, success: true });
88
      res.end();
89
    } catch (err: any) {
90
      console.error('Session prompt error:', err);
91
      sendEvent('error', { error: err.message, stack: err.stack });
92
      res.end();
93
    }
94
  } catch (error: any) {
95
    console.error('Agent setup error:', error);
96
    res.status(500).json({ error: error.message, stack: error.stack });
97
  }
98
});

1
import { Router } from 'express'
2
import { getSandbox } from '../services/e2b.js'
3
// Removed custom agentLoop from llm.ts since pi-agent handles it now
4

5
const router = Router()
6

7
router.post('/', async (req, res) => {
8
  const { sandboxId, instruction, llmConfig, sandboxApiKey } = req.body
9

10
  if (!sandboxId || !instruction || !llmConfig || !llmConfig.apiKey) {
11
    res.status(400).json({ success: false, error: 'Missing sandboxId, instruction, or llmConfig (apiKey required)' })
12
    return
13
  }
14

15
  const sandbox = getSandbox(sandboxId)
16
  if (!sandbox) {
17
    res.status(404).json({ success: false, error: 'Sandbox not found' })
18
    return
19
  }
20

21
  // Setup SSE
22
  res.setHeader('Content-Type', 'text/event-stream')
23
  res.setHeader('Cache-Control', 'no-cache')
24
  res.setHeader('Connection', 'keep-alive')
25

26
  const sendEvent = (event: string, data: any) => {
27
    res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
28
  }
29

30
  try {
31
    sendEvent('status', { message: '正在启动 pi-agent 进行自主屏幕操作...' })
32

33
    // Call pi-agent-service to process the instruction
34
    try {
35
      const piAgentRes = await fetch('http://localhost:4001/api/chat', {
36
        method: 'POST',
37
        headers: { 'Content-Type': 'application/json' },
38
        body: JSON.stringify({
39
          input: instruction,
40
          llmConfig,
41
          sandboxId,
42
          sandboxApiKey
43
        })
44
      });
45

46
      if (!piAgentRes.ok) {
47
        const errData = await piAgentRes.json().catch(() => ({}));
48
        throw new Error(errData.error || `HTTP error! status: ${piAgentRes.status}`);
49
      }
50

51
      // Stream the response from pi-agent-service to the frontend
52
      const reader = piAgentRes.body?.getReader();
53
      if (!reader) throw new Error("No response body reader available");
54

55
      const decoder = new TextDecoder('utf-8');
56
      let done = false;
57
      let finalResponse = '';
58

59
      while (!done) {
60
        const { value, done: readerDone } = await reader.read();
61
        if (readerDone) {
62
          done = true;
63
          break;
64
        }
65

66
        const chunk = decoder.decode(value, { stream: true });
67
        const lines = chunk.split('\n');
68

69
        for (let i = 0; i < lines.length; i++) {
70
          const line = lines[i];
71
          if (line.startsWith('event: ')) {
72
            const eventName = line.substring(7).trim();
73
            const dataLine = lines[i + 1];
74
            if (dataLine && dataLine.startsWith('data: ')) {
75
              const dataStr = dataLine.substring(6).trim();
76
              try {
77
                const dataObj = JSON.parse(dataStr);
78

79
                if (eventName === 'status') {
80
                  sendEvent('status', { message: dataObj.message });
81
                } else if (eventName === 'action') {
82
                  sendEvent('action', { description: dataObj.description });
83
                } else if (eventName === 'done') {
84
                  finalResponse = dataObj.response;
85
                  sendEvent('status', { message: `pi-agent 任务执行完成: ${finalResponse.substring(0, 50)}...` });
86
                  sendEvent('done', { message: 'Task completed' });
87
                  res.end();
88
                  return; // End the request cleanly
89
                } else if (eventName === 'error') {
90
                  console.error("pi-agent runtime error:", dataObj);
91
                  sendEvent('error', { message: `pi-agent 运行时异常: ${dataObj.error}` });
92
                  res.end();
93
                  return;
94
                }
95
              } catch (e) {
96
                console.error("Failed to parse SSE data:", dataStr);
97
              }
98
              i++; // skip data line
99
            }
100
          }
101
        }
102
      }
103

104
    } catch (e: any) {
105
      console.error("Failed to connect to pi-agent-service:", e.message);
106
      sendEvent('error', { message: `pi-agent 服务调度失败: ${e.message}` });
107
      res.end();
108
    }
109
  } catch (error: any) {
110
    sendEvent('error', { message: error.message });
111
    res.end();
112
  }
113
})
114

115
export default router