Spaces:
Runtime error
Runtime error
| #!/usr/bin/env python | |
| """ | |
| @Modified By: mashenquan, 2023/8/22. A definition has been provided for the return value of _think: returning false indicates that further reasoning cannot continue. | |
| """ | |
| import asyncio | |
| from pydantic import BaseModel | |
| from metagpt.actions import CollectLinks, ConductResearch, WebBrowseAndSummarize | |
| from metagpt.actions.research import get_research_system_text | |
| from metagpt.const import RESEARCH_PATH | |
| from metagpt.logs import logger | |
| from metagpt.roles import Role | |
| from metagpt.schema import Message | |
| class Report(BaseModel): | |
| topic: str | |
| links: dict[str, list[str]] = None | |
| summaries: list[tuple[str, str]] = None | |
| content: str = "" | |
| class Researcher(Role): | |
| def __init__( | |
| self, | |
| name: str = "David", | |
| profile: str = "Researcher", | |
| goal: str = "Gather information and conduct research", | |
| constraints: str = "Ensure accuracy and relevance of information", | |
| language: str = "en-us", | |
| **kwargs, | |
| ): | |
| super().__init__(name, profile, goal, constraints, **kwargs) | |
| self._init_actions([CollectLinks(name), WebBrowseAndSummarize(name), ConductResearch(name)]) | |
| self.language = language | |
| if language not in ("en-us", "zh-cn"): | |
| logger.warning(f"The language `{language}` has not been tested, it may not work.") | |
| async def _think(self) -> bool: | |
| if self._rc.todo is None: | |
| self._set_state(0) | |
| return True | |
| if self._rc.state + 1 < len(self._states): | |
| self._set_state(self._rc.state + 1) | |
| else: | |
| self._rc.todo = None | |
| return False | |
| async def _act(self) -> Message: | |
| logger.info(f"{self._setting}: ready to {self._rc.todo}") | |
| todo = self._rc.todo | |
| msg = self._rc.memory.get(k=1)[0] | |
| if isinstance(msg.instruct_content, Report): | |
| instruct_content = msg.instruct_content | |
| topic = instruct_content.topic | |
| else: | |
| topic = msg.content | |
| research_system_text = get_research_system_text(topic, self.language) | |
| if isinstance(todo, CollectLinks): | |
| links = await todo.run(topic, 4, 4) | |
| ret = Message("", Report(topic=topic, links=links), role=self.profile, cause_by=type(todo)) | |
| elif isinstance(todo, WebBrowseAndSummarize): | |
| links = instruct_content.links | |
| todos = (todo.run(*url, query=query, system_text=research_system_text) for (query, url) in links.items()) | |
| summaries = await asyncio.gather(*todos) | |
| summaries = list((url, summary) for i in summaries for (url, summary) in i.items() if summary) | |
| ret = Message("", Report(topic=topic, summaries=summaries), role=self.profile, cause_by=type(todo)) | |
| else: | |
| summaries = instruct_content.summaries | |
| summary_text = "\n---\n".join(f"url: {url}\nsummary: {summary}" for (url, summary) in summaries) | |
| content = await self._rc.todo.run(topic, summary_text, system_text=research_system_text) | |
| ret = Message("", Report(topic=topic, content=content), role=self.profile, cause_by=type(self._rc.todo)) | |
| self._rc.memory.add(ret) | |
| return ret | |
| async def _react(self) -> Message: | |
| while True: | |
| await self._think() | |
| if self._rc.todo is None: | |
| break | |
| msg = await self._act() | |
| report = msg.instruct_content | |
| self.write_report(report.topic, report.content) | |
| return msg | |
| def write_report(self, topic: str, content: str): | |
| if not RESEARCH_PATH.exists(): | |
| RESEARCH_PATH.mkdir(parents=True) | |
| filepath = RESEARCH_PATH / f"{topic}.md" | |
| filepath.write_text(content) | |
| if __name__ == "__main__": | |
| import fire | |
| async def main(topic: str, language="en-us"): | |
| role = Researcher(topic, language=language) | |
| await role.run(topic) | |
| fire.Fire(main) | |