### 调用 AnythingLLM API 的 Python 实现 以下是使用 Python 编写代码以调用 AnythingLLM API 并实现其功能的具体方法。该过程涵盖了基本的文本生成、聊天交互以及流式响应处理。 #### 1. **安装依赖项** 首先，确保已安装必要的库。可以通过 pip 安装 requests 和 aiohttp 库以便于 HTTP 请求和异步操作： ```bash pip install requests aiohttp ``` #### 2. **基础文本生成** 以下是一个简单的 Python 脚本，用于通过 AnythingLLM API 执行文本生成任务： ```python import requests def generate_text(prompt, model="ollama/qwen2.5:14b"): url = "http://<anythingllm-host>:<port>/api/generate" headers = {"Content-Type": "application/json"} data = { "model": model, "prompt": prompt } response = requests.post(url, json=data, headers=headers) if response.status_code == 200: result = response.json() return result.get('text', '') else: raise Exception(f"Error generating text: {response.status_code}, {response.text}") # 示例调用 if __name__ == "__main__": try: output = generate_text("解释一下量子计算的概念") print(output) except Exception as e: print(e) ``` 此函数向 AnythingLLM 发送 POST 请求，并传入指定的模型名称和提示语句[^1]。 #### 3. **多轮对话支持** 对于涉及上下文记忆或多轮对话的情况，可以改用 `/api/chat` 接口： ```python def chat_with_model(messages, model="ollama/qwen2.5:14b"): url = "http://<anythingllm-host>:<port>/api/chat" headers = {"Content-Type": "application/json"} data = { "model": model, "messages": messages } response = requests.post(url, json=data, headers=headers) if response.status_code == 200: result = response.json() return result['choices'][0]['message']['content'] else: raise Exception(f"Chat error: {response.status_code}, {response.text}") # 示例调用 if __name__ == "__main__": conversation_history = [ {"role": "user", "content": "什么是人工智能?"}, {"role": "assistant", "content": ""} ] reply = chat_with_model(conversation_history) print(reply) ``` 这里构建了一个包含用户提问的消息列表，并将其传递给 API[^2]。 #### 4. **流式响应处理** 如果需要逐块接收生成的结果，可以启用流式传输模式： ```python import asyncio from aiohttp import ClientSession async def stream_response(prompt, model="ollama/qwen2.5:14b"): url = f"http://<anythingllm-host>:<port>/api/stream_generate" async with ClientSession() as session: payload = {'model': model, 'prompt': prompt} async with session.post(url, json=payload) as resp: async for line in resp.content: decoded_line = line.decode('utf-8').strip() if decoded_line: yield decoded_line # 示例调用 async def main(): async for chunk in stream_response("创作一首关于星空的诗歌"): print(chunk, end="", flush=True) if __name__ == "__main__": asyncio.run(main()) ``` 上述代码展示了如何设置异步迭代器来连续读取服务器推送的数据片段[^3]。 --- ### 总结 这些示例覆盖了从简单的一次性文本生成到复杂的多回合对话管理再到高效的流媒体输出等多种常见需求场景下的解决方案。开发者可根据具体业务逻辑灵活选用合适的方式与 AnythingLLM 进行交互[^4]。