1
2
3
4

{
  "message": "写一个 add(a,b) 并用 execute_code 验证",
  "use_agent": true
}

1
2
3
4
5
6

{
  "conversation_id": "00000000-0000-0000-0000-000000000000",
  "message": "解释 FastAPI 依赖注入的作用，并给一个最小示例",
  "use_rag": true,
  "collection_name": "fastapi_docs"
}

1
2
3

curl -N http://localhost:8000/chat?stream=true \
  -H 'content-type: application/json' \
  -d '{"message":"写一个 add(a,b) 并用 execute_code 验证","use_agent":true}'

1
2
3
4

{
  "message": "你必须调用 execute_code 执行：print(1+1)",
  "tools": ["execute_code"]
}

1
2
3

curl http://localhost:8000/agent \
  -H 'content-type: application/json' \
  -d '{"message":"你必须调用 execute_code 执行：print(1+1)","tools":["execute_code"]}'

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109

@dataclass(frozen=True)
class Tool:
    name: str
    description: str
    parameters: JSONSchema
    handler: ToolHandler
    return_schema: JSONSchema | None = None

    def __post_init__(self) -> None:
        if not self.name.strip():
            raise ValueError("tool.name is required")
        if not self.description.strip():
            raise ValueError("tool.description is required")
        _validate_schema_definition(schema=self.parameters, path=f"tool:{self.name}.parameters")
        if self.return_schema is not None:
            _validate_schema_definition(schema=self.return_schema, path=f"tool:{self.name}.return_schema")

    def to_openai_tool(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

    def validate_input(self, payload: Mapping[str, Any]) -> None:
        if not isinstance(payload, dict):
            raise AppError(
                code="tool_input_invalid",
                message="Tool input must be an object.",
                status_code=400,
                details={"expected_type": "object", "actual_type": type(payload).__name__},
            )
        _validate_instance_against_schema(value=payload, schema=self.parameters, path="$")

    async def call(self, payload: Mapping[str, Any]) -> Any:
        self.validate_input(payload)
        if not isinstance(payload, dict):
            raise AppError(
                code="tool_input_invalid",
                message="Tool input must be an object.",
                status_code=400,
                details={"expected_type": "object", "actual_type": type(payload).__name__},
            )

        if asyncio.iscoroutinefunction(self.handler):
            return await self.handler(**payload)
        result = self.handler(**payload)
        if asyncio.iscoroutine(result):
            return await result
        return result


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}
        self._logger = structlog.get_logger()

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise AppError(
                code="tool_already_registered",
                message="Tool is already registered.",
                status_code=409,
                details={"name": tool.name},
            )
        self._tools[tool.name] = tool
        self._logger.info("tool_registered", name=tool.name)

    def get(self, name: str) -> Tool:
        tool = self._tools.get(name)
        if tool is None:
            raise AppError(
                code="tool_not_found",
                message="Tool not found.",
                status_code=404,
                details={"name": name},
            )
        return tool

    def list(self) -> list[Tool]:
        return list(self._tools.values())

    def to_openai_tools(self) -> list[dict[str, Any]]:
        return [t.to_openai_tool() for t in self.list()]

    async def call(self, *, name: str, payload: Mapping[str, Any]) -> Any:
        tool = self.get(name)
        start = time.perf_counter()
        try:
            result = await tool.call(payload)
            self._logger.info(
                "tool_call",
                name=name,
                success=True,
                latency_ms=int((time.perf_counter() - start) * 1000),
            )
            return result
        except Exception as exc:
            self._logger.exception(
                "tool_call",
                name=name,
                success=False,
                latency_ms=int((time.perf_counter() - start) * 1000),
                error=str(exc),
            )
            raise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

system_prompt = _build_system_prompt(tools=self._tools)
messages: list[dict[str, Any]] = [{"role": "system", "content": system_prompt}]
if history_messages:
    messages.extend(list(history_messages))
messages.append({"role": "user", "content": user_input})

steps: list[ReActStep] = []
recorder = trace or TraceRecorder()

for i in range(self._max_iterations):
    started_at_ms = recorder.start_step(step_index=i)
    resp = await self._llm.chat(messages=list(messages), temperature=0.0, stream=False)
    content = str(resp.choices[0].message.content or "")
    messages.append({"role": "assistant", "content": content})

    parsed = _parse_react_output(content)

    if parsed["type"] == "final":
        final_answer = parsed["final"]
        steps.append(
            ReActStep(
                thought=parsed.get("thought", ""),
                action_raw=parsed.get("action_raw", ""),
                tool_name=None,
                tool_args=None,
                observation=None,
            )
        )
        recorder.finish_step(
            step_index=i,
            started_at_ms=started_at_ms,
            thought=parsed.get("thought", ""),
            action_raw=parsed.get("action_raw", ""),
            tool_name=None,
            tool_args=None,
            observation=None,
            error=None,
        )
        return final_answer, steps

    tool_name = parsed["tool_name"]
    tool_args = parsed["tool_args"]
    observation, tool_error = await _call_tool_with_retry(
        tools=self._tools,
        tool_name=tool_name,
        tool_args=tool_args,
        max_retries=TOOL_MAX_RETRIES,
        logger=self._logger,
    )
    recorder.finish_step(
        step_index=i,
        started_at_ms=started_at_ms,
        thought=parsed.get("thought", ""),
        action_raw=parsed.get("action_raw", ""),
        tool_name=tool_name,
        tool_args=tool_args,
        observation=observation,
        error=tool_error,
    )
    steps.append(
        ReActStep(
            thought=parsed.get("thought", ""),
            action_raw=parsed.get("action_raw", ""),
            tool_name=tool_name,
            tool_args=tool_args,
            observation=observation,
        )
    )

    if tool_error is not None:
        observation_text = _format_tool_error_observation(
            tool_name=tool_name,
            tool_args=tool_args,
            error=tool_error,
        )
    else:
        observation_text = _format_tool_observation(
            tool_name=tool_name,
            tool_args=tool_args,
            observation=observation,
        )
    messages.append({"role": "user", "content": observation_text})

raise AppError(
    code="agent_max_iterations",
    message="Agent reached max iterations without producing a final answer.",
    status_code=408,
    details={"max_iterations": self._max_iterations},
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

return "\n".join(
    [
        "你是 DevAssist 的 ReAct Agent，擅长把复杂问题拆成可执行步骤，并调用工具获取证据。",
        "",
        "可用工具：",
        tool_block,
        "",
        "输出格式要求：",
        "- 你必须输出两段：Thought 与 Action",
        "- Action 只有两种：",
        "  1) tool call:",
        "     Action: tool:<tool_name>",
        "     args: <json object>",
        "  2) final answer:",
        "     Action: final: <final answer text>",
    ]
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

async def _handler(*, code: str, timeout_s: int = 5) -> dict[str, Any]:
    if timeout_s <= 0:
        raise AppError(
            code="tool_input_invalid",
            message="timeout_s must be a positive integer.",
            status_code=400,
            details={"timeout_s": timeout_s},
        )

    settings = get_settings()
    allowed_paths = (
        [p.strip() for p in settings.sandbox_allowed_paths.split(",") if p.strip()]
        if settings.sandbox_allowed_paths
        else []
    )
    safety = check_code_safety(code=code, allowed_paths=allowed_paths)
    logger = structlog.get_logger()
    if safety.issues:
        report = format_safety_report(safety)
        if safety.is_blocked:
            logger.warning("sandbox_code_blocked", report=report)
            raise AppError(
                code="sandbox_code_blocked",
                message="Code contains dangerous operations and cannot be executed.",
                status_code=400,
                details={"report": report},
            )
        logger.warning("sandbox_code_warning", report=report)

    return await execute_python(code=code, timeout_s=timeout_s)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

container = await asyncio.to_thread(
    client.containers.create,
    image=image,
    command=["python", "main.py"],
    name=f"devassist-sandbox-{uuid4().hex[:12]}",
    detach=True,
    network_disabled=True,
    mem_limit=memory_limit,
    pids_limit=DEFAULT_PIDS_LIMIT,
    read_only=True,
    tmpfs={"/tmp": f"rw,size={DEFAULT_TMPFS_SIZE}"},
    volumes={tmpdir: {"bind": "/work", "mode": "ro"}},
    working_dir="/work",
    environment={"PYTHONDONTWRITEBYTECODE": "1"},
)
await asyncio.to_thread(container.start)
wait_result = await asyncio.wait_for(asyncio.to_thread(container.wait), timeout=timeout_s)
exit_code = int((wait_result or {}).get("StatusCode", 0))

stdout_b = await asyncio.to_thread(container.logs, stdout=True, stderr=False)
stderr_b = await asyncio.to_thread(container.logs, stdout=False, stderr=True)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

async def build_history(self, *, conversation_id: UUID, query: str) -> list[dict[str, str]]:
    history = await self._short.get_history(conversation_id=conversation_id)
    if self._long is None:
        return history

    try:
        memories = await self._long.search(conversation_id=conversation_id, query=query, top_k=3)
    except Exception as exc:
        self._logger.error("long_term_memory_search_failed", conversation_id=str(conversation_id), error=str(exc))
        return history

    if not memories:
        return history
    text = "\n".join([f"- {m.strip()}" for m in memories if m and m.strip()])
    if not text:
        return history
    return [{"role": "system", "content": f"Long-term memory:\n{text}"}] + history


async def add_turn(self, *, conversation_id: UUID, user: str, assistant: str, llm: LLMClient) -> None:
    evicted = await self._short.add_turn(conversation_id=conversation_id, user=user, assistant=assistant)
    if not evicted:
        return
    if self._long is None:
        return

    buf = self._archive_buffers.get(conversation_id)
    if buf is None:
        buf = []
        self._archive_buffers[conversation_id] = buf
    buf.extend(evicted)
    if len(buf) < self._summarize_min_messages:
        return

    summary = await self._summarize_messages(llm=llm, messages=buf)
    buf.clear()
    if not summary.strip():
        return
    try:
        await self._long.add_summary(conversation_id=conversation_id, summary=summary)
    except Exception as exc:
        self._logger.error("long_term_memory_add_failed", conversation_id=str(conversation_id), error=str(exc))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

@dataclass(frozen=True)
class TraceStep:
    step_index: int
    thought: str
    action_raw: str
    tool_name: str | None
    tool_args: dict[str, Any] | None
    observation: Any | None
    error: str | None
    started_at_ms: int
    finished_at_ms: int

    @property
    def latency_ms(self) -> int:
        return max(0, self.finished_at_ms - self.started_at_ms)

    def to_dict(self) -> dict[str, Any]:
        return {
            "step_index": self.step_index,
            "thought": self.thought,
            "action_raw": self.action_raw,
            "tool_name": self.tool_name,
            "tool_args": self.tool_args,
            "observation": self.observation,
            "error": self.error,
            "started_at_ms": self.started_at_ms,
            "finished_at_ms": self.finished_at_ms,
            "latency_ms": self.latency_ms,
        }


class TraceRecorder:
    def start_step(self, *, step_index: int) -> int:
        _ = step_index
        return int(time.time() * 1000)

    def finish_step(
        self,
        *,
        step_index: int,
        started_at_ms: int,
        thought: str,
        action_raw: str,
        tool_name: str | None,
        tool_args: dict[str, Any] | None,
        observation: Any | None,
        error: str | None,
    ) -> TraceStep:
        finished_at_ms = int(time.time() * 1000)
        step = TraceStep(
            step_index=step_index,
            thought=thought,
            action_raw=action_raw,
            tool_name=tool_name,
            tool_args=tool_args,
            observation=observation,
            error=error,
            started_at_ms=started_at_ms,
            finished_at_ms=finished_at_ms,
        )
        self._steps.append(step)

        self._logger.info(
            "agent_trace_step",
            run_id=self._run_id,
            step_index=step_index,
            tool_name=tool_name,
            latency_ms=step.latency_ms,
            success=error is None,
        )
        return step

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

step_queue: asyncio.Queue[dict[str, object]] = asyncio.Queue()

class _QueueTraceRecorder(TraceRecorder):
    def finish_step(self, **kwargs):  # type: ignore[override]
        step = super().finish_step(**kwargs)
        try:
            step_queue.put_nowait(step.to_dict())
        except Exception:
            pass
        return step

trace = _QueueTraceRecorder(run_id=str(conversation_id))

while True:
    if agent_future.done() and step_queue.empty():
        break

    step_get = asyncio.create_task(step_queue.get())
    done, pending = await asyncio.wait({agent_future, step_get}, return_when=asyncio.FIRST_COMPLETED)

    if step_get in done:
        step = step_get.result()
        yield sse_event(data={"type": "step", **step})
        continue

    step_get.cancel()

1
2
3

Thought: 需要执行代码拿到可复现输出。
Action: tool:execute_code
args: {"code":"print(1+1)", "timeout_s": 3}

1
2

Thought: 已验证通过，可以总结。
Action: final: 这里是最终答案……

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

{
  "type": "step",
  "step_index": 0,
  "thought": "...",
  "action_raw": "...",
  "tool_name": "execute_code",
  "tool_args": {"code": "...", "timeout_s": 5},
  "observation": {"stdout": "...", "stderr": "...", "exit_code": 0, "duration_ms": 12},
  "error": null,
  "started_at_ms": 0,
  "finished_at_ms": 0,
  "latency_ms": 0
}

1

写一个 Python 函数实现二分查找，并用 execute_code 跑 5 个测试用例验证。最终回答里要包含 stdout/stderr/exit_code 证据。

1

先用 search_docs 检索 fastapi_docs 里关于依赖注入的说明，再用 5 条要点总结，并写出每条要点来自哪个 source/chunk_index。

1

docker compose up -d --build --force-recreate frontend

1

pnpm approve-builds --all

1

docker compose build backend

1

SANDBOX_IMAGE=devassist-backend