RAG检索基准测试发现54%的标识符类查询答案文档中根本不包含该标识符,导致精确匹配失效,挑战了「关键词搜索适合关键词查询」的常见假设。
我们将构建一个科学计算 Agent,它能够读取自然语言描述的物理或数学问题,编写 Python 代码求解,执行代码并返回格式化的结果。对于需要在浏览器和本地 REPL 之间反复切换的工程师和研究人员来说,这非常实用。我们将基于 Oxlo.ai 运行这个 Agent,因为它采用扁平化的按请求计价模式,长问题描述和多轮推理链不会导致成本增加。
需要从 https://portal.oxlo.ai 获取一个 Oxlo.ai API key。
安装 OpenAI SDK:pip install openai
每个项目我都会先验证 API 客户端是否正常工作。创建一个名为 scientific_agent.py 的文件,并将 OpenAI SDK 指向 Oxlo.ai。由于 Oxlo.ai 对热门模型没有冷启动问题,第一个请求的响应速度和后续请求一样快。
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
response = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Say 'Oxlo.ai client is ready'"},
],
)
print(response.choices[0].message.content)
System prompt 是 Agent 的工作描述,它告诉模型只输出一个 Python 代码块,使用标准库的 math 模块,并打印带单位的结果。我把它放在模块级常量中,这样可以在不触碰请求逻辑的情况下迭代措辞。
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a scientific computing assistant. Your goal is to solve quantitative problems accurately.
Rules:
1. Briefly explain your approach.
2. Write one Python code block inside triple backticks that solves the problem.
3. Use only the Python standard library and the `math` module.
4. Print the final answer with clear units.
5. State assumptions if the problem is under-specified.
Output format:
Approach: [brief text]
```python
import math
# calculations
print(f"Result: {value} units")
"""
response = client.chat.completions.create( model="kimi-k2.6", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": "Calculate the terminal velocity of a 2 mm diameter raindrop in air at 20 C using Stokes law."}, ], )
print(response.choices[0].message.content)
## Step 3: Add a sandboxed code executor
我们需要安全地运行模型生成的 Python 代码,不能完全信任它。我使用 exec 配合受限的 globals 字典,并捕获 stdout,这样 Agent 的 print 输出就变成了可以反馈到对话中的数据。
```python
import contextlib
import io
import re
import traceback
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a scientific computing assistant. Your goal is to solve quantitative problems accurately.
Rules:
1. Briefly explain your approach.
2. Write one Python code block inside triple backticks that solves the problem.
3. Use only the Python standard library and the `math` module.
4. Print the final answer with clear units.
5. State assumptions if the problem is under-specified.
Output format:
Approach: [brief text]
```python
import math
# calculations
print(f"Result: {value} units")
"""
def extract_python_block(text: str) -> str:
match = re.search(r"python\s*(.*?)", text, re.DOTALL)
if match:
return match.group(1).strip()
return text.strip()
def run_generated_code(code: str) -> dict: output_buffer = io.StringIO() result = {"stdout": "", "stderr": "", "success": False} restricted_globals = { "builtins": { "print": print, "range": range, "len": len, "abs": abs, "round": round, "pow": pow, "sum": sum, "min": min, "max": max, }, "math": import("math"), } try: with contextlib.redirect_stdout(output_buffer): exec(code, restricted_globals) result["stdout"] = output_buffer.getvalue() result["success"] = True except Exception: result["stderr"] = traceback.format_exc() return result
response = client.chat.completions.create( model="kimi-k2.6", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": "Calculate the terminal velocity of a 2 mm diameter raindrop in air at 20 C using Stokes law."}, ], )
raw_output = response.choices[0].message.content code = extract_python_block(raw_output) execution = run_generated_code(code)
print("--- GENERATED CODE ---") print(code) print("--- STDOUT ---") print(execution["stdout"]) print("--- STDERR ---") print(execution["stderr"])
## Step 4: Close the loop with multi-turn reasoning
原始的 stdout 不是好的用户体验。我们将执行结果反馈给模型,让它写出一个干净的最终答案。第二轮的花费和第一轮一样是固定的按请求计价,这就是为什么 Oxlo.ai 非常适合 Agent 化工作负载:增加更多上下文不会提高价格。
```python
import contextlib
import io
import re
import traceback
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a scientific computing assistant. Your goal is to solve quantitative problems accurately.
Rules:
1. Briefly explain your approach.
2. Write one Python code block inside triple backticks that solves the problem.
3. Use only the Python standard library and the `math` module.
4. Print the final answer with clear units.
5. State assumptions if the problem is under-specified.
Output format:
Approach: [brief text]
```python
import math
# calculations
print(f"Result: {value} units")
"""
FINAL_FORMAT_PROMPT = ( "Based on the execution output above, write a concise final answer for the user. " "Include the numeric result, units, and a short interpretation. Do not write code." )
def extract_python_block(text: str) -> str:
match = re.search(r"python\s*(.*?)", text, re.DOTALL)
if match:
return match.group(1).strip()
return text.strip()
def run_generated_code(code: str) -> dict: output_buffer = io.StringIO() result = {"stdout": "", "stderr": "", "success": False} restricted_globals = { "builtins": { "print": print, "range": range, "len": len, "abs": abs, "round": round, "pow": pow, "sum": sum, "min": min, "max": max, }, "math": import("math"), } try: with contextlib.redirect_stdout(output_buffer): exec(code, restricted_globals) result["stdout"] = output_buffer.getvalue() result["success"] = True except Exception: result["stderr"] = traceback.format_exc() return result
def solve_problem(user_message: str) -> str: # Turn 1: generate code response1 = client.chat.completions.create( model="kimi-k2.6", messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_message}, ], ) raw_output = response1.choices[0].message.content code = extract_python_block(raw_output)
# Execute
execution = run_generated_code(code)
execution_summary = f"Execution stdout:\n{execution['stdout']}\nExecution stderr:\n{execution['stderr']}"
# Turn 2: synthesize final answer
response2 = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
{"role": "assistant", "content": raw_output},
{"role": "user", "content": execution_summary + "\n\n" + FINAL_FORMAT_PROMPT},
],
)
return response2.choices[0].message.content
if name == "main": query = "Calculate the terminal velocity of a 2 mm diameter raindrop in air at 20 C using Stokes law." print(solve_problem(query))
## Step 5: Add error recovery
如果生成的代码抛出异常,我们将错误堆栈反馈给模型,请求它生成修正后的代码块。这使得 Agent 能够对抗简单的语法或逻辑错误,而不会导致整个流程崩溃。
```python
import contextlib
import io
import re
import traceback
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a scientific computing assistant. Your goal is to solve quantitative problems accurately.
Rules:
1. Briefly explain your approach.
2. Write one Python code block inside triple backticks that solves the problem.
3. Use only the Python standard library and the `math` module.
4. Print the final answer with clear units.
5. State assumptions if the problem is under-specified.
Output format:
Approach: [brief text]
```python
import math
# calculations
print(f"Result: {value} units")
"""
FINAL_FORMAT_PROMPT = ( "Based on the execution output above, write a concise final answer for the user. " "Include the numeric result, units, and a short interpretation. Do not write code." )
ERROR_CORRECTION_PROMPT = ( "The previous code produced an error. Please correct the code and return a complete, runnable Python block. " "Preserve the original approach unless it is fundamentally flawed." )
def extract_python_block(text: str) -> str:
match = re.search(r"python\s*(.*?)", text, re.DOTALL)
if match:
return match.group(1).strip()
return text.strip()
def run_generated_code(code: str) -> dict: output_buffer = io.StringIO() result = {"stdout": "", "stderr": "", "success": False} restricted_globals = { "builtins": { "print": print, "range": range, "len": len, "abs": abs, "round": round, "pow": pow, "sum": sum, "min": min, "max": max, }, "math": import("math"), } try: with contextlib.redirect_stdout(output_buffer): exec(code, restricted_globals) result["stdout"] = output_buffer.getvalue() result["success"] = True except Exception: result["stderr"] = traceback.format_exc() return result
def solve_problem(user_message: str, max_retries: int = 1) -> str: messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_message}, ]
# Turn 1: generate code
response = client.chat.completions.create(
model="kimi-k2.6",
messages=messages,
)
assistant_content = response.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_content})
code = extract_python_block(assistant_content)
execution = run_generated_code(code)
# Retry loop on error
retries = 0
while not execution["success"] and retries < max_retries:
error_msg = f"Error:\n{execution['stderr']}\n\n{ERROR_CORRECTION_PROMPT}"
messages.append({"role": "user", "content": error_msg})
response = client.chat.completions.create(
model="kimi-k2.6",
messages=messages,
)
assistant_content = response.choices[0].message.content
messages.append({"role": "assistant", "content": assistant_content})
code = extract_python_block(assistant_content)
execution = run_generated_code(code)
retries += 1
if not execution["success"]:
return f"Failed after {retries} retries. Last error:\n{execution['stderr']}"
# Turn final: synthesize answer
execution_summary = f"Execution stdout:\n{execution['stdout']}"
messages.append({"role": "user", "content": execution_summary + "\n\n" + FINAL_FORMAT_PROMPT})
response = client.chat.completions.create(
model="kimi-k2.6",
messages=messages,
)
return response.choices[0].message.content
if name == "main": query = "Calculate the terminal velocity of a 2 mm diameter raindrop in air at 20 C using Stokes law." print(solve_problem(query))
将最终脚本保存为 `scientific_agent.py`,导出你的 key,然后从终端运行。
```bash
export OXLO_API_KEY="sk-..."
python scientific_agent.py
示例输出(略有删减):
Approach: Stokes law gives terminal velocity v = (2/9) * (rho_p - rho_f) * g * r^2 / mu. I assume the raindrop is water (rho_p = 1000 kg/m3), air density at 20 C is about 1.204 kg/m3, and air dynamic viscosity is 1.81e-5 Pa.s.
Result: The terminal velocity is approximately 12.1 m/s. This is in the typical range for small raindrops, though Stokes law strictly applies to laminar flow with Reynolds number below about 1. In reality, a 2 mm drop may experience some turbulent drag, so the actual velocity would be slightly lower.
对于需要分步符号推理的问题,可以换用 deepseek-r1-671b 再生成代码;或者在沙盒和 system prompt 中加入 matplotlib,这样 Agent 就能在数值答案之外返回图表。如果你计划将此作为服务运行,可以用受限的 Docker 容器或 subprocess timeout 替换本地 exec 沙盒,以获得更强的隔离。